Multi-agent systems involve multiple AI agents collaborating on a task. Traccia makes it easy to trace these complex workflows while maintaining clear visibility into each agent's contribution, costs, and performance.

Common Multi-Agent Patterns

Orchestrator Pattern

A central orchestrator agent coordinates multiple specialized worker agents. Each worker handles specific tasks.

Pipeline Pattern

Agents are chained in sequence, with each agent processing the output of the previous one.

Collaborative Pattern

Multiple agents work on different aspects of a problem simultaneously, then combine results.

Hierarchical Pattern

Agents organized in layers, with higher-level agents delegating to lower-level specialists.

Example: Orchestrator Pattern

Here's how to implement an orchestrator that delegates to specialized agents while maintaining proper trace hierarchy and agent attribution:

multi_agent.py

python

from traccia import init, observe, runtime_config
from openai import OpenAI

# Initialize Traccia once
init(
    session_id="workflow-001",  # Shared session ID
    project_id="customer-support"
)

client = OpenAI()

@observe()
def orchestrator_agent(customer_query: str):
    """Main orchestrator that routes queries to specialized agents."""
    
    # Set orchestrator agent ID
    runtime_config.set_agent_id("orchestrator")
    
    # Classify the query
    classification = classify_query(customer_query)
    
    # Route to appropriate specialist
    if classification == "billing":
        return billing_agent(customer_query)
    elif classification == "technical":
        return technical_agent(customer_query)
    else:
        return general_agent(customer_query)

@observe()
def classify_query(query: str) -> str:
    """Classify customer query."""
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "Classify queries as: billing, technical, or general"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content.strip().lower()

@observe()
def billing_agent(query: str) -> str:
    """Handle billing-related queries."""
    
    # Change agent ID for this subtree
    runtime_config.set_agent_id("billing-agent")
    
    # Fetch customer data
    customer_data = fetch_billing_data()
    
    # Generate response
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"You are a billing specialist. Context: {customer_data}"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

@observe()
def technical_agent(query: str) -> str:
    """Handle technical support queries."""
    
    runtime_config.set_agent_id("technical-agent")
    
    # Search knowledge base
    kb_results = search_knowledge_base(query)
    
    # Generate response
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"You are a technical support agent. KB: {kb_results}"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

@observe(as_type="tool")
def fetch_billing_data():
    # Simulate database query
    return {"account_status": "active", "balance": 150.00}

@observe(as_type="tool")
def search_knowledge_base(query: str):
    # Simulate KB search
    return ["Article 1", "Article 2"]

@observe()
def general_agent(query: str) -> str:
    """Handle general queries."""
    runtime_config.set_agent_id("general-agent")
    
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

Mark host services as orchestrators

When this orchestrator pattern lives inside a long-running service (for example a demo portal or multi-agent API), initialize the SDK with service_role="orchestrator". This tells Traccia to never register the host service itself as an agent and to attribute traces and metrics only to the logical agents you tag with agent_id /runtime_config.run_identity().

Adding Governance & Fallbacks

In a multi-agent system, some agents may handle sensitive tasks that require guardrails. You can wrap specific agent calls with a guardrail_span to track policy evaluations and implement fallbacks:

governance.py

python

from traccia.governance import guardrail_span
from traccia.tracer.span import SpanStatus

@observe()
def billing_agent(query: str) -> str:
    runtime_config.set_agent_id("billing-agent")
    
    try:
        # Track guardrail evaluation for this sensitive operation
        with guardrail_span("check_billing_policy", policy_id="pol_billing_01", action="block") as gs:
            if not check_user_authorization(query):
                gs.set_status(SpanStatus.ERROR, "Unauthorized access attempt")
                raise PolicyViolationError("User not authorized for billing queries")
                
        # If guardrail passes, proceed with LLM call
        response = client.chat.completions.create(...)
        return response.choices[0].message.content
        
    except PolicyViolationError:
        # Fallback to human routing on policy violation
        runtime_config.set_agent_id("orchestrator")
        return "I need to transfer you to a human agent to handle this billing request."

Resulting Trace Hierarchy

The trace will show the complete workflow with agent attribution:

Trace: orchestrator_agent(session: workflow-001)

classify_query[orchestrator]0.5sGPT-3.5, $0.001

billing_agent[billing-agent]2.1s

fetch_billing_data[billing-agent]0.05s

openai.chat.completions[billing-agent]2.0sGPT-4, $0.024

Total cost: $0.025, Total duration: 2.6s

Agent attribution

Each span is tagged with the agent.id that was active when it was created, allowing you to track costs and performance per agent.

Session-Based Grouping

Use session IDs to group related traces across multiple agent invocations:

session_tracking.py

python

from traccia import init, observe, runtime_config

# Initialize with shared session
init()

