multi-agent-coordination

Multi-Agent Coordination Instrumentation

Instrument multi-agent systems to trace coordination, handoffs, and hierarchies.

Core Principle

Multi-agent traces must answer:

1. Which agent started the workflow?
2. Which agents were involved?
3. How did work flow between agents?
4. Why did handoffs occur?
5. What was the hierarchy (parent-child)?

Trace Hierarchy

Session/Conversation (root span)
└── Supervisor Agent Run
    ├── Planning Phase (span)
    │   └── LLM Call (span)
    ├── Delegate to Agent A (span)
    │   └── Agent A Run (child trace)
    │       ├── LLM Call
    │       └── Tool Call
    ├── Delegate to Agent B (span)
    │   └── Agent B Run (child trace)
    │       └── LLM Call
    └── Synthesis Phase (span)
        └── LLM Call

Essential Span Attributes

Agent Identity

span.set_attribute("agent.name", "researcher")
span.set_attribute("agent.type", "worker")  # supervisor, worker, critic
span.set_attribute("agent.run_id", str(uuid4()))
span.set_attribute("agent.framework", "langgraph")

Parent-Child Linking

# Parent agent creates child context
child_context = create_child_context(current_span)
span.set_attribute("agent.parent_id", parent_run_id)
span.set_attribute("agent.parent_name", "supervisor")

# Pass context to child agent
child_agent.run(input, trace_context=child_context)

Handoff Tracking

# Log handoff decision
span.set_attribute("handoff.from_agent", "supervisor")
span.set_attribute("handoff.to_agent", "researcher")
span.set_attribute("handoff.reason", "needs_web_search")
span.set_attribute("handoff.task_summary", "Find pricing data")

Workflow State

span.set_attribute("workflow.step", "research")
span.set_attribute("workflow.total_steps", 4)
span.set_attribute("workflow.agents_involved", 3)
span.set_attribute("workflow.state", "in_progress")

Framework Patterns

LangGraph

from langgraph.graph import StateGraph
from langfuse.decorators import observe

@observe(name="agent.supervisor")
def supervisor_node(state):
    # Supervisor logic
    span = get_current_span()
    span.set_attribute("agent.name", "supervisor")
    span.set_attribute("agent.decision", state["next_agent"])
    return state

@observe(name="agent.worker")
def worker_node(state):
    span = get_current_span()
    span.set_attribute("agent.name", "worker")
    span.set_attribute("agent.parent_name", "supervisor")
    return state

graph = StateGraph()
graph.add_node("supervisor", supervisor_node)
graph.add_node("worker", worker_node)