Connections define how messages are automatically forwarded from agents to other destinations. They enable agent pipelines, parallel processing, and message routing patterns.

Overview¶

Connections allow you to:

Forward messages between agents for sequential processing
Save messages to files with custom formatting
Process messages through custom Python functions
Create agent pipelines and workflows
Implement complex message routing patterns

LLMling-Agent supports three connection types:

Node Connection: Forward to another agent, team, or node
File Connection: Save messages to files with templating
Callable Connection: Process through Python functions

Configuration Reference¶

Node Connection Configuration¶

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)

Node Connection Configuration (YAML)
- type: node
  name: output_agent  # Name of target agent.
  connection_type: run  # How messages should be handled by the target agent:
  wait_for_completion: true  # Whether to wait for the result before continuing.
  queued: false  # Whether messages should be queued for manual processing.
  queue_strategy: latest  # How to process queued messages.
  priority: 0  # Priority of the task. Lower = higher priority.
  delay: null  # Delay before processing.
  filter_condition: null  # When to filter messages (using Talk.when()).
  stop_condition: null  # When to disconnect the connection.
  exit_condition: null  # When to exit the application (by raising SystemExit).
  transform: null  # Optional function to transform messages before forwarding.

File Connection Configuration¶

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

File Connection Configuration (YAML)
- type: file
  path: logs/messages.txt  # Path to output file. Supports variables: {date}, {time}, {agent}
  template: '  # Jinja2 template for message formatting.

    [{{ message.timestamp.strftime(''%Y-%m-%d %H:%M:%S'') }}] {{ message.name }}:
    {{ message.content }}

    {%- if message.forwarded_from %}

    (via: {{ message.forwarded_from|join('' -> '') }})

    {%- endif %}

    '
  encoding: utf-8  # File encoding to use.
  wait_for_completion: true  # Whether to wait for the result before continuing.
  queued: false  # Whether messages should be queued for manual processing.
  queue_strategy: latest  # How to process queued messages.
  priority: 0  # Priority of the task. Lower = higher priority.
  delay: null  # Delay before processing.
  filter_condition: null  # When to filter messages (using Talk.when()).
  stop_condition: null  # When to disconnect the connection.
  exit_condition: null  # When to exit the application (by raising SystemExit).
  transform: null  # Optional function to transform messages before forwarding.

Callable Connection Configuration¶

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.

Callable Connection Configuration (YAML)
- type: callable
  callable: mymodule.process_message  # Import path to the message processing function.
  kw_args: {}  # Additional kwargs to pass to the callable.
  wait_for_completion: true  # Whether to wait for the result before continuing.
  queued: false  # Whether messages should be queued for manual processing.
  queue_strategy: latest  # How to process queued messages.
  priority: 0  # Priority of the task. Lower = higher priority.
  delay: null  # Delay before processing.
  filter_condition: null  # When to filter messages (using Talk.when()).
  stop_condition: null  # When to disconnect the connection.
  exit_condition: null  # When to exit the application (by raising SystemExit).
  transform: null  # Optional function to transform messages before forwarding.

Connection Types¶

Node Connection¶

Forward messages to other agents or teams:

agents:
  analyzer:
    connections:
      - type: node
        name: "summarizer"
        connection_type: "run"
        wait_for_completion: true

Connection types:

run: Execute target agent with the message
queue: Add message to target's queue
forward: Simple message forwarding

File Connection¶

Save messages to files with Jinja2 templating:

agents:
  logger:
    connections:
      - type: file
        path: "logs/{date}/{agent}.txt"
        template: "[{timestamp}] {name}: {content}"
        encoding: "utf-8"

Callable Connection¶

Process messages through Python functions:

agents:
  processor:
    connections:
      - type: callable
        callable: "myapp.handlers:process_message"
        kw_args:
          format: "json"

Message Flow Control¶

Wait for Completion¶

Control whether to wait for the connection to complete:

connections:
  - type: node
    name: "worker"
    wait_for_completion: false  # Async, fire-and-forget

Priority¶

Control processing order when multiple messages are queued:

connections:
  - type: node
    name: "worker"
    priority: 1  # Lower numbers = higher priority

Delay¶

Add delays before processing:

connections:
  - type: node
    name: "worker"
    delay: 5.0  # Wait 5 seconds before forwarding

Conditions¶

Apply conditions to control when messages are forwarded:

connections:
  - type: node
    name: "expensive_agent"
    filter_condition:
      type: word_match
      words: ["urgent", "important"]
    stop_condition:
      type: cost_limit
      max_cost: 1.0

See Conditions Configuration for details on available condition types.

Queue Management¶

Control message queueing behavior:

connections:
  - type: node
    name: "worker"
    connection_type: "queue"
    queue_max_size: 100
    queue_trigger_size: 10  # Process when queue reaches this size

Configuration Notes¶

Connections are evaluated in the order they are defined
Multiple connections can be active simultaneously
File paths support variable substitution (date, agent name, etc.)
Callable connections can be sync or async functions
Queue connections allow batching messages for efficiency
Conditions provide fine-grained control over message flow
Priority affects order of processing when multiple messages are queued