def handle_conversation(conversation_id: str):
    """Handle a multi-turn conversation."""
    
    # Set session ID for the entire conversation
    runtime_config.set_session_id(f"conv-{conversation_id}")
    runtime_config.set_user_id("user-12345")
    
    # Each message becomes a separate trace with the same session_id
    while True:
        user_message = get_user_input()
        if user_message == "quit":
            break
        
        # This creates a new trace, but with the same session_id
        response = conversation_agent(user_message)
        send_response(response)

@observe()
def conversation_agent(message: str) -> str:
    """Process a single message in the conversation."""
    # All traces share the same session_id
    # You can query by session_id to see the full conversation flow
    return generate_response(message)

In the Traccia Platform, you can filter by session.id to see all traces from a single conversation or workflow.

Distributed Multi-Agent Systems (HTTP)

When agents run in separate services and communicate via HTTP, use context propagation to maintain trace continuity:

distributed.py

python

# Service 1: Orchestrator (with auto-instrumentation)
from traccia import init, observe
import requests

init(enable_patching=True)  # Auto-instruments requests

@observe()
def orchestrator(task: str):
    # Trace context automatically injected into headers
    response = requests.post(
        "http://agent-service:8000/process",
        json={"task": task}
    )
    return response.json()

# Service 2: Worker Agent (FastAPI with auto-instrumentation)
from fastapi import FastAPI
from traccia import init, observe
from traccia.instrumentation import install_http_middleware

app = FastAPI()
init(enable_patching=True)

# Install middleware for automatic context extraction
install_http_middleware(app)

@app.post("/process")
@observe()
def process_task(task: dict):
    # Trace context automatically extracted from headers
    # This span becomes a child of the orchestrator's span
    result = do_work(task)
    return {"result": result}

Automatic propagation

With enable_patching=True, trace context is automatically propagated across HTTP calls. No manual header management needed!

Best Practices

1. Use Consistent Session IDs

Set the same session_id for all agents involved in a single workflow to make it easy to trace the entire flow.

2. Set Unique Agent IDs

Give each agent a unique agent_id to track costs and performance independently.

3. Use Descriptive Span Names

Name spans after the agent's role (e.g., "triage-agent", "billing-specialist") rather than generic names like "process" or "handler".

4. Enable Context Propagation

For distributed systems, ensure enable_patching=True so trace context flows across HTTP boundaries.

Example: Pipeline Pattern

A sequential pipeline where each agent processes data and passes it to the next:

pipeline.py

python

from traccia import init, observe, runtime_config

init(session_id="pipeline-abc123")

@observe()
def data_pipeline(raw_data: str):
    """Main pipeline orchestrator."""
    runtime_config.set_agent_id("pipeline-orchestrator")
    
    # Stage 1: Extract
    extracted = extraction_agent(raw_data)
    
    # Stage 2: Transform
    transformed = transformation_agent(extracted)
    
    # Stage 3: Load
    result = loading_agent(transformed)
    
    return result

@observe()
def extraction_agent(data: str):
    """Extract structured data from raw text."""
    runtime_config.set_agent_id("extraction-agent")
    # ... extraction logic
    return extracted_data

@observe()
def transformation_agent(data: dict):
    """Transform and enrich data."""
    runtime_config.set_agent_id("transformation-agent")
    # ... transformation logic
    return transformed_data

@observe()
def loading_agent(data: dict):
    """Load data into destination."""
    runtime_config.set_agent_id("loading-agent")
    # ... loading logic
    return {"status": "success"}

Viewing Multi-Agent Traces

In Jaeger/Grafana

The trace waterfall will show all agents and their timing:

Each agent appears as a distinct span with its agent.id in attributes
Parent-child relationships show the execution flow
Expand spans to see LLM calls, token counts, and costs
Filter by service, agent ID, or session ID

In Traccia Platform

The platform provides agent-centric views:

See costs broken down by agent
Filter traces by session ID to see complete workflows
Compare performance across different agents
Set per-agent policies and budgets

Next Steps

Understanding Agents

Learn about agent metadata and identity

Advanced SDK Topics

Context propagation and manual spans

Cost Attribution

Track costs across multiple agents

Multi-Agent Tracing

Common Multi-Agent Patterns

Orchestrator Pattern

Pipeline Pattern

Collaborative Pattern

Hierarchical Pattern

Example: Orchestrator Pattern

Adding Governance & Fallbacks

Resulting Trace Hierarchy

Session-Based Grouping

Distributed Multi-Agent Systems (HTTP)

Best Practices

1. Use Consistent Session IDs

2. Set Unique Agent IDs

3. Use Descriptive Span Names

4. Enable Context Propagation

Example: Pipeline Pattern

Viewing Multi-Agent Traces

In Jaeger/Grafana

In Traccia Platform

Next Steps

Understanding Agents

Advanced SDK Topics

Cost Attribution