cleveragents-core/docs/YAML_SYNTAX.md

CleverAgents YAML Configuration Syntax

This document provides comprehensive documentation for CleverAgents YAML configuration files, including all sections, syntax, and expected behaviors.

Table of Contents

• Configuration Structure
• Global Configuration Section
• Agents Section
• Routes Section
• Merges Section
• Splits Section
• Templates Section
• Context Section
• Functions Reference
• Complete Examples
• Best Practices

Configuration Structure

A CleverAgents configuration file follows this structure:

# Optional: Global configuration
cleveragents:
  template_engine: JINJA2  # or MUSTACHE

# Required: Agent definitions
agents:
  agent_name:
    type: agent_type
    config:
      # Agent-specific configuration

# Required: Route definitions (unified streams and graphs)
routes:
  route_name:
    type: stream|graph|bridge
    # Route-specific configuration

# Optional: Stream operations
merges:
  - sources: [source_streams]
    target: target_stream

splits:
  - source: source_stream
    targets:
      target1: condition1
      target2: condition2

# Optional: Template definitions
templates:
  agents:
    template_name:
      # Template definition
  routes:
    template_name:
      # Template definition

# Optional: Global context
context:
  global:
    variable_name: value

Global Configuration Section

Optional global settings for the entire application:

cleveragents:
  template_engine: JINJA2  # or MUSTACHE (default: JINJA2)
  verbose: false            # Enable debug logging
  unsafe: false             # ⚠️ Allow code execution (use with caution)

Template Engines

JINJA2 (Default) - Python-powered templating with conditionals, loops, and filters

• 📚 Jinja2 Documentation
• Use for: Complex logic, Python expressions, custom filters

MUSTACHE - Simple, logic-less templating

• 📚 Mustache Documentation
• Use for: Simple substitutions, cross-language compatibility

Agents Section

The agents section defines AI agents that can process messages.

Agent Types

LLM Agents

agents:
  chat_agent:
    type: llm
    config:
      provider: openai|anthropic|google
      model: gpt-4|claude-3-5-sonnet|gemini-1.5-pro
      temperature: 0.7
      max_tokens: 1000
      system_prompt: "You are a helpful assistant"
      memory_enabled: true
      max_history: 10
      api_key: "optional-api-key"

LLM Agent Configuration:

• provider: LLM provider (openai, anthropic, google)
• model: Specific model name
• temperature: Creativity level (0.0-1.0)
• max_tokens: Maximum response length
• system_prompt: System message for the agent
• memory_enabled: Enable conversation memory
• max_history: Number of messages to remember
• api_key: Optional API key (can use environment variables)

Tool Agents

Tool agents execute specific functions and operations without requiring an LLM. They provide deterministic, fast execution for tasks like mathematical calculations, file operations, HTTP requests, and more.

agents:
  tool_agent:
    type: tool
    config:
      tools: ["math", "http", "file", "json", "shell"]
      safe_mode: true
      timeout: 30

Tool Agent Configuration:

• tools: List of available tools (see below)
• safe_mode: Enable safety restrictions (recommended: true)
• timeout: Execution timeout in seconds

Available Tools:

Tool	Description	Example Use Cases
math	Mathematical calculations and expressions	Calculate totals, perform conversions, evaluate formulas
http	Make HTTP requests (GET, POST, etc.)	Call APIs, fetch web content, webhook triggers
file	Read and write files	Load data, save results, process documents
json	Parse and manipulate JSON data	Transform API responses, extract fields, validate JSON
shell	Execute shell commands (⚠️ use with caution)	Run scripts, system operations, CLI tools
echo	Return input unchanged	Testing, debugging, passthrough operations

Example - Math Tool:

agents:
  calculator:
    type: tool
    config:
      tools: ["math"]
      safe_mode: true

# Usage: Send "2 + 2" → Returns "4"

Example - HTTP Tool:

agents:
  api_client:
    type: tool
    config:
      tools: ["http", "json"]
      safe_mode: true
      timeout: 10

# Usage: Fetch data from APIs, process JSON responses

Security Note:

• Always use safe_mode: true in production
• Be cautious with shell tool - only use in trusted environments
• Set appropriate timeout values to prevent hanging operations

