Skip to content

forward_targets

Class info

Classes

Name Children Inherits
CallableConnectionConfig
llmling_agent.models.forward_targets
Forward messages to a callable.
    ConnectionConfig
    llmling_agent.models.forward_targets
    Base model for message forwarding targets.
    FileConnectionConfig
    llmling_agent.models.forward_targets
    Write messages to a file using a template.
      NodeConnectionConfig
      llmling_agent.models.forward_targets
      Forward messages to another node.

        🛈 DocStrings

        Forward target models.

        CallableConnectionConfig

        Bases: ConnectionConfig

        Forward messages to a callable.

        The callable can be either sync or async and should have the signature: def process_message(message: ChatMessage[Any], **kwargs) -> Any

        Any additional kwargs specified in the config will be passed to the callable.

        Source code in src/llmling_agent/models/forward_targets.py 
        163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
        class CallableConnectionConfig(ConnectionConfig):
    """Forward messages to a callable.

    The callable can be either sync or async and should have the signature:
    def process_message(message: ChatMessage[Any], **kwargs) -> Any

    Any additional kwargs specified in the config will be passed to the callable.
    """

    type: Literal["callable"] = Field("callable", init=False)
    """Connection to a callable imported from given import path."""

    callable: ImportString[Callable[..., Any]]
    """Import path to the message processing function."""

    connection_type: Literal["run"] = Field("run", init=False, exclude=True)
    """Connection type (fixed to "run")"""

    kw_args: dict[str, Any] = Field(default_factory=dict)
    """Additional kwargs to pass to the callable."""

    async def process_message(self, message: ChatMessage[Any]) -> Any:
        """Process a message through the callable.

        Handles both sync and async callables transparently.
        """
        return await execute(self.callable, message, **self.kw_args)

    def get_provider(self) -> CallbackProvider:
        """Get provider for callable."""
        from llmling_agent_providers.callback import CallbackProvider

        return CallbackProvider(self.callable, name=self.callable.__name__)

        callable instance-attribute 

        callable: ImportString[Callable[..., Any]]

        Import path to the message processing function.

        connection_type class-attribute instance-attribute 

        connection_type: Literal['run'] = Field('run', init=False, exclude=True)

        Connection type (fixed to "run")

        kw_args class-attribute instance-attribute 

        kw_args: dict[str, Any] = Field(default_factory=dict)

        Additional kwargs to pass to the callable.

        type class-attribute instance-attribute 

        type: Literal['callable'] = Field('callable', init=False)

        Connection to a callable imported from given import path.

        get_provider 

        get_provider() -> CallbackProvider

        Get provider for callable.

        Source code in src/llmling_agent/models/forward_targets.py 
        191
192
193
194
195
        def get_provider(self) -> CallbackProvider:
    """Get provider for callable."""
    from llmling_agent_providers.callback import CallbackProvider

    return CallbackProvider(self.callable, name=self.callable.__name__)

        process_message async 

        process_message(message: ChatMessage[Any]) -> Any

        Process a message through the callable.

        Handles both sync and async callables transparently.

        Source code in src/llmling_agent/models/forward_targets.py 
        184
