How to monitor multi-agent systems in production

Silent handoff loops and uneven agent load make incidents hard to debug once you scale past a single prompt. To monitor multi-agent systems in production, you need correlation across agent boundaries, structured event emission, and per-step health signals that surface before users notice degradation. Treat each agent role as a microservice and instrument it with the same rigor you’d apply to a distributed backend.

Step 1: Assign a correlation ID to every top-level task

A multi-agent run is a tree of decisions, not a single call. Without a stable trace ID propagated through every agent and tool, you’re left grepping logs by timestamp. Create a context variable at the entrypoint and pass it explicitly in messages or via context propagation.

import uuid, contextvars

trace_id_ctx = contextvars.ContextVar("trace_id")

def start_trace() -> str:
    tid = uuid.uuid4().hex
    trace_id_ctx.set(tid)
    return tid

def current_trace() -> str | None:
    return trace_id_ctx.get(None)

If agents communicate over a queue, serialize current_trace() into the message envelope. If they call each other in-process, set the context var at the top of each agent’s run() method using the incoming ID. This takes 20 minutes and pays back the first time a supervisor agent silently drops a subtask.

Step 2: Emit structured transition events

Plain text logs are useless for aggregation. Define a minimal event schema and emit one JSON line per agent action: handoff, LLM call, tool use, or error. Keep the schema stable; add optional meta for variable data.

import json, time, logging

logger = logging.getLogger("agent.events")

def emit_transition(*, agent: str, action: str, parent: str | None = None, meta: dict | None = None):
    event = {
        "ts": time.time_ns(),
        "trace_id": current_trace(),
        "agent": agent,
        "action": action,
        "parent": parent,
        "meta": meta or {},
    }
    logger.info(json.dumps(event))

What to include in meta

• For LLM steps: model name, prompt/completion tokens, latency.
• For tool steps: status, latency, error type.
• For handoffs: source and target agent, queue depth if applicable.

A handoff from planner to researcher should look like:

{"ts":1710000000000000,"trace_id":"abc","agent":"planner","action":"handoff","parent":null,"meta":{"to":"researcher"}}

Step 3: Instrument LLM calls with token and latency metrics

Agents burn tokens unpredictably when prompts drift. Wrap your model client so every completion logs usage and latency against the active trace. If you route agent traffic through a gateway such as n4n.ai, you get per-token usage metering and automatic fallback when a provider is rate-limited, which removes the need to write your own retry and accounting layer.

from openai import OpenAI

client = OpenAI()  # or point base_url at your gateway

def llm_step(agent: str, messages: list):
    start = time.time_ns()
    resp = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
    latency_ms = (time.time_ns() - start) / 1e6
    emit_transition(
        agent=agent,
        action="llm",
        meta={
            "model": resp.model,
            "prompt_tokens": resp.usage.prompt_tokens,
            "completion_tokens": resp.usage.completion_tokens,
            "latency_ms": round(latency_ms, 1),
        },
    )
    return resp.choices[0].message.content

Track prompt vs completion token ratio per agent. A researcher agent that suddenly emits 3x completion tokens is either looping or hitting a degenerate prompt.

Step 4: Wrap tool invocations with status and timing

Tools are where multi-agent systems actually fail: a stale API credential, a malformed argument, a timeout. Wrap every tool with a decorator that records success, error, and latency.

def instrument_tool(name: str):
    def deco(fn):
        def wrapped(*args, **kwargs):
            start = time.time_ns()
            try:
                result = fn(*args, **kwargs)
                status = "ok"
            except Exception as e:
                result = str(e)
                status = "error"
            latency_ms = (time.time_ns() - start) / 1e6
            emit_transition(
                agent="tool",
                action=name,
                meta={"status": status, "latency_ms": round(latency_ms, 1), "error": result if status == "error" else None},
            )
            if status == "error":
                raise
            return result
        return wrapped
    return deco

Apply it directly:

@instrument_tool("sql_query")
def sql_query(q: str):
    # real implementation
    ...

Now a broken database connection shows up as tool/sql_query/error in your event stream instead of a generic 500.

Step 5: Export metrics to a time-series backend

Events are for forensics; metrics are for alerts. Use Prometheus or any StatsD-compatible sink to count steps and record latencies. This lets you plot “researcher LLM p95 latency” or “tool error rate by agent” on a dashboard.

from prometheus_client import Counter, Histogram, start_http_server

STEP_COUNT = Counter("agent_step_total", "Agent steps", ["agent", "action", "status"])
STEP_LATENCY = Histogram("agent_step_latency_ms", "Step latency", ["agent", "action"])

def record_metric(agent: str, action: str, status: str, latency_ms: float):
    STEP_COUNT.labels(agent, action, status).inc()
    STEP_LATENCY.labels(agent, action).observe(latency_ms)

Call record_metric inside emit_transition or as a side effect in the wrappers above. Start the exporter on a port and scrape it with Grafana. Set alerts on:

• rate(agent_step_total{status="error"}[5m]) above a floor.
• Latency histogram p95 for llm actions spiking 2x baseline.
• Token count per trace exceeding a hard budget.

Step 6: Detect agent loops and stalls

The classic multi-agent bug is two agents ping-ponging forever. Monitor multi-agent systems in production by counting transitions per trace ID in a rolling window. If a trace exceeds a sane limit (say 25 steps), abort and page.

from collections import defaultdict

_trace_steps = defaultdict(int)
MAX_STEPS = 25

def tick_trace(trace_id: str) -> bool:
    _trace_steps[trace_id] += 1
    if _trace_steps[trace_id] > MAX_STEPS:
        emit_transition(agent="supervisor", action="loop_abort", meta={"steps": _trace_steps[trace_id]})
        return False
    return True

Call tick_trace(current_trace()) at the top of every agent loop iteration. For stalls, track last-event timestamp per trace; a trace with no event for 120s while not in a terminal state is a stall. A simple background sweeper can emit supervisor/stall and kill the run.

Step 7: Build a trace reconstruction view

Your dashboard should let you paste a trace ID and see the full agent tree: which agent ran, what it called, token cost, and where it died. Store events in a searchable backend (Elasticsearch, ClickHouse, or even Postgres with jsonb). The query is trivial:

SELECT ts, agent, action, meta
FROM agent_events
WHERE trace_id = 'abc'
ORDER BY ts ASC;

If you can’t answer “why did task X cost $0.40 and take 90s” within two minutes, your monitoring is incomplete.

Verify success

You’ve correctly instrumented the system when:

1. Every production incident can be traced by a single ID from entrypoint to failure.
2. A Grafana panel shows per-agent error rate, p95 latency, and token spend.
3. An automated alert fired on a simulated loop (run two agents that intentionally bounce a message 30 times; confirm abort and page).
4. A new engineer can reconstruct a failed multi-agent task from raw events without reading application code.

Monitoring multi-agent systems in production is not about fancy dashboards; it’s about making invisible handoffs visible. Ship the correlation ID first, then events, then metrics. The loop detector is what keeps your bill and your pager quiet.