Agentic pipelines usually fail where logs are verbose and traces are long. I built a small diagnostic agent that ingests execution traces from broken agent runs and returns structured remediation plans. Because we will run it on Oxlo.ai, we can pass the full trace into every request without watching the token meter spin.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Initialize the Oxlo.ai client
I start with the standard OpenAI SDK pointed at Oxlo.ai. This is a literal drop-in replacement, so no custom adapters are needed.
from openai import OpenAI
import json
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY"
)
# Quick connectivity check
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[{"role": "user", "content": "ping"}],
)
print(response.choices[0].message.content)Step 2: Define the troubleshooting agent's system prompt
The agent must output valid JSON and avoid generic advice. I lock the schema in the system prompt so the model stays focused on root cause, immediate fix, and prevention.
SYSTEM_PROMPT = """You are a site-reliability agent that diagnoses broken agentic workloads.
Analyze the provided execution trace and return a single JSON object with exactly these keys:
- root_cause: one sentence describing why the workload failed
- fix: concrete shell or code commands to resolve the issue
- prevention: one configuration or code change to stop recurrence
Rules:
1. Do not guess. Base every conclusion on evidence in the trace.
2. If a tool loops, identify the loop condition.
3. If a schema is wrong, show the corrected payload.
4. Output ONLY the JSON object, no markdown fences."""Step 3: Build a synthetic trace generator
To test locally I wrote a helper that simulates three common agentic pathologies: looping tool calls, hallucinated parameters, and unhandled exceptions after a timeout.
def generate_loop_trace():
trace = {
"agent_id": "inventory-bot-v2",
"start_time": "2024-05-21T14:02:00Z",
"steps": []
}
# Simulate a loop: same tool called 4 times with identical args
for i in range(4):
trace["steps"].append({
"step": i + 1,
"tool": "check_stock",
"input": {"sku": "ABC-123", "warehouse": "east"},
"output": {"stock": 0},
"status": "success"
})
# Final crash
trace["steps"].append({
"step": 5,
"tool": "place_order",
"input": {"sku": "ABC-123", "qty": 500},
"output": {"error": "Supplier API timeout after 30s"},
"status": "failure"
})
return json.dumps(trace, indent=2)
def generate_schema_trace():
trace = {
"agent_id": "crm-sync-v1",
"steps": [{
"step": 1,
"tool": "create_lead",
"input": {"name": "Alice", "email": "alice@example.com", "budget": "high"},
"output": {"error": "ValidationError: field 'budget' expects type decimal, received string"},
"status": "failure"
}]
}
return json.dumps(trace, indent=2)Step 4: Send traces to the model and parse structured output
This is the core diagnostic loop. I pass the full trace as the user message and enable JSON mode. On Oxlo.ai this costs the same flat rate whether the trace is two hundred tokens or twenty thousand, which matters when you are debugging long context windows or multi-turn agent logs.
def diagnose(trace_json: str, model: str = "kimi-k2.6") -> dict:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Execution trace:\n{trace_json}"},
],
response_format={"type": "json_object"},
temperature=0.1,
)
raw = response.choices[0].message.content
return json.loads(raw)
# Test on the loop trace
loop_trace = generate_loop_trace()
result = diagnose(loop_trace)
print(json.dumps(result, indent=2))Run it
Putting it all together, I run both failure modes and print the diagnoses. The loop trace triggers a loop-detection fix, while the schema trace triggers a type-correction fix.
if __name__ == "__main__":
cases = [
("loop", generate_loop_trace()),
("schema", generate_schema_trace()),
]
for label, trace in cases:
print(f"\n=== {label} ===")
report = diagnose(trace)
print(json.dumps(report, indent=2))Example output:
=== loop ===
{
"root_cause": "The agent looped on check_stock because stock remained 0 and there was no exit condition before attempting place_order.",
"fix": "Add a guard clause: if stock == 0, skip place_order and alert procurement instead of retrying.",
"prevention": "Implement a max_loop_count=2 constraint in the agent executor and validate supplier API health before calling place_order."
}
=== schema ===
{
"root_cause": "The create_lead tool received a string for the budget field when a decimal is required.",
"fix": "Cast the budget value to decimal before the API call: decimal.Decimal(budget) if isinstance(budget, str).",
"prevention": "Add Pydantic validation on the agent side before invoking create_lead."
}Wrap-up
This agent is already useful as a local debugging script, but its real power comes from running inside your CI pipeline. Two concrete next steps: wire the diagnose function into a LangGraph watchdog node so it fires automatically on failure, or stream agent traces from production into an Oxlo.ai-powered async worker that batches remediation reports every hour. If your current provider bills per token, compare your long-context debugging costs against Oxlo.ai flat per-request pricing at https://oxlo.ai/pricing.
Top comments (0)