Back to Research

Agents are classes that compose like lego blocks

Mahmoud Tolba, MNN

Authors

Nov 1, 2025

6 min read

Agents are classes that compose like lego blocks

What’s New?

This guide demonstrates how to build a production-ready hierarchical multi-agent architecture using DSPy—a programmatic, Pythonic framework for composing intelligent agents.

The DSPy Multi-Agent Architecture Pattern

Traditional agent frameworks often require complex configuration files, service meshes, and custom routing logic. DSPy takes a fundamentally different approach - agents are Python classes that compose like LEGO blocks.

The breakthrough insight: Subagents can be wrapped as tools for parent agents. This creates clean hierarchical delegation without any infrastructure overhead.

Core Architecture Patterns

1. Signatures for Type Safety: Define clear input/output contracts using dspy.Signature with descriptions and type hints for better code clarity.

2. ReAct Module for Reasoning: Built-in reasoning-action-observation dspy.Reactloops that automatically select tools based on context—no manual prompt engineering required.

3. Hierarchical Composition Through Wrapping: Each subagent is a dspy.Module with its own tools. Subagents wrap as callable functions, and the main agent receives them as tools. This creates a clean delegation hierarchy: MainAgent → SubAgents → Tools. Here's how the multi-agent architecture works in our example.

Key points:

Each subagent is a dspy.Module with its own Signature and ReAct program.

The @trace_agent decorator provides execution visibility without modifying agent logic.

Subagents wrap as callable functions (e.g., weather_checker(), time_checker()).

The main agent treats these wrapped subagents as tools in its ReAct program.

ReAct handles reasoning ("which tool should I use?") and action ("call that tool").

The main agent can call multiple subagents sequentially (iteration 1, iteration 2, etc.) to handle multi-part queries.

Results flow back up the hierarchy automatically and are synthesized into a coherent final answer.

The beauty of this design: no hidden routing layer, no service mesh—just pure Python function calls. The main agent doesn't know it's calling other agents; it simply calls functions and receives results.

Code Walkthrough

Let's break down the code cell by cell to understand how this hierarchical agent system works.

python

import dspy
from typing import List
import json
from datetime import datetime
import pytz
import requests
import time
from functools import wraps

We import DSPy for agent orchestration, along with standard libraries for time handling and API calls.

python

## Trajectory Inspection

def print_react_trajectory(prediction: dspy.Prediction):
    """Print ReAct trajectory in compact format"""
    if not hasattr(prediction, 'trajectory') or not prediction.trajectory:
        return

    trajectory = prediction.trajectory
    i = 0
    while f'thought_{i}' in trajectory:
        print(f"\n   Thinking: {trajectory.get(f'thought_{i}', '')}")
        print(f"   Action: {trajectory.get(f'tool_name_{i}', '')}({trajectory.get(f'tool_args_{i}', {})})")
        print(f"   Result: {trajectory.get(f'observation_{i}', '')}")
        i += 1

