Debugging why one agent's output silently breaks another

Debugging agent output failures in a multi-agent system starts with accepting that the bug is almost never in the model’s reasoning—it’s in the data contract between two processes you assumed were compatible. When Agent A returns a dict with a missing key, Agent B doesn’t crash at the handoff; it crashes 400 tokens later when it tries to use that key, and the stack trace points to the wrong place. You need to treat inter-agent messages as a network protocol, not a function call.

Step 1: Reproduce the silent break with full payload logging

The first move is to stop trusting your in-memory objects. Serialize the raw completion from every agent to disk before any parsing happens. If you are on the OpenAI SDK, wrap the client call so you never lose the original message.content string.

import json, time, os
from openai import OpenAI

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

def log_agent_call(name, messages, **kwargs):
    resp = client.chat.completions.create(model="gpt-4o-mini", messages=messages, **kwargs)
    raw = resp.model_dump()
    os.makedirs("traces", exist_ok=True)
    with open(f"traces/{name}_{int(time.time()*1000)}.json", "w") as f:
        json.dump(raw, f, indent=2)
    return resp

Run the pipeline once against the failing input. You now have a timestamped artifact for each agent step. Do not skip this even if you think the error is obvious—silent breaks usually mean your mental model of the output is wrong.

If you run async agents, the wrapper must be a coroutine and log before the await returns the parsed object. Streaming complicates this: buffer the deltas in a list, join them after the stream closes, then write the reconstructed string. A partial trace is worse than none because it hides truncation.

Step 2: Diff the expected shape against the actual shape

Open the trace file from Agent A. Look at choices[0].message.content. In a healthy system this is either pure JSON or a stable text template. In a broken one you will see markdown fences, trailing commentary, or a key renamed from steps to action_items.

Stripping markdown contamination

Models routinely wrap JSON in ```json blocks. Write a strict extractor and use it everywhere:

import re, json

def extract_json(text):
    match = re.search(r"```(?:json)?\s*(.*?)```", text, re.DOTALL)
    if match:
        return json.loads(match.group(1).strip())
    # fallback: try the whole string
    return json.loads(text.strip())

Run this on the saved trace. If it raises, you have found the first contract violation. Note the exact failure mode: JSONDecodeError at position 142 means a truncated response, not a schema drift. Another trap is type coercion—the model returns "confidence": "0.9" (string) instead of a float. Standard json.loads accepts it, but downstream strict validators may not.

Step 3: Enforce a strict schema at every boundary

Ad-hoc dict["key"] access is how silent breaks propagate. Define a pydantic model for each agent’s output and parse at the edge.

from pydantic import BaseModel, ValidationError, Field, ConfigDict

class AgentAOutput(BaseModel):
    model_config = ConfigDict(strict=True)
    task_id: str
    steps: list[str] = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0)

class ContractViolation(Exception):
    pass

def parse_agent_a(raw_text: str) -> AgentAOutput:
    try:
        data = extract_json(raw_text)
        return AgentAOutput(**data)
    except (json.JSONDecodeError, ValidationError) as e:
        raise ContractViolation(f"Agent A contract broken: {e}") from e

Now Agent B receives a typed object, not a hope. If Agent A omits steps, the error is raised at the handoff, with a message naming the culprit. This single change removes most cross-agent ghosts.

Version your schemas. When you change AgentAOutput, bump a schema_version field and reject mismatches explicitly. A gateway that honors client routing directives can tag requests with version headers, but you should not rely on the model to self-report. Strict mode forces strings-to-float mismatches to surface immediately instead of leaking into Agent B’s math.

Step 4: Trace which model and provider processed each step

Silent breaks often appear after a fallback. If your orchestrator uses automatic provider failover, Agent A might have been served by a different backend mid-incident, and that backend formats JSON slightly differently.

If you route through a gateway like n4n.ai, the per-token usage metering and forwarded cache-control hints let you correlate which backend handled each agent without adding custom instrumentation to every call. The forwarded cache-control hints mean a cached response from provider X may carry a different serialization style than a fresh one from provider Y. Otherwise, inject a trace_id into the user field or metadata:

import uuid

trace_id = uuid.uuid4().hex
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    user=trace_id,  # appears in usage logs on most gateways
)

Grep your traces for that trace_id and confirm the model field and any provider extension in the response. A mismatch between expected and actual provider is a leading cause of formatting drift that passes local tests but fails in production.

Step 5: Isolate the failing payload in a unit test

Take the raw content string from the broken trace and write a pytest that feeds it through parse_agent_a. This converts a flaky production incident into a deterministic red test.

def test_agent_a_contract_from_trace():
    with open("traces/agent_a_1710000000000.json") as f:
        raw = json.load(f)
    text = raw["choices"][0]["message"]["content"]
    # Before fix: this raises ContractViolation
    out = parse_agent_a(text)
    assert out.confidence <= 1.0
    assert len(out.steps) > 0

Run it. It should fail. Now you have a tight loop: edit the extractor or schema, re-run, repeat. No need to spin up the whole pipeline.

Add a second test that replays Agent A’s output into Agent B’s entry function to prove the downstream consumer survives the corrected contract. Store these traces as golden files in tests/fixtures/ so future prompt changes can be diffed against known-good handoffs.

Step 6: Make the handoff fail loud, not silent

Even with schemas, Agent B can misuse a valid object. Add explicit guards at the start of each consumer.

def run_agent_b(a_output: AgentAOutput):
    if a_output.confidence < 0.2:
        raise ContractViolation(f"Refusing low-confidence handoff: {a_output.confidence}")
    # build messages for Agent B using only declared fields
    messages = [{"role": "system", "content": "Execute steps."},
                {"role": "user", "content": "\n".join(a_output.steps)}]
    return log_agent_call("agent_b", messages)

The goal is that any break names the agent and the field. A KeyError inside a 200-line prompt builder is not acceptable. Wrap the whole multi-agent run in a supervisor that catches ContractViolation and emits a structured error with the trace_id. Consider a circuit breaker: if Agent A violates contract three times in a row, halt the workflow instead of letting Agent B spin on garbage.

Step 7: Verify success with a contract test suite

Verification is not “it ran once.” Build a small corpus of saved traces covering happy path, markdown-wrapped, truncated, and schema-drift cases. Run them in CI on every change to agent prompts or schemas.

pytest tests/contract/ --tb=short

A green suite means debugging agent output failures is now a solved class of bug for your team: every handoff is validated, every raw payload is reproducible, and every provider switch is visible. Re-run the full pipeline with the original failing input and confirm the supervisor logs a clean trace_id with no ContractViolation.

If you still see a silent break, return to Step 1 with the new trace—usually it’s a handoff you missed because an agent called a sub-agent directly instead of through the logged wrapper. Close that gap and the system becomes debuggable end to end. Add alerting on ContractViolation rate so a future drift shows up in metrics before users report garbled output.