Composite Agents

Composite agents combine multiple agents, graphs, and streams into a single reusable unit. They act as containers that encapsulate complex multi-step workflows, making them easy to reuse and maintain.

What Composite Agents Do:

• Bundle multiple agents into one logical unit
• Orchestrate complex workflows internally
• Provide a single interface to complex processing pipelines
• Enable modular, reusable agent architectures

Why Use Composite Agents:

• Modularity: Package complex logic into reusable components
• Maintainability: Update internal logic without changing external interfaces
• Composition: Build sophisticated systems from smaller, tested units
• Encapsulation: Hide complexity behind a simple agent interface

agents:
  composite_agent:
    type: composite
    config:
      components:
        agents:
          - name: sub_agent1
            type: llm
        graphs:
          - name: sub_graph1
            type: graph
        streams:
          - name: sub_stream1
            type: stream
      routing:
        input: sub_agent1    # Entry point for messages
        output: sub_stream1  # Exit point for results

Example Use Case - Research Assistant:

agents:
  research_assistant:
    type: composite
    config:
      components:
        agents:
          - name: searcher
            type: tool
            config:
              tools: ["http"]
          - name: analyzer
            type: llm
            config:
              model: gpt-4
          - name: summarizer
            type: llm
            config:
              model: gpt-3.5-turbo
        streams:
          - name: search_stream
            type: stream
          - name: analysis_stream
            type: stream
      routing:
        input: searcher      # Start with search
        output: summarizer   # End with summary

# Use it like any other agent:
routes:
  main:
    operators:
      - type: map
        params:
          agent: research_assistant  # The entire workflow in one agent

Routes Section

The routes section defines data flow through your agent network using a unified system.

Stream Routes

routes:
  my_stream:
    type: stream  # REQUIRED
    stream_type: cold|hot|replay
    operators:
      - type: map
        params:
          agent: agent_name
      - type: filter
        params:
          condition:
            type: content_contains
            text: "keyword"
      - type: debounce
        params:
          duration: 1.0
      - type: buffer
        params:
          count: 5
      - type: throttle
        params:
          duration: 2.0
    subscriptions:
      - source_stream
    publications:
      - target_stream
      - __output__
    agents:
      - agent_name
    initial_value: "default"
    buffer_size: 10

Stream Types:

• cold: Start processing when subscribed
• hot: Always active, replay last value
• replay: Replay all previous values

Operators:

map - Transform messages using agents or functions

📚 See Functions Reference for detailed documentation on writing custom functions, message object structure, and examples.

With agent:

- type: map
  params:
    agent: my_agent  # Use an agent to transform messages

With function (simple code):

- type: map
  params:
    function: "lambda msg: msg.content.upper()"  # Convert to uppercase

More function examples:

# Add prefix to messages
- type: map
  params:
    function: "lambda msg: f'[PROCESSED] {msg.content}'"

# Extract specific field
- type: map
  params:
    function: "lambda msg: msg.metadata.get('user_id', 'unknown')"

# Simple calculations
- type: map
  params:
    function: "lambda msg: str(len(msg.content.split()))"  # Count words

filter - Filter messages based on conditions

- type: filter
  params:
    condition:
      type: content_contains
      text: "keyword"

debounce - Wait for quiet period before processing

- type: debounce
  params:
    duration: 1.0  # Wait 1 second of silence

buffer - Collect messages before processing

- type: buffer
  params:
    count: 5  # Process every 5 messages

throttle - Limit processing rate

- type: throttle
  params:
    duration: 2.0  # Max once per 2 seconds

catch - Handle errors

- type: catch
  params:
    handler: error_handler_agent

retry - Retry failed operations

- type: retry
  params:
    max_attempts: 3
    delay: 1.0

Graph Routes

routes:
  my_graph:
    type: graph  # REQUIRED
    entry_point: start
    nodes:
      process_node:
        type: agent
        agent: agent_name
      conditional_node:
        type: conditional
        condition:
          type: content_contains
          text: "keyword"
      function_node:
        type: function
        function: my_function
    edges:
      - source: start
        target: process_node
      - source: process_node
        target: conditional_node
        condition:
          type: metadata_has
          key: success
      - source: conditional_node
        target: end
    checkpointing: true
    checkpoint_dir: "./checkpoints"
    enable_time_travel: true
    parallel_execution: true
    state_class: "my_module.MyState"