def trace_agent(agent_name: str):
    """Decorator to trace agent execution"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"\n[{agent_name}] STARTED - Input: {kwargs}")
            start_time = time.time()

            try:
                result = func(*args, **kwargs)

                if isinstance(result, dspy.Prediction):
                    print_react_trajectory(result)
                    print("\n   Final Answer:")
                    for key, value in result.items():
                        if key not in ['trajectory', 'reasoning']:
                            print(f"      {key}: {value}")

                print(f"\n[{agent_name}] FINISHED - {time.time() - start_time:.2f}s\n")
                return result

            except Exception as e:
                print(f"\n[{agent_name}] FAILED - {str(e)} ({time.time() - start_time:.2f}s)\n")
                raise
        return wrapper
    return decorator

Why observability matters: The @trace_agentdecorator provides crucial visibility into agent behavior: 1. Visual Execution Flow: See exactly when agents start/stop and how long they take 2. ReAct Trajectory Inspection: Watch the reasoning loop in real-time—what the agent thinks, which tools it chooses, and what results it gets 3. Debugging & Optimization: Quickly identify bottlenecks, failed tool calls, or incorrect reasoning 4. Production Monitoring: Essential for understanding agent behavior in complex multi-agent systems This decorator transforms opaque agent execution into transparent, debuggable workflows. You'll see output like:

markdown

 [MathAgent] STARTED
   Input: {'math_query': 'What is 5+3?'}

   Thinking: I need to add two numbers
   Action: add_numbers({'a': 5, 'b': 3})
   Result: 8

   Final Answer:
      math_result: 8

   Completed in 1.23s
 [MathAgent] FINISHED

Tools definition

python

def add_numbers(a: float, b: float) -> float:
    """Add two numbers together and return the sum."""
    return a + b

def multiply_numbers(a: float, b: float) -> float:
    """Multiply two numbers together and return the product."""
    return a * b

math_tools = [add_numbers, multiply_numbers

Each tool is a simple Python function with a clear docstrings. DSPy's ReAct module reads these docstrings to understand what each tool does and when to use it.

python

class MathAgentSignature(dspy.Signature):
    """Signature for mathematical operations agent"""
    math_query: str = dspy.InputField(desc="A mathematical question or operation request")
    math_result: str = dspy.OutputField(desc="The result of the mathematical operation")

class MathAgent(dspy.Module):
    """Agent that handles mathematical operations using ReAct."""
    def __init__(self):
        super().__init__()
        self.react_program = dspy.ReAct(
            signature=MathAgentSignature,
            tools=math_tools,
            max_iters=3
        )

    @trace_agent("MathAgent")
    def forward(self, math_query: str) -> dspy.Prediction:
        """Process mathematical queries and return results"""
        return self.react_program(math_query=math_query)

Key insights:

Signature defines the input/output contract: math_query in, math_result out.

ReAct is DSPy's built-in reasoning-action module. It automatically:

Decides which tool to use

Calls the appropriate tool

Observes the result

Repeats if needed (up to 3 iterations)

Module pattern makes agents composable—each agent is a reusable component.

@trace_agent decorator wraps the forward method to provide execution visibility without changing agent logic.

Text Agent (Similar Structure)

python

### Text Agent

def count_words(text: str) -> int:
    """Count the number of words in the given text."""
    return len(text.split())

def reverse_text(text: str) -> str:
    """Reverse the given text and return it backwards."""
    return text[::-1]

text_tools = [count_words, reverse_text]

class TextAgentSignature(dspy.Signature):
    """Signature for text processing operations agent"""
    text_query: str = dspy.InputField(desc="A text processing question or operation request")
    text_result: str = dspy.OutputField(desc="The result of the text processing operation")

class TextAgent(dspy.Module):
    """Agent that handles text operations using ReAct."""

    def __init__(self):
        super().__init__()
        self.react_program = dspy.ReAct(
            signature=TextAgentSignature,
            tools=text_tools,
            max_iters=3
        )

    @trace_agent("TextAgent")
    def forward(self, text_query: str) -> dspy.Prediction:
        """Process text queries and return results"""
        return self.react_program(text_query=text_query)

Same pattern as MathAgent, but with text-specific tools. This consistency makes the codebase easy to understand and extend. All agents use the @trace_agent decorator for uniform observability.

python

### Time Agent - Real-World API Integration

def get_usa_time() -> str:
    """Get the current time in USA (Eastern Time)."""
    usa_tz = pytz.timezone('America/New_York')
    usa_time = datetime.now(usa_tz)
    return usa_time.strftime("%Y-%m-%d %H:%M:%S %Z")

def get_china_time() -> str:
    """Get the current time in China (Beijing Time)."""
    china_tz = pytz.timezone('Asia/Shanghai')
    china_time = datetime.now(china_tz)
    return china_time.strftime("%Y-%m-%d %H:%M:%S %Z")

time_tools = [get_usa_time, get_china_time]

class TimeAgentSignature(dspy.Signature):
    """Find the timezone or location in the input query and return the corresponding time."""
    time_query: str = dspy.InputField(desc="A time-related question or timezone request")
    time_result: str = dspy.OutputField(desc="The current time in the requested timezone")

class TimeAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.react_program = dspy.ReAct(
            signature=TimeAgentSignature,
            tools=time_tools,
            max_iters=3
        )

    @trace_agent("TimeAgent")
    def forward(self, time_query: str) -> dspy.Prediction:
        return self.react_program(time_query=time_query)

Why this works better

Tools can be anything—pure functions, API calls, database queries. DSPy doesn't care. It just sees callable functions.

python

### Weather Agent - Complex External APIs

def get_weather_by_city(city_name: str) -> str:
    """Get current weather information for a given city using Open-Meteo API."""
    try:
        # Geocoding to get coordinates
        geocoding_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city_name}&count=1"
        geo_response = requests.get(geocoding_url, timeout=10)
        geo_data = geo_response.json()

        if not geo_data.get('results'):
            return f"City '{city_name}' not found"

        location = geo_data['results'][0]
        lat, lon = location['latitude'], location['longitude']

        # Get weather data
        weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,wind_speed_10m"
        weather_response = requests.get(weather_url, timeout=10)
        weather_data = weather_response.json()

        current = weather_data['current']
        return f"Weather in {location['name']}, {location.get('country')}: Temperature: {current['temperature_2m']}°C, Humidity: {current['relative_humidity_2m']}%, Wind Speed: {current['wind_speed_10m']} km/h"

    except Exception as e:
        return f"Error fetching weather: {str(e)}"

def compare_city_temperatures(city1: str, city2: str) -> str:
    """Compare temperatures between two cities."""
    try:
        temps = {}
        for city in [city1, city2]:
            # Get coordinates and weather data
            geocoding_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1"
            geo_response = requests.get(geocoding_url, timeout=10)
            geo_data = geo_response.json()

            location = geo_data['results'][0]
            weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={location['latitude']}&longitude={location['longitude']}&current=temperature_2m"
            weather_response = requests.get(weather_url, timeout=10)
            temps[city] = weather_response.json()['current']['temperature_2m']

        diff = abs(temps[city1] - temps[city2])
        warmer = city1 if temps[city1] > temps[city2] else city2
        return f"{city1}: {temps[city1]}°C, {city2}: {temps[city2]}°C. {warmer} is warmer by {diff}°C"
    except Exception as e:
        return f"Error comparing temperatures: {str(e)}"

weather_tools = [get_weather_by_city, compare_city_temperatures]

class WeatherAgentSignature(dspy.Signature):
    """Find weather and temperature information and structure into the requested output form."""
    weather_query: str = dspy.InputField(desc="A weather-related question about cities or temperature comparison")
    weather_result: str = dspy.OutputField(desc="Weather information or temperature comparison results")

class WeatherAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.react_program = dspy.ReAct(
            signature=WeatherAgentSignature,
            tools=weather_tools,
            max_iters=3
        )

    @trace_agent("WeatherAgent")
    def forward(self, weather_query: str) -> dspy.Prediction:
        return self.react_program(weather_query=weather_query)

Integration in production

The WeatherAgent shows how to wrap complex external APIs as tools. Error handling, API calls, data parsing—all hidden inside the tool function.

python

### The Magic - Wrapping Subagents as Tools

def math_calculator(math_query: str) -> str:
    """
    Math Calculator Agent: Performs mathematical operations
    - **Addition**: Adds two numbers together
    - **Multiplication**: Multiplies two numbers together
    - **Smart Operation Selection**: ReAct decides which tool to use
    - Output key: "math_result"
    """
    math_agent = MathAgent()
    prediction = math_agent(math_query=math_query)
    return prediction.math_result

def text_processor(text_query: str) -> str:
    """
    Text Processor Agent: Performs text operations
    - **Word Counter**: Counts words in text
    - **Text Reverser**: Reverses text backwards
    - **Smart Operation Selection**: ReAct decides which tool to use
    - Output key: "text_result"
    """
    text_agent = TextAgent()
    prediction = text_agent(text_query=text_query)
    return prediction.text_result

def time_checker(time_query: str) -> str:
    """
    Time Checker Agent: Provides current time information
    - **USA Time**: Returns current time in USA (Eastern Time)
    - **China Time**: Returns current time in China (Beijing Time)
    - **Smart Timezone Selection**: ReAct decides which tool to use
    - Output key: "time_result"
    """
    time_agent = TimeAgent()
    prediction = time_agent(time_query=time_query)
    return prediction.time_result

def weather_checker(weather_query: str) -> str:
    """
    Weather Checker Agent: Provides real-time weather information
    - **City Weather**: Gets current weather for any city worldwide
    - **Temperature Comparison**: Compares temperatures between two cities
    - **Smart Tool Selection**: ReAct decides which weather tool to use
    - Output key: "weather_result"
    """
    weather_agent = WeatherAgent()
    prediction = weather_agent(weather_query=weather_query)
    return prediction.weather_result

The key innovation

Each subagent is wrapped in a simple function that:

Instantiates the agent

Calls it with the query

Returns the result as a string

These wrapper functions become tools for the main agent. The main agent treats them as regular tools—unaware that entire agents operate underneath.

python

### Main Agent - The Intelligent Coordinator

class MainAgentSignature(dspy.Signature):
    """You are an intelligent coordinator that breaks down complex queries into sub-tasks.

    Your responsibilities:
    1. Analyze the user's query and identify distinct tasks (math, text, time, weather)
    2. Break multi-part queries into separate sub-queries for each specialist agent
    3. Call the appropriate agent tools in sequence to gather all needed information
    4. Synthesize results from multiple agents into a coherent final answer

    Available agents and their capabilities:
    - math_calculator: Mathematical operations (addition, multiplication)
    - text_processor: Text operations (word count, text reversal)
    - time_checker: Time queries (USA time, China time)
    - weather_checker: Weather queries (city weather, temperature comparisons)

    For multi-part queries like "What's 5+3 and weather in Cairo?", call math_calculator first,
    then weather_checker, and combine both results in your final answer.
    """
    user_query: str = dspy.InputField(desc="User's potentially multi-part query requiring one or more specialist agents")
    final_answer: str = dspy.OutputField(desc="Complete answer combining results from all relevant agents")


class MainAgent(dspy.Module):
    """Main coordinator agent that intelligently delegates to specialist subagents.

    This agent can:
    - Handle single queries: "What's the time in USA?" → routes to time_checker
    - Handle multi-part queries: "What's 5+3 and weather in Cairo?" → routes to both math_calculator and weather_checker
    - Coordinate multiple agent calls in sequence
    - Synthesize results from multiple agents into coherent answers
    """

    def __init__(self):
        super().__init__()
        # Configure ReAct with all specialist agents as tools
        self.root_program = dspy.ReAct(
            signature=MainAgentSignature,
            tools=[math_calculator, text_processor, time_checker, weather_checker],
            max_iters=5  # Allow multiple iterations for multi-part queries
        )

    def forward(self, user_query: str):
        results = self.root_program(user_query=user_query)
        return results

The orchestrator

The MainAgent uses the same ReAct pattern, but its tools are the wrapped subagents. When a user asks "What's 5+3?", the MainAgent's ReAct module:

python

### Configuration and Execution

# Configure the language model
dspy.configure(lm=dspy.LM('ollama_chat/qwen3:4b',
                          api_base='http://localhost:11434',
                          api_key=''))

# Initialize the root agent
root_agent = MainAgent()

# Test with a multi-part query
if __name__ == "__main__":
    test_query = "Can you tell me the weather in Berlin and the current time in China?"
    result = root_agent(user_query=test_query)
    print(result.final_answer)

Output of the agent

markdown


 [WeatherAgent] STARTED
    Input: {'weather_query': 'Berlin'}

   Thinking: You need to retrieve the current weather information for Berlin using the get_weather_by_city tool.
   Action: get_weather_by_city({'city_name': 'Berlin'})
   Result: Weather in Berlin, Germany: Temperature: 11.9°C, Humidity: 62%, Wind Speed: 15.5 km/h

   Thinking: The weather information for Berlin has been retrieved and contains the necessary details.
   Action: finish({})
   Result: Completed.

   Final Answer:
      weather_result: Weather in Berlin, Germany: Temperature: 11.9°C, Humidity: 62%, Wind Speed: 15.5 km/h

   Completed in 19.89s
 [WeatherAgent] FINISHED


 [TimeAgent] STARTED
    Input: {'time_query': 'China'}

   Thinking: Choosing to get the current time in China (Beijing Time) since the query is about China.
   Action: get_china_time({})
   Result: 2025-11-03 19:53:50 CST

   Thinking: The task is complete as the time for China (Beijing Time) has been retrieved.
   Action: finish({})
   Result: Completed.

   Final Answer:
     time_result: 2025-11-03 19:53:50 CST

   Completed in 18.97s
 [TimeAgent] FINISHED

The weather in Berlin is 11.9°C with 62% humidity and 15.5 km/h wind. The current time in China is 2025-11-03 19:53:50 CST.

Model agnostic: DSPy works with any LLM—OpenAI, Anthropic, local models via Ollama, or cloud providers.

Takeaways

DSPy makes multi-agent systems Pythonic - No folder structure, just Python classes

Signatures provide type safety - Clear contracts between components

ReAct is built-in - No need to implement reasoning loops manually

Agents compose cleanly - Subagents become tools through simple wrapping

Model agnostic - Works with any LLM provider

Observability through decorators - The @trace_agent decorator provides execution visibility without modifying agent logic

With DSPy's programmatic approach, you can build, test, and iterate quickly. The code is easy to debug, version control, and extend. Most importantly, it feels like writing regular Python—agents are just classes, tools are just functions, and composition is just passing functions as arguments.

## Trajectory Inspection def print_react_trajectory(prediction: dspy.Prediction): """Print ReAct trajectory in compact format""" if not hasattr(prediction, 'trajectory') or not prediction.trajectory: return trajectory = prediction.trajectory i = 0 while f'thought_{i}' in trajectory: print(f"\n Thinking: {trajectory.get(f'thought_{i}', '')}") print(f" Action: {trajectory.get(f'tool_name_{i}', '')}({trajectory.get(f'tool_args_{i}', {})})") print(f" Result: {trajectory.get(f'observation_{i}', '')}") i += 1 def trace_agent(agent_name: str): """Decorator to trace agent execution""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): print(f"\n[{agent_name}] STARTED - Input: {kwargs}") start_time = time.time() try: result = func(*args, **kwargs) if isinstance(result, dspy.Prediction): print_react_trajectory(result) print("\n Final Answer:") for key, value in result.items(): if key not in ['trajectory', 'reasoning']: print(f" {key}: {value}") print(f"\n[{agent_name}] FINISHED - {time.time() - start_time:.2f}s\n") return result except Exception as e: print(f"\n[{agent_name}] FAILED - {str(e)} ({time.time() - start_time:.2f}s)\n") raise return wrapper return decorator