185
186
187
188
189
        async def process_message(self, message: ChatMessage[Any]) -> Any:
    """Process a message through the callable.

    Handles both sync and async callables transparently.
    """
    return await execute(self.callable, message, **self.kw_args)

        ConnectionConfig

        Bases: BaseModel

        Base model for message forwarding targets.

        Source code in src/llmling_agent/models/forward_targets.py 
        25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
        class ConnectionConfig(BaseModel):
    """Base model for message forwarding targets."""

    type: str = Field(init=False)
    """Connection type."""

    wait_for_completion: bool = Field(True)
    """Whether to wait for the result before continuing.

    If True, message processing will wait for the target to complete.
    If False, message will be forwarded asynchronously.
    """

    queued: bool = False
    """Whether messages should be queued for manual processing."""

    queue_strategy: Literal["concat", "latest", "buffer"] = "latest"
    """How to process queued messages."""

    priority: int = 0
    """Priority of the task. Lower = higher priority."""

    delay: timedelta | None = None
    """Delay before processing."""

    filter_condition: Condition | None = None
    """When to filter messages (using Talk.when())."""

    stop_condition: Condition | None = None
    """When to disconnect the connection."""

    exit_condition: Condition | None = None
    """When to exit the application (by raising SystemExit)."""

    transform: ImportString[Callable[[Any], Any | Awaitable[Any]]] | None = None
    """Optional function to transform messages before forwarding."""

    model_config = ConfigDict(frozen=True, use_attribute_docstrings=True, extra="forbid")

        delay class-attribute instance-attribute 

        delay: timedelta | None = None

        Delay before processing.

        exit_condition class-attribute instance-attribute 

        exit_condition: Condition | None = None

        When to exit the application (by raising SystemExit).

        filter_condition class-attribute instance-attribute 

        filter_condition: Condition | None = None

        When to filter messages (using Talk.when()).

        priority class-attribute instance-attribute 

        priority: int = 0

        Priority of the task. Lower = higher priority.

        queue_strategy class-attribute instance-attribute 

        queue_strategy: Literal['concat', 'latest', 'buffer'] = 'latest'

        How to process queued messages.

        queued class-attribute instance-attribute 

        queued: bool = False

        Whether messages should be queued for manual processing.

        stop_condition class-attribute instance-attribute 

        stop_condition: Condition | None = None

        When to disconnect the connection.

        transform class-attribute instance-attribute 

        transform: ImportString[Callable[[Any], Any | Awaitable[Any]]] | None = None

        Optional function to transform messages before forwarding.

        type class-attribute instance-attribute 

        type: str = Field(init=False)

        Connection type.

        wait_for_completion class-attribute instance-attribute 

        wait_for_completion: bool = Field(True)

        Whether to wait for the result before continuing.

        If True, message processing will wait for the target to complete. If False, message will be forwarded asynchronously.

        FileConnectionConfig

        Bases: ConnectionConfig

        Write messages to a file using a template.

        The template receives the full message object for formatting. Available fields include: - timestamp: When the message was created - name: Name of the sender - content: Message content - role: Message role (user/assistant/system) - model: Model used (if any) - cost_info: Token usage and cost info - forwarded_from: Chain of message forwarding

        Source code in src/llmling_agent/models/forward_targets.py 
         98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
        class FileConnectionConfig(ConnectionConfig):
    """Write messages to a file using a template.

    The template receives the full message object for formatting.
    Available fields include:
    - timestamp: When the message was created
    - name: Name of the sender
    - content: Message content
    - role: Message role (user/assistant/system)
    - model: Model used (if any)
    - cost_info: Token usage and cost info
    - forwarded_from: Chain of message forwarding
    """

    type: Literal["file"] = Field("file", init=False)
    """Connection to a file."""

    connection_type: Literal["run"] = Field("run", init=False, exclude=True)
    """Connection type (fixed to "run")"""

    path: str
    """Path to output file. Supports variables: {date}, {time}, {agent}"""

    template: str = DEFAULT_MESSAGE_TEMPLATE
    """Jinja2 template for message formatting."""

    encoding: str = "utf-8"
    """File encoding to use."""

    def format_message(self, message: ChatMessage[Any]) -> str:
        """Format a message using the template."""
        from jinja2 import Template

        template = Template(self.template)
        return template.render(message=message)

    def resolve_path(self, context: dict[str, str]) -> UPath:
        """Resolve path template with context variables."""
        from upath import UPath

        now = datetime.now()
        date = now.strftime("%Y-%m-%d")
        time_ = now.strftime("%H-%M-%S")
        variables = {"date": date, "time": time_, **context}
        return UPath(self.path.format(**variables))

    def get_provider(self) -> CallbackProvider:
        """Get provider for file writing."""
        from jinja2 import Template
        from upath import UPath

        from llmling_agent_providers.callback import CallbackProvider

        path_obj = UPath(self.path)
        template_obj = Template(self.template)

        async def write_message(message: str) -> str:
            formatted = template_obj.render(message=message)
            path_obj.write_text(formatted + "\n", encoding=self.encoding)
            return ""

        name = f"file_writer_{path_obj.stem}"
        return CallbackProvider(write_message, name=name)

        connection_type class-attribute instance-attribute 

        connection_type: Literal['run'] = Field('run', init=False, exclude=True)

        Connection type (fixed to "run")

        encoding class-attribute instance-attribute 

        encoding: str = 'utf-8'

        File encoding to use.

        path instance-attribute 

        path: str

        Path to output file. Supports variables: {date}, {time}, {agent}

        template class-attribute instance-attribute 

        template: str = DEFAULT_MESSAGE_TEMPLATE

        Jinja2 template for message formatting.

        type class-attribute instance-attribute 

        type: Literal['file'] = Field('file', init=False)

        Connection to a file.

        format_message 

        format_message(message: ChatMessage[Any]) -> str

        Format a message using the template.

        Source code in src/llmling_agent/models/forward_targets.py 
        127
