Source code for intents.language.intent_language

"""
**Intent** language files have the following structure:

.. code-block:: yaml

    examples:
      - an example utterance
      - another example utterance
      - an example utterance with $foo{42} as a numeric parameter

    slot_filling_prompts:
      foo:
        - Tell me the value for "foo"

    responses:
      default:
        - text:
          - A plain text response
          - An alternative response
          - Another alternative, referencing $foo as a paramneter
      rich:
        - text:
          - A text response for rich clients
        - quick_replies:
          - a reply chip
          - another reply chip

Let's look at the sections of this file.

* **examples** contain example utterances that will be used to predict the given
  intent. If your intent has a parameter, it can be referenced as
  $parameter_name{example value}. You can omit this section if your intent is
  not meant to be predicted (some intents are trigger-only)
* **slot_filling_prompt** are used when your intent has a mandatory parameter,
  and this parameter could not be matched in the user message. These prompts
  will be used to ask the User about that parameter. You can omit this section
  if your intent has no mandatory parameters, or if you don't want to define
  custom prompts.
* **responses** contain messages that Agent will send to User in response to the
  Intent. Two response groups are available:

  * **default** can only contain plain-text messages. It is good practice to
    always provide text-only response for situations where rich ones can't be rendered,
    such as vocal assistants, smartphone notifications and such. The `text`
    response type is specified in :class:`TextIntentResponse`
  * **rich** responses allow some extra types:
    :class:`QuickRepliesIntentResponse`, :class:`ImageIntentResponse`,
    :class:`CardIntentResponse` and :class:`CustomPayloadIntentResponse`

"""
import os
import re
import random
import logging
import warnings
from enum import Enum
from dataclasses import dataclass, field
from typing import List, Dict, Union, Tuple, Type

import yaml

# pylint: disable=unused-import
import intents # Needed to generate docs
from intents import Intent, EntityMixin
from intents.language import agent_language, LanguageCode

logger = logging.getLogger(__name__)

#
# Example Utterances
#