[MathAgent] STARTED Input: {'math_query': 'What is 5+3?'} Thinking: I need to add two numbers Action: add_numbers({'a': 5, 'b': 3}) Result: 8 Final Answer: math_result: 8 Completed in 1.23s [MathAgent] FINISHED

def add_numbers(a: float, b: float) -> float: """Add two numbers together and return the sum.""" return a + b def multiply_numbers(a: float, b: float) -> float: """Multiply two numbers together and return the product.""" return a * b math_tools = [add_numbers, multiply_numbers

class MathAgentSignature(dspy.Signature): """Signature for mathematical operations agent""" math_query: str = dspy.InputField(desc="A mathematical question or operation request") math_result: str = dspy.OutputField(desc="The result of the mathematical operation") class MathAgent(dspy.Module): """Agent that handles mathematical operations using ReAct.""" def __init__(self): super().__init__() self.react_program = dspy.ReAct( signature=MathAgentSignature, tools=math_tools, max_iters=3 ) @trace_agent("MathAgent") def forward(self, math_query: str) -> dspy.Prediction: """Process mathematical queries and return results""" return self.react_program(math_query=math_query)

### Text Agent def count_words(text: str) -> int: """Count the number of words in the given text.""" return len(text.split()) def reverse_text(text: str) -> str: """Reverse the given text and return it backwards.""" return text[::-1] text_tools = [count_words, reverse_text] class TextAgentSignature(dspy.Signature): """Signature for text processing operations agent""" text_query: str = dspy.InputField(desc="A text processing question or operation request") text_result: str = dspy.OutputField(desc="The result of the text processing operation") class TextAgent(dspy.Module): """Agent that handles text operations using ReAct.""" def __init__(self): super().__init__() self.react_program = dspy.ReAct( signature=TextAgentSignature, tools=text_tools, max_iters=3 ) @trace_agent("TextAgent") def forward(self, text_query: str) -> dspy.Prediction: """Process text queries and return results""" return self.react_program(text_query=text_query)

### Time Agent - Real-World API Integration def get_usa_time() -> str: """Get the current time in USA (Eastern Time).""" usa_tz = pytz.timezone('America/New_York') usa_time = datetime.now(usa_tz) return usa_time.strftime("%Y-%m-%d %H:%M:%S %Z") def get_china_time() -> str: """Get the current time in China (Beijing Time).""" china_tz = pytz.timezone('Asia/Shanghai') china_time = datetime.now(china_tz) return china_time.strftime("%Y-%m-%d %H:%M:%S %Z") time_tools = [get_usa_time, get_china_time] class TimeAgentSignature(dspy.Signature): """Find the timezone or location in the input query and return the corresponding time.""" time_query: str = dspy.InputField(desc="A time-related question or timezone request") time_result: str = dspy.OutputField(desc="The current time in the requested timezone") class TimeAgent(dspy.Module): def __init__(self): super().__init__() self.react_program = dspy.ReAct( signature=TimeAgentSignature, tools=time_tools, max_iters=3 ) @trace_agent("TimeAgent") def forward(self, time_query: str) -> dspy.Prediction: return self.react_program(time_query=time_query)

### Weather Agent - Complex External APIs def get_weather_by_city(city_name: str) -> str: """Get current weather information for a given city using Open-Meteo API.""" try: # Geocoding to get coordinates geocoding_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city_name}&count=1" geo_response = requests.get(geocoding_url, timeout=10) geo_data = geo_response.json() if not geo_data.get('results'): return f"City '{city_name}' not found" location = geo_data['results'][0] lat, lon = location['latitude'], location['longitude'] # Get weather data weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,wind_speed_10m" weather_response = requests.get(weather_url, timeout=10) weather_data = weather_response.json() current = weather_data['current'] return f"Weather in {location['name']}, {location.get('country')}: Temperature: {current['temperature_2m']}°C, Humidity: {current['relative_humidity_2m']}%, Wind Speed: {current['wind_speed_10m']} km/h" except Exception as e: return f"Error fetching weather: {str(e)}" def compare_city_temperatures(city1: str, city2: str) -> str: """Compare temperatures between two cities.""" try: temps = {} for city in [city1, city2]: # Get coordinates and weather data geocoding_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1" geo_response = requests.get(geocoding_url, timeout=10) geo_data = geo_response.json() location = geo_data['results'][0] weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={location['latitude']}&longitude={location['longitude']}&current=temperature_2m" weather_response = requests.get(weather_url, timeout=10) temps[city] = weather_response.json()['current']['temperature_2m'] diff = abs(temps[city1] - temps[city2]) warmer = city1 if temps[city1] > temps[city2] else city2 return f"{city1}: {temps[city1]}°C, {city2}: {temps[city2]}°C. {warmer} is warmer by {diff}°C" except Exception as e: return f"Error comparing temperatures: {str(e)}" weather_tools = [get_weather_by_city, compare_city_temperatures] class WeatherAgentSignature(dspy.Signature): """Find weather and temperature information and structure into the requested output form.""" weather_query: str = dspy.InputField(desc="A weather-related question about cities or temperature comparison") weather_result: str = dspy.OutputField(desc="Weather information or temperature comparison results") class WeatherAgent(dspy.Module): def __init__(self): super().__init__() self.react_program = dspy.ReAct( signature=WeatherAgentSignature, tools=weather_tools, max_iters=3 ) @trace_agent("WeatherAgent") def forward(self, weather_query: str) -> dspy.Prediction: return self.react_program(weather_query=weather_query)

### The Magic - Wrapping Subagents as Tools def math_calculator(math_query: str) -> str: """ Math Calculator Agent: Performs mathematical operations - **Addition**: Adds two numbers together - **Multiplication**: Multiplies two numbers together - **Smart Operation Selection**: ReAct decides which tool to use - Output key: "math_result" """ math_agent = MathAgent() prediction = math_agent(math_query=math_query) return prediction.math_result def text_processor(text_query: str) -> str: """ Text Processor Agent: Performs text operations - **Word Counter**: Counts words in text - **Text Reverser**: Reverses text backwards - **Smart Operation Selection**: ReAct decides which tool to use - Output key: "text_result" """ text_agent = TextAgent() prediction = text_agent(text_query=text_query) return prediction.text_result def time_checker(time_query: str) -> str: """ Time Checker Agent: Provides current time information - **USA Time**: Returns current time in USA (Eastern Time) - **China Time**: Returns current time in China (Beijing Time) - **Smart Timezone Selection**: ReAct decides which tool to use - Output key: "time_result" """ time_agent = TimeAgent() prediction = time_agent(time_query=time_query) return prediction.time_result def weather_checker(weather_query: str) -> str: """ Weather Checker Agent: Provides real-time weather information - **City Weather**: Gets current weather for any city worldwide - **Temperature Comparison**: Compares temperatures between two cities - **Smart Tool Selection**: ReAct decides which weather tool to use - Output key: "weather_result" """ weather_agent = WeatherAgent() prediction = weather_agent(weather_query=weather_query) return prediction.weather_result

### Main Agent - The Intelligent Coordinator class MainAgentSignature(dspy.Signature): """You are an intelligent coordinator that breaks down complex queries into sub-tasks. Your responsibilities: 1. Analyze the user's query and identify distinct tasks (math, text, time, weather) 2. Break multi-part queries into separate sub-queries for each specialist agent 3. Call the appropriate agent tools in sequence to gather all needed information 4. Synthesize results from multiple agents into a coherent final answer Available agents and their capabilities: - math_calculator: Mathematical operations (addition, multiplication) - text_processor: Text operations (word count, text reversal) - time_checker: Time queries (USA time, China time) - weather_checker: Weather queries (city weather, temperature comparisons) For multi-part queries like "What's 5+3 and weather in Cairo?", call math_calculator first, then weather_checker, and combine both results in your final answer. """ user_query: str = dspy.InputField(desc="User's potentially multi-part query requiring one or more specialist agents") final_answer: str = dspy.OutputField(desc="Complete answer combining results from all relevant agents") class MainAgent(dspy.Module): """Main coordinator agent that intelligently delegates to specialist subagents. This agent can: - Handle single queries: "What's the time in USA?" → routes to time_checker - Handle multi-part queries: "What's 5+3 and weather in Cairo?" → routes to both math_calculator and weather_checker - Coordinate multiple agent calls in sequence - Synthesize results from multiple agents into coherent answers """ def __init__(self): super().__init__() # Configure ReAct with all specialist agents as tools self.root_program = dspy.ReAct( signature=MainAgentSignature, tools=[math_calculator, text_processor, time_checker, weather_checker], max_iters=5 # Allow multiple iterations for multi-part queries ) def forward(self, user_query: str): results = self.root_program(user_query=user_query) return results

### Configuration and Execution # Configure the language model dspy.configure(lm=dspy.LM('ollama_chat/qwen3:4b', api_base='http://localhost:11434', api_key='')) # Initialize the root agent root_agent = MainAgent() # Test with a multi-part query if __name__ == "__main__": test_query = "Can you tell me the weather in Berlin and the current time in China?" result = root_agent(user_query=test_query) print(result.final_answer)

[WeatherAgent] STARTED Input: {'weather_query': 'Berlin'} Thinking: You need to retrieve the current weather information for Berlin using the get_weather_by_city tool. Action: get_weather_by_city({'city_name': 'Berlin'}) Result: Weather in Berlin, Germany: Temperature: 11.9°C, Humidity: 62%, Wind Speed: 15.5 km/h Thinking: The weather information for Berlin has been retrieved and contains the necessary details. Action: finish({}) Result: Completed. Final Answer: weather_result: Weather in Berlin, Germany: Temperature: 11.9°C, Humidity: 62%, Wind Speed: 15.5 km/h Completed in 19.89s [WeatherAgent] FINISHED [TimeAgent] STARTED Input: {'time_query': 'China'} Thinking: Choosing to get the current time in China (Beijing Time) since the query is about China. Action: get_china_time({}) Result: 2025-11-03 19:53:50 CST Thinking: The task is complete as the time for China (Beijing Time) has been retrieved. Action: finish({}) Result: Completed. Final Answer: time_result: 2025-11-03 19:53:50 CST Completed in 18.97s [TimeAgent] FINISHED The weather in Berlin is 11.9°C with 62% humidity and 15.5 km/h wind. The current time in China is 2025-11-03 19:53:50 CST.