Node Types:

• agent: Process using an agent
• conditional: Conditional routing
• function: Custom function execution

Graph Features:

• checkpointing: Save state for recovery
• enable_time_travel: Allow state rollback
• parallel_execution: Run nodes in parallel
• state_class: Custom state management

Bridge Routes

routes:
  adaptive_route:
    type: stream
    # ... stream configuration ...
    bridge:
      upgrade_conditions:
        needs_state: true
        message_count: 5
        custom_predicate: "lambda msg, cfg: 'NEEDS_STATE' in msg.content"
      downgrade_conditions:
        idle_time: 300
        state_size: 2
        no_conditionals_used: true
      state_extractor: "lambda msg: {'last_message': msg.content}"
      state_flattener: "lambda state: state.data.get('summary', '')"
      preserve_subscriptions: true
      preserve_checkpointing: true

Merges Section

Merge multiple streams into one:

merges:
  - sources: [stream1, stream2, stream3]
    target: combined_stream
  - sources: [__input__]
    target: main_processor

Splits Section

Split one stream into multiple streams:

splits:
  - source: input_stream
    targets:
      high_priority:
        type: content_contains
        text: "urgent"
      normal_priority:
        type: content_not_contains
        text: "urgent"
      error_stream:
        type: metadata_has
        key: error

Split Conditions:

• content_contains: Message contains text
• content_not_contains: Message doesn't contain text
• metadata_has: Message has metadata key
• metadata_equals: Metadata equals value
• custom: Custom condition function

Templates Section

Define reusable templates:

templates:
  agents:
    basic_llm:
      type: llm
      config:
        provider: openai
        model: gpt-4
        temperature: 0.7
        system_prompt: "{{system_message}}"

  routes:
    chat_stream:
      type: stream
      stream_type: cold
      operators:
        - type: map
          params:
            agent: "{{agent_name}}"
      publications:
        - __output__

  graphs:
    simple_workflow:
      type: graph
      nodes:
        process:
          type: agent
          agent: "{{agent_name}}"
      edges:
        - source: start
          target: process
        - source: process
          target: end

# Use templates
agents:
  my_agent:
    template: basic_llm
    params:
      system_message: "You are a helpful assistant"

  my_chat:
    template: chat_stream
    params:
      agent_name: my_agent

Context Section

Global variables and configuration:

context:
  global:
    app_name: "My CleverAgents App"
    debug_mode: true
    max_retries: 3
    timeout: 30
    api_keys:
      openai: "sk-your-key"
      anthropic: "your-key"
    user_preferences:
      language: "en"
      theme: "dark"

Functions Reference

Functions allow you to process messages using Python code without creating agents. This is useful for lightweight transformations, filtering, and data manipulation.

Message Object Structure

Functions receive a StreamMessage object with the following attributes:

class StreamMessage:
    content: Any              # The main message content (string, dict, list, etc.)
    metadata: Dict[str, Any]  # Additional metadata about the message
    source_stream: str        # Name of the stream that produced this message
    timestamp: float          # Unix timestamp when message was created

Writing Functions

Functions are Python lambda expressions that take a msg parameter (the StreamMessage object) and return a value.

Basic Syntax:

"lambda msg: <expression>"

Common Patterns

1. Content Transformation

# Convert to uppercase
function: "lambda msg: msg.content.upper()"

# Strip whitespace and lowercase
function: "lambda msg: msg.content.strip().lower()"

# Add prefix/suffix
function: "lambda msg: f'[PROCESSED] {msg.content}'"

# Replace text
function: "lambda msg: msg.content.replace('old', 'new')"

2. Metadata Access

# Extract metadata field
function: "lambda msg: msg.metadata.get('user_id', 'unknown')"

# Check metadata existence
function: "lambda msg: 'error' in msg.metadata"

# Combine content with metadata
function: "lambda msg: f'{msg.metadata.get(\"user\")}: {msg.content}'"