[docs]class UtteranceChunk: """ An Example Utterance can be seen as a sequence of Chunks, where each Chunk is either a mapped Entity, or a plain text string. """
[docs]@dataclass class TextUtteranceChunk(UtteranceChunk): """ An Utterance Chunk that is a static, plain text string. """ text: str
[docs]@dataclass class EntityUtteranceChunk(UtteranceChunk): """ An Utterance Chunk that is a matched entity """ entity_cls: Type[EntityMixin] parameter_name: str parameter_value: str
# TODO: check that parameter_value is one of the entries in custom entities RE_EXAMPLE_PARAMETERS = re.compile(r"\$(?P<parameter_name>[\w]+)\{(?P<parameter_value>[^\}]+)\}")
[docs]class ExampleUtterance(str): """ One of the example Utterances of a given Intent. """ # TODO: check for escape characters - intent is possibly intent_cls def __init__(self, example: str, intent: Intent): self._intent = intent self.chunks() # Will check parameters def __new__(cls, example: str, intent: Intent): return super().__new__(cls, example)
[docs] def chunks(self): """ Return the Utterance as a sequence of :class:`UtteranceChunk`. Each chunk is either a plain text string, or a mapped Entity. >>> utterance = ExampleUtterance("My name is $user_name{Guido}!", intents.user_gives_name) >>> utterance.chunks() [ TextUtteranceChunk(text="My name is "), EntityUtteranceChunk(entity_cls=Sys.Person, parameter_name="user_name", parameter_value="Guido"), TextUtteranceChunk(text="!") ] .. warning:: This method doesn't handle escaping yet. """ # TODO: handle escaping parameter_schema = self._intent.parameter_schema result = [] last_end = 0 for m in RE_EXAMPLE_PARAMETERS.finditer(self): m_start, m_end = m.span() m_groups = m.groupdict() if m_start > 0: result.append(TextUtteranceChunk(text=self[last_end:m_start])) if (parameter_name := m_groups['parameter_name']) not in parameter_schema: raise ValueError(f"Example '{self}' references parameter ${parameter_name}, but intent {self._intent.name} does not define such parameter.") entity_cls = parameter_schema[parameter_name].entity_cls result.append(EntityUtteranceChunk( entity_cls=entity_cls, parameter_name=m_groups['parameter_name'], parameter_value=m_groups['parameter_value'] )) last_end = m_end last_chunk = TextUtteranceChunk(text=self[last_end:]) if last_chunk.text: result.append(last_chunk) return result
# # Responses #
[docs]class IntentResponseGroup(Enum): """ Intent responses are divided in groups. The same intent can be answered with a set of plain-text responses (:const:`IntentResponseGroup.DEFAULT`), or with rich content (:const:`IntentResponseGroup.RICH`) that includes cards, images and quick replies. """ DEFAULT = "default" RICH = "rich"
[docs]class IntentResponse: """ One of the Response Utterances of a given Intent. """
[docs] @classmethod def from_yaml(cls, data: dict): """ Instantiate an IntentResponse from language data, as it's found in its YAML file. Typically, IntentResponse is a dataclass and `data` is a dict of fields; however specific subclasses may override with custom parameters. Args: data: A response object, as it is read from the YAML file """ return cls(**data)
[docs] def render(self, intent: Intent): """ Render the Response object by replacing parameter references with values from the give :class:`Intent` instance. This is done on every member of the Response class. As `IntentResponse` objects are frozen, `self` won't change. Instead a **new object** will be returned, with rendered members. .. warning:: This function currently doesn't handle escaping. .. code-block:: python >>> intent = OrderCoffee(roast="dark") # This typically comes from a Prediction >>> response = TextIntentResponse(choices=["I like $roast roasted coffee as well", "$roast roast, good choice!"] >>> response.render(intent) TextIntentResponse(choices=["I like dark roasted coffee as well", "dark roast, good choice!"]) Args: intent: The Intent instance that will be used to read parameter values """ # TODO: handle list/dict values differently parameter_dict = {k: intent.__dict__[k] for k in intent.parameter_schema} result_args = {} try: dataclass_fields = getattr(self, "__dataclass_fields__") except AttributeError as exc: raise ValueError(f"Response '{self}' doesn't seem to be a dataclass. If this is a " "custom IntentResponse class, make sure to add a @dataclass decorator; " "otherwise, please file an issue on the Intents repository.") from exc for field_name in dataclass_fields: if isinstance(field_name, (str, list)): result_args[field_name] = _render_response( self.__dict__[field_name], parameter_dict ) else: result_args[field_name] = self.__dict__[field_name] return self.__class__(**result_args)
def _render_response(data: Union[str, list], parameter_dict: Dict[str, str]): """ Render some response data by replacing parameter references with the one contained in `parameter_dict`. This function is called by :meth:`IntentResponse.render` on each of the Response members. """ if not data: return data if isinstance(data, str): result = data for k, v in parameter_dict.items(): # result = result.replace(f"${k}", str(v)) # TODO: handle escaping result = re.sub(r"(?:(?<=[^\w])|^)\$%s(?:(?=[^\w])|$)" % k, str(v), result) return result if isinstance(data, list): return [_render_response(x, parameter_dict) for x in data] raise NotImplementedError(f"Unsupported IntentResponse member type: `{type(data)}`. " "Please open an issue at the Intents repository")
[docs]@dataclass(frozen=True) class TextIntentResponse(IntentResponse): """ A plain text response. The actual response is picked randomly from a pool of choices. In the YAML definition a text response can either be a string, as in .. code-block:: yaml responses: default: - text: This is a response Or a list of choices (the output fulfillment message will be chosen randomly among the different options) .. code-block:: yaml responses: default: - text: - This is a response - This is an alternative response Args: choices: A list of equivalent responses. One will be chosen at random at prediction time """ choices: List[str]
[docs] @classmethod def from_yaml(cls, data: Union[str, List[str]]): if isinstance(data, str): return cls([data]) assert isinstance(data, list) return cls(data)
[docs] def random(self): """ Pick one of the available choices. It is recommended to :meth:`~IntentResponse.render` the response first. .. code-block:: python >>> intent = OrderCoffee(roast="dark") # This typically comes from a Prediction >>> response = TextIntentResponse(choices=["I like $roast roasted coffee as well", "$roast roast, good choice!"] >>> response.render(intent).random() "dark roast, good choice!" """ return random.choice(self.choices)
[docs]@dataclass(frozen=True) class QuickRepliesIntentResponse(IntentResponse): """ A set of Quick Replies that can be used to answer the Intent. Each reply must be shorter than 20 characters. In the YAML definition a quick replies response can either be a string, as in .. code-block:: yaml rich: - quick_replies: Order Pizza Or a list of replies, that will be rendered as separate chips .. code-block:: yaml rich: - quick_replies: - Order Pizza - Order Beer Args: replies: A list of possible User replies to the current intent """ replies: List[str] def __post_init__(self): # TODO: find policy for rendered replies for rep in self.replies: if len(rep) > 20: raise ValueError(f"Quick Replies must be shorter than 20 chars. Quick reply '{rep}' is {len(rep)} chars long.")
[docs] @classmethod def from_yaml(cls, data: Union[str, List[str]]): if isinstance(data, str): return cls([data]) assert isinstance(data, list) return cls(data)
[docs]@dataclass(frozen=True) class ImageIntentResponse(IntentResponse): """ A simple image, defined by its URL and an optional title In the YAML definition an image response can either be a string with the image URL, as in .. code-block:: yaml rich: - image: https://example.com/image.png Or an object with the image URL and a title, as in .. code-block:: yaml rich: - image: url: https://example.com/image.png title: An example image Args: url: A publicly accessible image URL title: A title for the image """ url: str title: str = None
[docs] @classmethod def from_yaml(cls, data: Union[str, List[str]]): if isinstance(data, str): return cls(url=data) assert isinstance(data, dict) return cls(**data)
[docs]@dataclass(frozen=True) class CardIntentResponse(IntentResponse): """ A simple content card that can be rendered on many platforms. In the YAML, this is defined as Or an object with the image URL and a title, as in .. code-block:: yaml rich: - card: title: The card title subtitle: An optional subtitle image: https://example.com/image.jpeg link: https://example.com/ Args: title: The card title subtitle: A short subtitle image: URL of an image to be used as cover link: A link to follow if User taps the card """ title: str subtitle: str = None image: str = None link: str = None
[docs]@dataclass(frozen=True) class CustomPayloadIntentResponse(IntentResponse): """ Custom Payloads are objects with arbitrary fields, they are supported by Dialogflow in every response group, including "Default". Currently they can only be defined in the YAML as free form payloads; support for marshalling or generation from code is expected in future developments. Args: name: A name that identifies the payload type payload: Any JSON-serializable object """ name: str payload: dict
[docs] @classmethod def from_yaml(cls, data: Dict[str, dict]): """ In the YAML definition a custom payload is defined as follows .. code-block:: yaml rich: - custom: custom_location: latitude: 45.484907 longitude: 9.203299 name: Piazza Duca D'Aosta, Milano NOTE: while not currently enforced, consistency is expected between payload names and their fields. Future versions of the library will marshal custom payloads against dataclass schemas. """ if not isinstance(data, dict): raise ValueError(f"A custom payload is expected to be a dict in the form 'payload_name: {{\"foo\": \"bar\"}}. Found: {data}") if len(data) != 1: raise ValueError(f"A custom payload is expected to contain a single key representing the payload name, mapping to its value (e.g. 'location: {{\"latitude\": 42, ...}}'). Found {len(data)} keys: {data.keys()}") payload_name = list(data.keys())[0] payload_content = list(data.values())[0] if not isinstance(payload_content, dict): raise ValueError(f"Custom payloads are expected to be dictionaries. {payload_name} has value: {payload_content}") return CustomPayloadIntentResponse(payload_name, payload_content)
[docs]class IntentResponseDict(dict): """ This is dict of Intent responses, divided by group (`Dict[IntentResponseGroup, List[IntentResponse]]`): .. code-block:: python IntentResponseDict({ IntentResponseGroup.DEFAULT: [ TextIntentResponse(choices=["An anternative", "Another alternative]) ], IntentResponseGroup.RICH: [ TextIntentResponse(choices=["Text response for Rich group"]), QuickRepliesIntentResponse(replies=["A reply", "Another reply"]) ] }) In addition to a standard dict, a :meth:`~IntentResponseDict.for_group` convenience method is provided, to select the most suitable messages for a given group. """ def __call__(self, response_group: IntentResponseGroup=IntentResponseGroup.RICH): warnings.warn("prediction.fulfillment_messages(group) is deprecated, and will be " "removed in 0.4. Please update your code to use " "prediction.fulfillment_messages.for_group() instead", DeprecationWarning) return self.for_group(response_group)
[docs] def for_group( self, response_group: IntentResponseGroup=IntentResponseGroup.RICH ) -> List[IntentResponse]: """ Return a list of fulfillment messages that are suitable for the given Response Group. The following scenarios may happen: * :class:`language.IntentResponseGroup.DEFAULT` is requested -> Message in the `DEFAULT` group will be returned * :class:`language.IntentResponseGroup.RICH` is requested * `RICH` messages are defined -> `RICH` messages are returned * No `RICH` message is defined -> `DEFAULT` messages are returned If present, messages in the "rich" group will be returned: >>> prediction.fulfillment_messages() [TextIntentResponse(choices=['I like travelling too! How can I help?']), QuickRepliesIntentResponse(replies=['Recommend a hotel', 'Send holiday photo', 'Where the station?'])] Alternatively, I can ask for plain-text default messages: >>> from intents.language import IntentResponseGroup >>> prediction.fulfillment_messages(IntentResponseGroup.DEFAULT) [TextIntentResponse(choices=['Nice, I can send you holiday pictures, or recommend an hotel'])] Args: response_group: The Response Group to fetch responses for """ if response_group == IntentResponseGroup.RICH and \ not self.get(response_group): response_group = IntentResponseGroup.DEFAULT return self.get(response_group, [])
# # Language Data Loader #
[docs]@dataclass class IntentLanguageData: """ Language data for an Intent consists of three resources: * Example Utterances * Slot Filling Prompts * Responses **Example Utterances** are the messages that Agent will be trained on to recognize the Intent. **Responses**, intuitively, are the Agent's response messages that will be sent to User once the Intent is recognized. They are divided in groups: a :const:`IntentResponseGroup.DEFAULT` group (mandatory) can only contain plain text responses. A :const:`IntentResponseGroup.RICH` group can provide intent responses that include cards, images and quick replies. **Slot Filling Promps** are used to solve parameters that couldn't be tagged in the original message. For instance a `OrderPizza` intent may have a `pizza_type` parameter. When User asks "I'd like a pizza" we want to fill the slot by asking "What type of pizza?". `slot_filling_prompts` will map parameters to their prompts: `{"pizza_type": ["What type of pizza?"]}` Args: example_utterance: A list of User utterances that should be associated with the intent slot_filling_prompts: A map of prompts to use when required parameters are missing responses: The Responses that Agent will send to User when the intent is predicted """ example_utterances: List[ExampleUtterance] = field(default_factory=list) slot_filling_prompts: Dict[str, List[str]] = field(default_factory=dict) responses: Dict[IntentResponseGroup, List[IntentResponse]] = field(default_factory=list)
[docs]def intent_language_data( agent_cls: "intents.model.agent.AgentType", intent_cls: Type[Intent], language_code: LanguageCode=None ) -> Dict[LanguageCode, IntentLanguageData]: """ Return Language Data for the given Intent. Args: agent_cls: The Agent class that registered the Intent intent_cls: The Intent to load language data for language_code: A specific language code to load. If not present, all available languages will be returned """ if "__intent_language_data__" in intent_cls.__dict__: result = intent_cls.__intent_language_data__ if language_code and language_code not in result: raise KeyError(f"Intent '{intent_cls}' in Agent '{agent_cls}' doesn't seem to define " f"language data for language '{language_code}") return result try: language_folder = agent_language.agent_language_folder(agent_cls) if not language_code: result = {} for language_code in agent_cls.languages: language_data = intent_language_data(agent_cls, intent_cls, language_code) result[language_code] = language_data[language_code] return result if isinstance(language_code, str): language_code = LanguageCode(language_code) language_file = os.path.join(language_folder, language_code.value, f"{intent_cls.name}.yaml") if not os.path.isfile(language_file): raise ValueError(f"Language file not found for intent '{intent_cls.name}'. Expected path: {language_file}. Language files are required even if the intent doesn't need language; in this case, use an empty file.") with open(language_file, 'r') as f: language_data = yaml.load(f.read(), Loader=yaml.FullLoader) if not language_data: return IntentLanguageData([], {}, []) examples_data = language_data.get('examples', []) responses_data = language_data.get('responses', []) examples = [ExampleUtterance(s, intent_cls) for s in examples_data] responses = _build_responses(responses_data) language_data = IntentLanguageData( example_utterances=examples, slot_filling_prompts=language_data.get('slot_filling_prompts', {}), responses=responses ) return {language_code: language_data} except Exception as e: raise RuntimeError(f"Failed to load language data for intent {intent_cls.name} (see stacktrace above for root cause).") from e
[docs]def render_responses(intent: Intent, language_data: IntentLanguageData) -> Tuple[IntentResponseDict, str]: """ Return a copy of responses in `language_data` where intent parameter references are replaced with their values from the given :class:`Intent` instance. Args: intent: The intent to read parameters from language_data: A collection of responses for the given intent Return: Intent responses, and plaintext version """ result_messages: IntentResponseDict = IntentResponseDict() for group, response_list in language_data.responses.items(): result_messages[group] = [r.render(intent) for r in response_list] rendered_plaintext = [r.random() for r in result_messages.get(IntentResponseGroup.DEFAULT, [])] result_plaintext = " ".join(rendered_plaintext) return result_messages, result_plaintext
def _build_responses(responses_data: dict): result = {} response_group: str responses: List[dict] for response_group, responses in responses_data.items(): try: response_group = IntentResponseGroup(response_group) except ValueError as exc: raise NotImplementedError(f"Unsupported Response Group '{response_group}' in " + "'responses'. Currently, only 'default' and 'rich' are supported") from exc result[response_group] = [] for r in responses: assert len(r) == 1 for r_type, r_data in r.items(): if response_group == IntentResponseGroup.DEFAULT and r_type != 'text': raise ValueError(f"Message type {r_type} found in response group 'default'. " + "Only 'text' type is allowed in 'default': please define the additional " + "'rich' response group to use rich responses.") if r_type == 'text': result[response_group].append(TextIntentResponse.from_yaml(r_data)) elif r_type == 'quick_replies': result[response_group].append(QuickRepliesIntentResponse.from_yaml(r_data)) elif r_type == 'image': result[response_group].append(ImageIntentResponse.from_yaml(r_data)) elif r_type == 'card': result[response_group].append(CardIntentResponse.from_yaml(r_data)) elif r_type == 'custom': result[response_group].append(CustomPayloadIntentResponse.from_yaml(r_data)) else: raise NotImplementedError(f"Unsupported response type '{r_type}'. Currently, " + "only 'text' is supported") return result