128
129
130
131
132
        def format_message(self, message: ChatMessage[Any]) -> str:
    """Format a message using the template."""
    from jinja2 import Template

    template = Template(self.template)
    return template.render(message=message)

        get_provider 

        get_provider() -> CallbackProvider

        Get provider for file writing.

        Source code in src/llmling_agent/models/forward_targets.py 
        144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
        def get_provider(self) -> CallbackProvider:
    """Get provider for file writing."""
    from jinja2 import Template
    from upath import UPath

    from llmling_agent_providers.callback import CallbackProvider

    path_obj = UPath(self.path)
    template_obj = Template(self.template)

    async def write_message(message: str) -> str:
        formatted = template_obj.render(message=message)
        path_obj.write_text(formatted + "\n", encoding=self.encoding)
        return ""

    name = f"file_writer_{path_obj.stem}"
    return CallbackProvider(write_message, name=name)

        resolve_path 

        resolve_path(context: dict[str, str]) -> UPath

        Resolve path template with context variables.

        Source code in src/llmling_agent/models/forward_targets.py 
        134
135
136
137
138
139
140
141
142
        def resolve_path(self, context: dict[str, str]) -> UPath:
    """Resolve path template with context variables."""
    from upath import UPath

    now = datetime.now()
    date = now.strftime("%Y-%m-%d")
    time_ = now.strftime("%H-%M-%S")
    variables = {"date": date, "time": time_, **context}
    return UPath(self.path.format(**variables))

        NodeConnectionConfig

        Bases: ConnectionConfig

        Forward messages to another node.

        This configuration defines how messages should flow from one node to another, including: - Basic routing (which node, what type of connection) - Message queueing and processing strategies - Timing controls (priority, delay) - Execution behavior (wait for completion)

        Source code in src/llmling_agent/models/forward_targets.py 
        65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
        class NodeConnectionConfig(ConnectionConfig):
    """Forward messages to another node.

    This configuration defines how messages should flow from one node to another,
    including:
    - Basic routing (which node, what type of connection)
    - Message queueing and processing strategies
    - Timing controls (priority, delay)
    - Execution behavior (wait for completion)
    """

    type: Literal["node"] = Field("node", init=False)
    """Connection to another node."""

    name: str
    """Name of target agent."""

    connection_type: ConnectionType = "run"
    """How messages should be handled by the target agent:
    - run: Execute message as a new run
    - context: Add message to agent's context
    - forward: Forward message to agent's outbox
    """

        connection_type class-attribute instance-attribute 

        connection_type: ConnectionType = 'run'

        How messages should be handled by the target agent: - run: Execute message as a new run - context: Add message to agent's context - forward: Forward message to agent's outbox

        name instance-attribute 

        name: str

        Name of target agent.

        type class-attribute instance-attribute 

        type: Literal['node'] = Field('node', init=False)

        Connection to another node.