3. Data Extraction

# Count words
function: "lambda msg: str(len(msg.content.split()))"

# Extract first word
function: "lambda msg: msg.content.split()[0] if msg.content else ''"

# Parse JSON content
function: "lambda msg: msg.content.get('field') if isinstance(msg.content, dict) else ''"

4. Filtering Conditions

# Length check
function: "lambda msg: len(msg.content) > 10"

# Content check
function: "lambda msg: 'urgent' in msg.content.lower()"

# Metadata check
function: "lambda msg: msg.metadata.get('priority', 0) > 5"

# Time-based filter
function: "lambda msg: msg.timestamp > some_timestamp"

5. Complex Transformations

# Multi-step processing
function: "lambda msg: msg.content.strip().lower().replace('  ', ' ')"

# Conditional transformation
function: "lambda msg: msg.content.upper() if len(msg.content) < 10 else msg.content"

# JSON manipulation
function: "lambda msg: {**msg.content, 'processed': True} if isinstance(msg.content, dict) else msg.content"

Function Best Practices

DO:

• ✅ Keep functions simple and focused on one task
• ✅ Use descriptive variable names even in lambdas
• ✅ Handle edge cases (empty strings, missing keys, etc.)
• ✅ Use .get() with defaults when accessing dict keys
• ✅ Chain simple transformations for readability

DON'T:

• ❌ Write complex multi-line logic in functions
• ❌ Perform I/O operations (use tool agents instead)
• ❌ Access external state or variables
• ❌ Modify mutable objects in metadata
• ❌ Use functions for operations that need LLM reasoning

When to Use Functions vs Agents

Use Functions For	Use Agents For
Text transformation	AI reasoning and generation
Data extraction	Complex decision making
Filtering	Natural language understanding
Simple calculations	Context-aware responses
Format conversion	Multi-step reasoning
Validation	Creative tasks

Complete Function Examples

Example 1: Message Sanitizer

routes:
  sanitizer:
    type: stream
    operators:
      # Remove extra whitespace
      - type: map
        params:
          function: "lambda msg: ' '.join(msg.content.split())"

      # Remove special characters
      - type: map
        params:
          function: "lambda msg: ''.join(c for c in msg.content if c.isalnum() or c.isspace())"

      # Truncate to 100 characters
      - type: map
        params:
          function: "lambda msg: msg.content[:100] + '...' if len(msg.content) > 100 else msg.content"

Example 2: Smart Router

routes:
  router:
    type: stream
    operators:
      # Add routing metadata
      - type: map
        params:
          function: "lambda msg: msg.copy_with(metadata={**msg.metadata, 'route': 'urgent' if 'urgent' in msg.content.lower() else 'normal'})"

Example 3: Data Enrichment

routes:
  enricher:
    type: stream
    operators:
      # Add word count
      - type: map
        params:
          function: "lambda msg: msg.copy_with(metadata={**msg.metadata, 'word_count': len(msg.content.split())})"

      # Add character count
      - type: map
        params:
          function: "lambda msg: msg.copy_with(metadata={**msg.metadata, 'char_count': len(msg.content)})"

Python Lambda Syntax Reference

For more information on Python lambda functions, see:

• 📚 Python Lambda Expressions
• 📚 Python String Methods
• 📚 Python Dictionary Methods

Complete Examples

Basic Chat with Memory

agents:
  chat_agent:
    type: llm
    config:
      provider: openai
      model: gpt-4
      temperature: 0.8
      memory_enabled: true
      max_history: 10
      system_prompt: |
        You are a helpful and friendly AI assistant with memory.
        You can remember our conversation and refer back to previous topics.

routes:
  chat_stream:
    type: stream
    stream_type: cold
    operators:
      - type: map
        params:
          agent: chat_agent
    publications:
      - __output__

merges:
  - sources: [__input__]
    target: chat_stream

context:
  global:
    conversation_mode: true
    memory_enabled: true

Simple Message Processing with Functions

Process messages using simple code without agents:

