message.py

Overview

The message.py file defines a message generation component for the InfiniFlow system, intended to produce dynamic textual content based on predefined templates and input parameters. It supports both streaming and batch generation of messages, including the use of Jinja2 templating for advanced formatting. The component can interpolate variables, handle partial or asynchronous content streaming, and ensures execution timeouts for stability.

This component is likely used within a larger agent or workflow system where dynamic message generation is required, such as chatbots, notifications, or interactive flows.

Classes and Functions

Class: MessageParam

Defines the parameters for the Message component.

Description

• Inherits from ComponentParamBase.

• Holds configuration data necessary for generating messages.

• Validates the parameters to ensure correctness.

Attributes

Methods

• check() -> bool: Validates that content is not empty and stream is a boolean value. Raises errors if validation fails.

Usage Example

params = MessageParam()
params.content = ["Hello, {user}!"]
params.stream = False
params.check()  # Returns True if valid

Class: Message

The main component class responsible for generating messages.

Description

• Inherits from ComponentBase.

• Uses templates from content parameter to create messages.

• Supports variable interpolation, streaming, and Jinja2 templating.

• Honors a configurable execution timeout to prevent long-running tasks.

Class Attributes

Methods

get_kwargs(script: str, kwargs: dict = {}, delimeter: str = None) -> tuple[str, dict[str, str | list | Any]]

Parses input template text (script) to extract variable placeholders and prepares keyword arguments for rendering.

• Parameters:

  • script (str): The template text containing placeholders.

  • kwargs (dict, optional): Existing keyword arguments to incorporate.

  • delimeter (str, optional): Delimiter to join list-type variables into strings.

• Returns:

  • tuple: A tuple containing:

    • The processed script string with placeholders replaced by sanitized variable names.

    • A dictionary of keyword arguments ready for template rendering.

• Details:

  • Extracts input elements from the script.

  • Converts partial functions, lists (joined by delimiter), or other types to string representations.

  • Replaces special characters (@, ., :) in placeholder names with underscores for compatibility.

  • Updates the component's input with the processed variable values.

• Example:

script = "Hello, {%user.name%}! Your score is {%score%}."
kwargs = {"score": 42}
processed_script, processed_kwargs = message_instance.get_kwargs(script, kwargs)
# processed_script: "Hello, {user_name}! Your score is {score}."
# processed_kwargs: {"user_name": ..., "score": 42}

_stream(rand_cnt: str) -> Generator[str, None, None]

Generates content incrementally by streaming parts of the message template.

• Parameters:

  • rand_cnt (str): The selected message template string to stream.

• Yields:

  • Chunks of the message content as strings.

• Details:

  • Uses regex to find variable placeholders within the template.

  • Looks up variable values from the internal canvas (context).

  • Handles partial functions by yielding their output incrementally.

  • Caches variable expansions to avoid redundant computation.

  • Accumulates and sets the full combined content as the output at the end.

• Use Case:

  • Useful for streaming large or dynamically generated messages, e.g., chat responses.

_is_jinjia2(content: str) -> bool

Checks if the given string contains Jinja2 template syntax.

• Parameters:

  • content (str): The content string to check.

• Returns:

  • bool: True if Jinja2 delimiters or tags are found, else False.

• Details:

  • Searches for patterns like {% ... %}, {{, and }}.

_invoke(**kwargs)

Main execution method called to generate the message content.

• Parameters:

  • **kwargs: Keyword arguments for variable substitution in the message.

• Returns:

  • None (sets component output internally).

• Details:

  • Selects a random template from the content list.

  • If streaming is enabled and the template does not use Jinja2 syntax, sets output to a streaming generator.

  • Otherwise, processes the template with get_kwargs and renders it with Jinja2.

  • Performs a secondary regex substitution of variables in the final rendered content as a fallback.

  • Uses a timeout decorator (default 10 minutes) to limit execution time.

• Example:

message = Message()
message._param.content = ["Hello, {{ user }}!"]
message._param.stream = False
message._invoke(user="Alice")
print(message.get_output("content"))  # "Hello, Alice!"

thoughts() -> str

Returns an empty string; placeholder for potential future use or interface compliance.

Important Implementation Details

• Variable Placeholder Syntax:
  Variables inside templates use a custom syntax like {%variable_name%} which are parsed and replaced. The system replaces special characters in variable names to underscores for safe Jinja2 processing.

• Streaming Support:
  When stream is enabled and the template is not a Jinja2 template, the message content is streamed as a generator to allow partial content delivery, useful for large or real-time messages.

• Jinja2 Integration:
  The component leverages Jinja2 templating for complex message formatting and variable interpolation, providing powerful template capabilities.

• Timeout Handling:
  The _invoke method is decorated with a timeout to prevent long-running executions, controlled by the COMPONENT_EXEC_TIMEOUT environment variable or defaulting to 10 minutes.

• Partial Functions Support:
  The component can handle functools.partial wrapped generators or functions as variable values, iterating over them to yield content progressively.

Interaction With Other System Components

• Base Classes:
  Inherits from ComponentBase and ComponentParamBase which likely provide core component lifecycle, parameter management, and input/output handling capabilities.

• Template Rendering:
  Utilizes Jinja2 (jinja2.Template) for template compilation and rendering.

• Agent Canvas:
  Accesses self._canvas.get_variable_value(exp) to resolve variables dynamically at runtime, indicating integration with a shared variable or context storage managed by the agent system.

• Timeout Utility:
  Uses the timeout decorator from api.utils.api_utils to guard execution time.

• Input/Output Management:
  Uses self.set_input_value and self.set_output methods to manage component inputs and outputs, adhering to the component framework's conventions.

Visual Diagram: Class Structure

classDiagram
    class MessageParam {
        +content: list
        +stream: bool
        +outputs: dict
        +check() bool
    }
    class Message {
        +component_name: str
        +get_kwargs(script: str, kwargs: dict, delimeter: str) tuple
        -_stream(rand_cnt: str) generator
        -_is_jinjia2(content: str) bool
        -_invoke(**kwargs) void
        +thoughts() str
    }
    MessageParam <|-- Message

Summary

The message.py file implements a flexible and powerful message generation component within the InfiniFlow agent system. It supports templated messages with variable interpolation, streaming output, and Jinja2 templating integration. It is designed to be robust with validation, timeout controls, and partial content streaming, making it suitable for interactive or real-time applications requiring dynamic textual output.

This component is tightly coupled with the agent's variable context (_canvas), the component base framework, and utility functions, enabling it to serve as a dynamic message producer in workflows or conversational agents.