API Reference - Main Package#

async def draw_intent_graph_from_config(path: str) -> IntentRouter:
    """
    Load the intent router from the configuration file.
    """
    log.debug("Loading YAML configuration file", config_file_path=path)

    with open(path, "r", encoding="utf-8") as file:
        config = yaml.safe_load(file)
    log.debug("Loading bot interface", bot_interface_config=json.dumps(config, indent=4))

    plugins = config.pop("plugins")
    log.debug("Collected_plugins", plugins=plugins)
    for plugin in plugins:
        import_plugin(plugin)

    # Remove YAML extension from path
    path = path.rsplit(".", 1)[0]
    intent_router = IntentRouter(config.pop("conversation", {}))
    return await to_image(intent_router, path + ".png")

def main():
    """
    Entry point for the Intentional CLI.
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("path", help="the path to the configuration file to load.", type=str)
    parser.add_argument("--draw", help="Draw the graph.", action="store_true")
    parser.add_argument(
        "--log-cli-level",
        help="Select the logging level to the console.",
        type=str,
        default="INFO",
    )
    parser.add_argument("--log-file", help="Path to the log file.", type=str)
    parser.add_argument(
        "--log-file-level",
        help="Select the logging level to the file. Ignore if no path is specified with --log-file",
        type=str,
        default="DEBUG",
    )
    args = parser.parse_args()

    # Set the CLI log level
    cli_level = logging.getLevelName(args.log_cli_level.upper())
    structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(cli_level))

    if args.log_file:
        # https://www.structlog.org/en/stable/standard-library.html
        file_level = logging.getLevelName(args.log_file_level.upper())
        timestamper = structlog.processors.TimeStamper(fmt="iso")
        pre_chain = [
            structlog.stdlib.add_log_level,
            structlog.stdlib.ExtraAdder(),
            timestamper,
        ]
        logging.config.dictConfig(
            {
                "version": 1,
                "disable_existing_loggers": False,
                "handlers": {
                    "default": {
                        "level": cli_level,
                        "class": "logging.StreamHandler",
                        "formatter": "colored",
                    },
                    "file": {
                        "level": file_level,
                        "class": "logging.handlers.WatchedFileHandler",
                        "filename": args.log_file,
                        "formatter": "json",
                    },
                },
                "formatters": {
                    "json": {
                        "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
                        "fmt": "%(asctime)s %(levelname)s %(message)s",
                    },
                    "colored": {
                        "()": structlog.stdlib.ProcessorFormatter,
                        "processors": [
                            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                            structlog.dev.ConsoleRenderer(colors=True),
                        ],
                        "foreign_pre_chain": pre_chain,
                    },
                },
                "loggers": {
                    "": {
                        "handlers": ["default", "file"],
                        "level": "DEBUG",
                        "propagate": True,
                    },
                },
            }
        )
        structlog.configure(
            processors=[
                structlog.stdlib.add_log_level,
                structlog.stdlib.PositionalArgumentsFormatter(),
                timestamper,
                structlog.processors.StackInfoRenderer(),
                structlog.processors.format_exc_info,
                structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
            ],
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
            cache_logger_on_first_use=True,
        )

    if args.draw:
        asyncio.run(draw_intent_graph_from_config(args.path))
        return

    bot = load_configuration_file(args.path)
    asyncio.run(bot.run())

async def to_bytes(intent_router: IntentRouter, mermaid_domain: str = "https://mermaid.ink/img/") -> bytes:
    """
    Uses mermaid.ink to render the intent's graph into an image.

    Args:
        intent_router: the intents graph to draw.
        mermaid_domain: the domain of your Mermaid instance, if you have your own.
            Defaults to the public mermaid.ink domain.

    Returns:
        The bytes of the resulting image. To save them into an image file, do:

        ```python
        image = to_image(intent_router)
        with open(image_path, "wb") as imagefile:
            imagefile.write(image)
        ```
    """
    url = to_mermaid_link(intent_router, mermaid_domain)
    resp = requests.get(url, timeout=10)
    if resp.status_code >= 400:
        resp.raise_for_status()
    return resp.content

async def to_image(intent_router: IntentRouter, image_path: Path) -> None:
    """
    Saves an image of the intent's graph at the given path.

    Args:
        intent_router: the intents graph to draw.
        image_path: where to save the resulting image
    """
    image = await to_bytes(intent_router)
    with open(image_path, "wb") as imagefile:
        imagefile.write(image)

def to_mermaid_diagram(intent_router: IntentRouter) -> str:
    """
    Creates a textual representation of the intents graph in a way that Mermaid.ink can render.

    Keep in mind that this function should be able to render also malformed graphs as
    far as possible, because it can be used as a debugging tool to visualize bad bot
    configurations within error messages.

    Args:
        intent_router: the intents graph to draw.

    Returns:
        A string containing the description of the graph.
    """
    stages = {node: f'{node}["<b>{node}</b>"]' for node in intent_router.stages}
    unique_counter = 0
    connections_list = []

    processed_nodes = set()
    for origin, target, key in intent_router.graph.edges:
        # 'end' is reserved in Mermaid
        if target == "_end_":
            unique_counter += 1
            edge_string = f'{stages[origin]} -- {key} --> END{unique_counter}["<b>end</b>"]:::highlight'
        elif target == "_backtrack_":
            unique_counter += 1
            edge_string = f'{stages[origin]} -- {key} --> BACKTRACK{unique_counter}("<b>backtrack</b>"):::highlight'

        else:
            edge_string = f"{stages[origin]} -- {key} --> {stages.get(target, target)}"
        connections_list.append(edge_string)

        # Indirect connections
        if origin not in processed_nodes:
            processed_nodes.add(origin)

            if intent_router.stages[origin].accessible_from == ["_start_"]:
                edge_string = f'START("<b>start</b>"):::highlight ---> {stages[origin]}'
            elif intent_router.stages[origin].accessible_from == ["_all_"]:
                unique_counter += 1
                edge_string = f'ALL{unique_counter}("<b>all</b>"):::highlight ---> {stages[origin]}'
            elif intent_router.stages[origin].accessible_from:
                unique_counter += 1
                accessible_from_str = ",<br>".join(intent_router.stages[origin].accessible_from)
                edge_string = f'FROM{unique_counter}("<b>{accessible_from_str}</b>"):::highlight ---> {stages[origin]}'
            else:
                continue
            connections_list.append(edge_string)

    connections = "\n".join(connections_list)
    graph_styled = MERMAID_STYLED_TEMPLATE.format(connections=connections)
    log.debug("Mermaid graph created", mermaid_graph=graph_styled)
    return graph_styled

def to_mermaid_link(intent_router: IntentRouter, mermaid_domain: str = "https://mermaid.ink/img/") -> str:
    """
    Generated a URL that contains a rendering of the graph of the intents into an image.

    Args:
        intent_router: the intents graph to draw.
        mermaid_domain: the domain of your Mermaid instance, if you have your own.
            Defaults to the public mermaid.ink domain.

    Returns:
        A URL on Mermaid.ink with the graph of the intents.
    """
    graph_styled = to_mermaid_diagram(intent_router)
    graphbytes = graph_styled.encode("ascii")
    base64_bytes = base64.b64encode(graphbytes)
    base64_string = base64_bytes.decode("ascii")
    return mermaid_domain + base64_string