routes:
  message_processor:
    type: stream
    stream_type: cold
    operators:
      # Clean up input
      - type: map
        params:
          function: "lambda msg: msg.content.strip().lower()"

      # Filter out short messages
      - type: filter
        params:
          function: "lambda msg: len(msg.content) > 5"

      # Add timestamp prefix
      - type: map
        params:
          function: "lambda msg: f'[{msg.metadata.get(\"timestamp\", \"now\")}] {msg.content}'"

      # Only process during business hours (example)
      - type: filter
        params:
          function: "lambda msg: 9 <= int(msg.metadata.get('hour', 12)) <= 17"

    publications:
      - __output__

merges:
  - sources: [__input__]
    target: message_processor

# Usage: Lightweight processing without LLM costs
# Input: "  HELLO WORLD  " → Output: "[now] hello world"

Multi-Agent Research Pipeline

agents:
  researcher:
    type: llm
    config:
      provider: openai
      model: gpt-4
      system_prompt: "Research and gather information on topics"

  analyzer:
    type: llm
    config:
      provider: openai
      model: gpt-4
      system_prompt: "Analyze and synthesize research findings"

  writer:
    type: llm
    config:
      provider: openai
      model: gpt-4
      system_prompt: "Write comprehensive reports based on analysis"

routes:
  research_stream:
    type: stream
    stream_type: cold
    operators:
      - type: map
        params:
          agent: researcher
      - type: buffer
        params:
          time: 2.0
    publications:
      - analysis_stream

  analysis_stream:
    type: stream
    stream_type: cold
    operators:
      - type: map
        params:
          agent: analyzer
    publications:
      - writing_stream

  writing_stream:
    type: stream
    stream_type: cold
    operators:
      - type: map
        params:
          agent: writer
    publications:
      - __output__

merges:
  - sources: [__input__]
    target: research_stream

Stateful Workflow with LangGraph

agents:
  classifier:
    type: llm
    config:
      provider: openai
      model: gpt-3.5-turbo
      system_prompt: "Classify the input as: question, command, or statement"

  question_handler:
    type: llm
    config:
      provider: openai
      model: gpt-4
      system_prompt: "Answer questions concisely and accurately"

  command_handler:
    type: tool
    config:
      tools: ["execute", "search"]

routes:
  workflow_graph:
    type: graph
    entry_point: start
    nodes:
      classify:
        type: agent
        agent: classifier
      handle_question:
        type: agent
        agent: question_handler
      handle_command:
        type: agent
        agent: command_handler
    edges:
      - source: start
        target: classify
      - source: classify
        target: handle_question
        condition:
          type: content_contains
          text: "question"
      - source: classify
        target: handle_command
        condition:
          type: content_contains
          text: "command"
      - source: handle_question
        target: end
      - source: handle_command
        target: end
    checkpointing: true

  input_stream:
    type: stream
    stream_type: cold
    operators:
      - type: graph_execute
        params:
          graph: workflow_graph
    publications:
      - __output__

merges:
  - sources: [__input__]
    target: input_stream

Best Practices

1. Configuration Organization

• Use descriptive names for agents and routes
• Group related configurations together
• Use templates for reusable patterns
• Keep configurations modular

2. Memory Management

• Enable memory for conversational agents
• Set appropriate max_history limits
• Use context variables for global state

3. Error Handling

• Use catch operators for error handling
• Implement retry logic for unreliable operations
• Use splits to route errors to error handlers

4. Performance

• Use throttle and debounce for rate limiting
• Implement buffer for batch processing
• Use hot streams for real-time updates

5. Testing

• Start with simple configurations
• Test each component individually
• Use verbose mode for debugging
• Validate configurations before deployment

Common Issues

Configuration Errors

• Missing type field in routes (REQUIRED)
• Invalid agent types
• Circular dependencies in merges/splits
• Missing required configuration fields

Runtime Errors

• API key not configured
• Invalid model names
• Network connectivity issues
• Memory limits exceeded

Debugging Tips

• Use --verbose flag for detailed logging
• Check agent capabilities with /help in interactive mode
• Validate YAML syntax before running
• Test with simple configurations first

Additional Resources

• CleverAgents Documentation
• RxPy Documentation
• LangGraph Documentation
• YAML Syntax Reference

For more examples and advanced configurations, see the examples/ directory in the CleverAgents repository.