Module 5: Multi-Step Reasoning and Planning

6. Reasoning Traces and Explainability

Overview

Your agent already plans, decomposes tasks, and self-corrects. But when something goes wrong — and in production something always goes wrong — you have a problem: where exactly did it go wrong? Without visibility into the internal reasoning, debugging an agent is like debugging a program with no logs: you run it, it fails, and you only see the final output. Reasoning traces solve this by making every step of the agent's thinking visible.

A reasoning trace isn't a generic log. It's a structured record of what the agent thought at each step, why it made each decision, what alternatives it considered, and how confident it was. Logs say "the search tool was called"; a trace says "the agent decided to search the web because the question requires up-to-date data, it considered using the local database but discarded it because the data is from 2023, confidence: 0.8."

The three main use cases are debugging (finding where it failed), observability (monitoring in production), and explainability (telling the user "I did X because Y"). In this capsule you'll implement all three: capturing traces in the state, defining a structured format, analyzing traces for debugging, integrating with LangSmith, and turning traces into user-friendly explanations.

What a Reasoning Trace Is

More than logs — structured reasoning

Traditional log:
  [14:32:01] INFO: tool_call: search("AI regulation EU")
  [14:32:03] INFO: tool_result: 3 results found

Reasoning trace:
  Step 1:
    thought: "The question asks to compare EU vs US regulation."
    decision: "Search EU regulation first — it has a more defined framework."
    action: search("AI regulation EU 2024")
    confidence: 0.85
    alternatives_considered: ["search both in parallel", "search US first"]

The trace captures the chain of reasoning, not just the chain of execution. When you read a trace, you can reconstruct the agent's mental process and find exactly where its logic broke.

Anatomy of a trace

Layer 1: Metadata
  ├── trace_id, task, model, timestamps

Layer 2: Steps (the core)
  ├── Step 1: { thought, decision, action, observation, confidence }
  ├── Step 2: { thought, decision, action, observation, confidence }
  └── Step N: ...

Layer 3: Summary
  ├── final_answer, total_tool_calls, total_tokens

The five fields of each step:

thought: the reasoning before the action
decision: what it decided to do and why
action: the concrete action (tool call, answer, re-plan)
observation: the action's result
confidence: how sure it was (0.0-1.0)

Don't confuse traces with chain-of-thought prompting. CoT is a prompting technique. A trace is an observability artifact that you build — it captures what the agent did regardless of whether you used CoT.

Capturing Traces in the State

Adding trace fields to the AgentState

The natural place to capture traces in LangGraph is the state:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from datetime import datetime

def add_trace_entries(existing: list, new: list) -> list:
    """Reducer that accumulates trace entries without overwriting."""
    return existing + new

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    trace: Annotated[list[dict], add_trace_entries]
    current_step: int
    plan: list[str]
    results: dict

The add_trace_entries reducer is key: each node returns new entries and the reducer concatenates them. You never lose the history.

Nodes that record their reasoning

Each node writes to the trace with the real reasoning that led to its decision:

model = init_chat_model("openai:gpt-4.1-mini")

def plan_node(state: AgentState) -> dict:
    task = state["task"]
    step = state.get("current_step", 0)
    response = model.invoke([
        SystemMessage(content="Generate 3-5 steps to solve the task. One per line."),
        HumanMessage(content=f"Task: {task}"),
    ])
    steps = [s.strip() for s in response.content.strip().split("\n") if s.strip()]
    return {
        "plan": steps,
        "trace": [{
            "step": step, "node": "planner", "timestamp": datetime.now().isoformat(),
            "thought": f"Analyzing task: '{task}'. I need to break it down.",
            "decision": f"Generate a plan with {len(steps)} steps.",
            "action": "plan_generation", "observation": f"Plan: {steps}", "confidence": 0.8,
        }],
        "current_step": step + 1,
        "messages": [SystemMessage(content=f"Plan: {steps}")],
    }

def execute_node(state: AgentState) -> dict:
    plan = state.get("plan", [])
    step = state.get("current_step", 0)
    results = dict(state.get("results", {}))
    if not plan:
        return {"trace": [{"step": step, "node": "executor",
            "timestamp": datetime.now().isoformat(), "thought": "There are no pending steps.",
            "decision": "Finish.", "action": "none", "observation": "Empty plan.",
            "confidence": 1.0}], "current_step": step + 1}

    current_task = plan[0]
    context = "\n".join(f"[{k}]: {v[:200]}" for k, v in results.items()) or "No context."
    response = model.invoke([HumanMessage(content=(
        f"Context:\n{context}\n\nExecute: {current_task}\nBe concise."
    ))])
    results[f"step_{step}"] = response.content
    return {
        "plan": plan[1:], "results": results,
        "trace": [{
            "step": step, "node": "executor", "timestamp": datetime.now().isoformat(),
            "thought": f"Executing: '{current_task}'.",
            "decision": f"Invoke the LLM for '{current_task}'.",
            "action": f"llm_invoke: {current_task}",
            "observation": response.content[:300], "confidence": 0.75,
        }],
        "current_step": step + 1,
        "messages": [SystemMessage(content=f"Completed: {current_task}")],
    }

Pattern: a trace wrapper decorator

If you have many nodes, a decorator avoids repeating boilerplate. It injects step, node, timestamp, and duration automatically:

from functools import wraps

def traced(node_name: str):
    def decorator(func):
        @wraps(func)
        def wrapper(state):
            step, start = state.get("current_step", 0), datetime.now()
            result = func(state)
            base = {"step": step, "node": node_name,
                    "timestamp": start.isoformat(),
                    "elapsed_seconds": (datetime.now() - start).total_seconds()}
            if "trace" in result and result["trace"]:
                for entry in result["trace"]:
                    entry.update({k: v for k, v in base.items() if k not in entry})
            else:
                result["trace"] = [{**base, "thought": "Auto.", "decision": f"Run {node_name}.",
                    "action": node_name, "observation": "Done.", "confidence": 0.5}]
            return result
        return wrapper
    return decorator

Structured Trace Format

A Pydantic model for production

Loose dictionaries work for prototypes. In production you need a validated schema:

from pydantic import BaseModel, Field

class TraceEntry(BaseModel):
    step: int
    node: str
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
    elapsed_seconds: float = Field(default=0.0)
    thought: str
    decision: str
    action: str
    observation: str
    confidence: float = Field(ge=0.0, le=1.0)
    alternatives_considered: list[str] = Field(default_factory=list)
    error: str | None = Field(default=None)

class FullTrace(BaseModel):
    trace_id: str
    task: str
    model: str = "gpt-4.1-mini"
    started_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    entries: list[TraceEntry] = Field(default_factory=list)
    final_answer: str | None = None

Factory and serialization

import uuid, json
from pathlib import Path

def create_trace_entry(step, node, thought, decision, action,
                       observation, confidence=0.5, error=None) -> dict:
    return TraceEntry(step=step, node=node, thought=thought, decision=decision,
        action=action, observation=observation, confidence=confidence, error=error).model_dump()

def save_trace(trace: FullTrace, directory: str = "./traces") -> str:
    Path(directory).mkdir(exist_ok=True)
    filepath = Path(directory) / f"{trace.trace_id}.json"
    filepath.write_text(json.dumps(trace.model_dump(), indent=2, ensure_ascii=False))
    return str(filepath)

Debugging with Traces

The trace as a map of the error

When an agent delivers an incorrect result, the question isn't "what went wrong?" — it's "where did it go wrong?" Example: you ask "What's the current population of Tokyo?" and it answers "37.4 million" (the metro area, not the city).

failed_trace = [
    {"step": 0, "node": "planner",
     "thought": "A simple factual question.",
     "decision": "Search directly.", "confidence": 0.9},
    {"step": 1, "node": "executor",
     "action": "search('Tokyo population')",
     "observation": "Tokyo metro: 37.4M. Tokyo city: 13.96M.",
     "confidence": 0.7},
    {"step": 2, "node": "synthesizer",
     "thought": "I have the data. The first number is 37.4M.",
     "decision": "Use 37.4M as the answer.",     # ← ERROR HERE
     "confidence": 0.85},
]

Step 1 got two data points but step 2 picked the first one without distinguishing metro vs city. Without the trace, you would have hunted for the bug in the search tool. With the trace, you know the fix belongs in the synthesizer.

Analysis functions

def find_low_confidence_steps(entries, threshold=0.6):
    return [e for e in entries if e.get("confidence", 1.0) < threshold]

def find_error_steps(entries):
    return [e for e in entries if e.get("error")]

def analyze_trace(entries: list[dict]) -> dict:
    low_conf = find_low_confidence_steps(entries)
    errors = find_error_steps(entries)
    nodes = [e["node"] for e in entries]
    return {
        "total_steps": len(entries),
        "low_confidence_steps": len(low_conf),
        "error_steps": len(errors),
        "decision_chain": [
            f"[Step {e['step']}] {e['node']}: {e.get('decision', '')}"
            for e in entries
        ],
        "node_distribution": {n: nodes.count(n) for n in set(nodes)},
        "avg_confidence": sum(e.get("confidence", 0) for e in entries) / max(len(entries), 1),
        "suspicious_steps": low_conf + errors,
    }

Comparing successful vs failed runs

A powerful pattern: compare the failed trace with a successful one for the same task. Look for the divergence point — the first step where the decisions differ. If they diverge at step 2, you know steps 0-1 were identical and the error started there.

def compare_traces(good: list[dict], bad: list[dict]) -> dict:
    good_d = [e.get("decision", "") for e in good]
    bad_d = [e.get("decision", "") for e in bad]
    divergence = next(
        (i for i in range(min(len(good_d), len(bad_d))) if good_d[i] != bad_d[i]), None
    )
    return {"divergence_at_step": divergence,
        "good_avg_conf": sum(e.get("confidence", 0) for e in good) / max(len(good), 1),
        "bad_avg_conf": sum(e.get("confidence", 0) for e in bad) / max(len(bad), 1)}

LangSmith for Observability

Setup

Every LangChain/LangGraph invocation gets traced automatically:

pip install langsmith

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_..."
os.environ["LANGSMITH_PROJECT"] = "research-agent-v2"

With those three lines, every invocation gets sent to LangSmith. You don't need to change your code.

Custom traces with `@traceable`

For functions that aren't LangChain's:

from langsmith import traceable

@traceable(name="quality_check")
def quality_check(response: str, query: str) -> dict:
    checks = {
        "has_answer": len(response) > 50,
        "mentions_query_terms": any(t in response.lower() for t in query.lower().split()),
        "reasonable_length": 50 < len(response) < 5000,
    }
    return {"passed": all(checks.values()), "checks": checks}

In LangSmith, this function shows up as a span inside the agent's trace.

Metadata and monitoring

Add metadata so you can find traces later:

@traceable(
    name="research_agent_run",
    tags=["production", "v2"],
    metadata={"user_id": "user_123", "priority": "high"},
)
def run_agent(query: str) -> str:
    result = agent.invoke({"messages": [HumanMessage(content=query)], "task": query,
                           "trace": [], "current_step": 0, "plan": [], "results": {}})
    return result["messages"][-1].content

In LangSmith you can filter: "show me production v2 runs where user_id=user_123". When a user reports a problem, you find their trace in seconds. For programmatic monitoring:

from langsmith import Client
client = Client()
runs = list(client.list_runs(
    project_name="research-agent-v2",
    filter='and(eq(status, "error"), gte(start_time, "2025-03-01"))',
))
print(f"Errors since March: {len(runs)}")

Explainability for Users

From internal traces to human narratives

Traces are for developers. Users need narratives: "I did X because Y." You need to turn the trace into something understandable.

def trace_to_explanation(entries: list[dict], task: str) -> str:
    parts = [f"To answer «{task}»:\n"]
    for i, entry in enumerate(entries):
        node = entry.get("node", "")
        if node == "planner":
            parts.append(f"**Step {i+1} — Planning:** I analyzed your question and created a plan.")
        elif node == "executor":
            action = entry.get("action", "")
            topic = action.split("(")[-1].strip(")'\"") if "(" in action else action
            obs = entry.get("observation", "")[:100]
            parts.append(f"**Step {i+1} — Research:** I searched about {topic}. I found: {obs}")
        elif node == "synthesizer":
            parts.append(f"**Step {i+1} — Synthesis:** I combined the information into an answer.")
    return "\n\n".join(parts)

Explanations with an LLM

For more natural explanations, an LLM can take the trace and narrate the process in the first person ("First I searched...", "Then I compared..."). The prompt should forbid technical jargon like "nodes" or "state".

Levels of explainability

from typing import Literal

def explain(entries: list[dict], task: str,
            level: Literal["brief", "detailed", "technical"] = "brief") -> str:
    if level == "brief":
        tools = len([e for e in entries if "search" in e.get("action", "")])
        return f"I researched your question in {len(entries)} steps, consulting {tools} sources."
    elif level == "detailed":
        return trace_to_explanation(entries, task)
    elif level == "technical":
        return json.dumps(entries, indent=2, ensure_ascii=False)

brief: One line — confirmation that the agent worked
detailed: A step-by-step narrative
technical: The full JSON for developers and auditors

Connection to the Project

In this module's project (capsule 08, Research Agent with Planning and Reflection):

The state includes trace: list[dict] with the add_trace_entries reducer. Each node writes entries with its reasoning.
The reflection node uses the trace to evaluate the run. It can catch "the agent searched only one source when the task required a comparison" — invisible if you only look at the answer.
The output includes the answer + an explanation generated from the trace.

In later modules:

M6 (Memory): Traces get persisted in long-term memory. The agent consults past traces to see which strategy worked — meta-learning based on history.
M9 (Testing): Traces are the basis of trajectory evaluation: "did it reach the right output for the right reasons?"
M10 (Production): LangSmith traces for continuous monitoring, alerts when confidence drops, per-node latency dashboards.

Troubleshooting

Problem 1: Traces grow too large and eat memory

Symptom: On long runs, the trace field accumulates entries with long observations. A heavy state.

Solution: Cap the size of observations:

MAX_OBS_LENGTH = 500

def safe_trace_entry(**kwargs) -> dict:
    obs = kwargs.get("observation", "")
    if len(obs) > MAX_OBS_LENGTH:
        kwargs["observation"] = obs[:MAX_OBS_LENGTH] + "... [truncated]"
    return create_trace_entry(**kwargs)

Problem 2: The trace doesn't capture tool errors

Symptom: A tool fails but the trace shows the step as successful or skips it.

Solution: Wrap it with explicit capture:

def execute_with_trace(tool_fn, tool_args, step, node):
    try:
        result = tool_fn(**tool_args)
        return create_trace_entry(step=step, node=node, thought="Executing the tool.",
            decision=f"Call {tool_fn.__name__}", action=f"{tool_fn.__name__}({tool_args})",
            observation=str(result)[:500], confidence=0.8)
    except Exception as e:
        return create_trace_entry(step=step, node=node, thought="Executing the tool.",
            decision=f"Call {tool_fn.__name__}", action=f"{tool_fn.__name__}({tool_args})",
            observation="FAILED", confidence=0.0, error=str(e))

Problem 3: Confidence scores are always the same

Symptom: Every step has confidence: 0.5. They don't reflect real uncertainty.

Solution: Make the LLM generate the confidence:

class ActionWithConfidence(BaseModel):
    action: str
    reasoning: str
    confidence: float = Field(ge=0.0, le=1.0)

confidence_model = model.with_structured_output(ActionWithConfidence)

Problem 4: LangSmith doesn't show custom functions

Symptom: LangChain invocations show up but your functions don't.

Solution: Decorate them with @traceable. Only decorated functions or LangChain invocations get sent.

from langsmith import traceable

@traceable(name="my_analysis")
def my_function(data):
    return result

Problem 5: Generic traces that don't help you debug

Symptom: Entries with "Processing data" and "Run the next step" — no useful info.

Solution: Ask the LLM to verbalize its reasoning before acting. If reading the trace gives you no more information than looking at the output, the trace is badly built.

Exercises

Exercise 1: Create a TraceEntry (Easy)

Define a Pydantic TraceEntry and create an instance for a step where the agent decides to search for "Python frameworks 2025".

See solution

from pydantic import BaseModel, Field
from datetime import datetime

class TraceEntry(BaseModel):
    step: int
    node: str
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
    thought: str
    decision: str
    action: str
    observation: str
    confidence: float = Field(ge=0.0, le=1.0)

entry = TraceEntry(
    step=1, node="executor",
    thought="The user wants up-to-date frameworks. I need recent data.",
    decision="Search 'Python frameworks 2025' for current-year data.",
    action="search('top Python frameworks 2025')",
    observation="FastAPI, Django, Flask lead. Litestar is growing. Pydantic AI is emerging.",
    confidence=0.85,
)
print(entry.model_dump_json(indent=2))

Exercise 2: Diagnose a failed trace (Medium)

Given this trace, implement diagnose_trace to identify the problematic step:

trace = [
    {"step": 0, "node": "planner", "decision": "Break it into 3 steps", "confidence": 0.9},
    {"step": 1, "node": "executor", "action": "search('Python vs Rust performance')",
     "observation": "Rust is 10-40x faster on CPU-bound work", "confidence": 0.85},
    {"step": 2, "node": "executor", "action": "search('Rust web frameworks')",
     "observation": "Actix-web, Axum — a mature ecosystem", "confidence": 0.80},
    {"step": 3, "node": "synthesizer", "decision": "Recommend Rust for all backends",
     "observation": "Rust is superior to Python for backends", "confidence": 0.60},
]
# Error: it ignores Python's advantages for AI/ML backends

See solution

def diagnose_trace(entries):
    low_conf = [(e["step"], e.get("confidence", 1)) for e in entries if e.get("confidence", 1) < 0.7]
    searches = [e.get("action", "") for e in entries if "search" in e.get("action", "")]

    issues = []
    if not any("Python" in s and ("advantage" in s.lower() or "benefit" in s.lower()) for s in searches):
        issues.append("No search was done for Python's specific advantages")

    suspect = min(low_conf, key=lambda x: x[1])[0] if low_conf else entries[-1]["step"]
    return {
        "suspect_step": suspect,
        "issues": issues,
        "diagnosis": f"Error at step {suspect}. Low confidence ({low_conf}) "
                     f"and a search biased toward Rust without researching Python's advantages.",
    }

print(diagnose_trace(trace)["diagnosis"])
# "Error at step 3. Low confidence [...] and a biased search..."

Exercise 3: Implement the @traced decorator (Medium)

Create a decorator that automatically injects step, node, timestamp, and duration into each trace entry.

See solution

from functools import wraps
from datetime import datetime

def traced(node_name: str):
    def decorator(func):
        @wraps(func)
        def wrapper(state):
            step = state.get("current_step", 0)
            start = datetime.now()
            result = func(state)
            elapsed = (datetime.now() - start).total_seconds()

            base = {"step": step, "node": node_name,
                    "timestamp": start.isoformat(), "elapsed_seconds": round(elapsed, 3)}

            if "trace" in result and result["trace"]:
                for entry in result["trace"]:
                    entry.update({k: v for k, v in base.items() if k not in entry})
            else:
                result["trace"] = [{**base, "thought": "Auto.", "decision": f"Run {node_name}.",
                    "action": node_name, "observation": "Done.", "confidence": 0.5}]
            return result
        return wrapper
    return decorator

@traced("researcher")
def research_node(state):
    return {"trace": [{"thought": "Look up data.", "decision": "Web search.",
        "action": "search('AI 2025')", "observation": "5 results.", "confidence": 0.8}]}

result = research_node({"current_step": 3})
print(result["trace"][0]["node"])   # "researcher"
print(result["trace"][0]["step"])   # 3

Exercise 4: Generate a user-friendly explanation (Medium)

Implement a function that turns a 4-entry trace into an explanation with no technical jargon.

See solution

def generate_user_explanation(entries: list[dict], task: str) -> str:
    parts = [f"To answer «{task}»:\n"]
    search_topics = []

    for entry in entries:
        node = entry.get("node", "")
        action = entry.get("action", "")

        if node == "planner":
            parts.append("1. I analyzed your question and created a research plan.")
        elif node == "executor" and "search" in action:
            topic = action.split("(")[-1].strip(")'\"") if "(" in action else action
            search_topics.append(topic)
            obs = entry.get("observation", "")[:100]
            parts.append(f"{len(search_topics)+1}. I searched for «{topic}» → {obs}")
        elif node == "synthesizer":
            parts.append(f"{len(search_topics)+2}. I combined {len(search_topics)} findings "
                         f"to give you a complete answer.")
    return "\n".join(parts)

sample = [
    {"node": "planner", "action": "plan", "observation": "3 sub-tasks"},
    {"node": "executor", "action": "search('FastAPI vs Django')",
     "observation": "FastAPI leads on async, Django on full-stack"},
    {"node": "executor", "action": "search('Flask 2025')",
     "observation": "Flask has 23% market share, growing slowly"},
    {"node": "synthesizer", "action": "synthesize", "observation": "Done"},
]
print(generate_user_explanation(sample, "Best Python framework for a backend?"))

Exercise 5: A LangGraph graph with complete tracing (Hard)

Implement a StateGraph with planner → executor (loop) → synthesizer → explainer. Each node adds trace entries. The explainer node generates an explanation from the accumulated trace.

See solution

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from datetime import datetime

model = init_chat_model("openai:gpt-4.1-mini")

def add_traces(a: list, b: list) -> list:
    return a + b

class TracedState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    trace: Annotated[list[dict], add_traces]
    plan: list[str]
    results: dict
    step_counter: int
    explanation: str

def planner(state: TracedState) -> dict:
    task = state["task"]
    resp = model.invoke([
        SystemMessage(content="Generate 3 research steps, one per line."),
        HumanMessage(content=f"Task: {task}"),
    ])
    steps = [s.strip() for s in resp.content.strip().split("\n") if s.strip()][:5]
    return {
        "plan": steps, "step_counter": 1,
        "trace": [{"step": 0, "node": "planner", "timestamp": datetime.now().isoformat(),
            "thought": f"Analyzing: '{task}'.", "decision": f"A {len(steps)}-step plan.",
            "action": "plan", "observation": str(steps), "confidence": 0.85}],
        "messages": [SystemMessage(content=f"Plan: {len(steps)} steps")],
    }

def executor(state: TracedState) -> dict:
    plan, results = state.get("plan", []), dict(state.get("results", {}))
    step = state.get("step_counter", 1)
    if not plan:
        return {"plan": [], "trace": []}
    current, ctx = plan[0], "\n".join(f"- {v[:150]}" for v in results.values())
    resp = model.invoke([HumanMessage(content=f"Context:\n{ctx}\n\nExecute: {current}")])
    results[f"step_{step}"] = resp.content
    return {
        "plan": plan[1:], "results": results, "step_counter": step + 1,
        "trace": [{"step": step, "node": "executor", "timestamp": datetime.now().isoformat(),
            "thought": f"Executing: '{current}'.", "decision": f"LLM invoke: {current}.",
            "action": f"execute: {current}", "observation": resp.content[:300], "confidence": 0.75}],
    }

def synthesizer(state: TracedState) -> dict:
    all_r = "\n\n".join(f"**{k}**: {v}" for k, v in state.get("results", {}).items())
    step = state.get("step_counter", 0)
    resp = model.invoke([HumanMessage(content=f"Task: {state['task']}\nResults:\n{all_r}\nSynthesize.")])
    return {
        "trace": [{"step": step, "node": "synthesizer", "timestamp": datetime.now().isoformat(),
            "thought": "Synthesizing.", "decision": "Final answer.",
            "action": "synthesize", "observation": resp.content[:300], "confidence": 0.85}],
        "messages": [resp],
    }

def explainer(state: TracedState) -> dict:
    parts, topics = [f"To answer «{state['task']}»:"], []
    for e in state.get("trace", []):
        if e["node"] == "planner":
            parts.append("- I created a research plan.")
        elif e["node"] == "executor":
            t = e.get("action", "").replace("execute: ", "")
            topics.append(t)
            parts.append(f"- I researched: {t}")
        elif e["node"] == "synthesizer":
            parts.append(f"- I combined {len(topics)} findings into an answer.")
    return {"explanation": "\n".join(parts)}

def route(state: TracedState) -> str:
    return "executor" if state.get("plan") else "synthesizer"

g = StateGraph(TracedState)
g.add_node("planner", planner)
g.add_node("executor", executor)
g.add_node("synthesizer", synthesizer)
g.add_node("explainer", explainer)
g.add_edge(START, "planner")
g.add_conditional_edges("planner", route, {"executor": "executor", "synthesizer": "synthesizer"})
g.add_conditional_edges("executor", route, {"executor": "executor", "synthesizer": "synthesizer"})
g.add_edge("synthesizer", "explainer")
g.add_edge("explainer", END)

agent = g.compile()
result = agent.invoke({"messages": [], "task": "Advantages of FastAPI over Django for APIs",
    "trace": [], "plan": [], "results": {}, "step_counter": 0, "explanation": ""})

print("=== Answer ===")
print(result["messages"][-1].content[:300])
print("\n=== Explanation ===")
print(result["explanation"])
print(f"\n=== Trace ({len(result['trace'])} entries) ===")
for e in result["trace"]:
    print(f"  [{e['step']}] {e['node']}: {e.get('decision', '')}")

Flow: planner → executor (loop) → synthesizer → explainer → END. Each node feeds the trace, and the explainer turns it into a narrative.

Summary

In this capsule you learned:

A reasoning trace is a structured record of the agent's thinking. It isn't a log — it captures what it thought, what it decided, why, and how sure it was. The difference between "search was called" and "it decided to search X because Y, with confidence 0.8."
Traces get captured in LangGraph's state. A trace field with a reducer that accumulates entries. Each node writes its reasoning. The @traced decorator automates the metadata.
A Pydantic format guarantees consistency. TraceEntry with step, node, timestamp, thought, decision, action, observation, confidence. Validated schemas — essential in production.
Traces are the main debugging tool. You read the trace to find where the reasoning broke. find_low_confidence_steps and compare_traces speed up the diagnosis.
LangSmith extends tracing into production. 3 environment variables. @traceable for custom functions. Metadata and tags for search and alerts.
Explainability turns traces into narratives for users. Three levels (brief, detailed, technical) for different audiences.

Next capsule: Comparing Reasoning Patterns — ReAct vs Plan-and-Execute vs Reflection, the same case solved with each pattern, with metrics so you can choose with data.

Additional Resources

LangSmith Documentation — Tracing, evaluation, and monitoring for LLM applications
LangSmith Tracing Guide — Setup, @traceable, metadata, filtering runs
LangGraph Observability — Debugging in LangGraph graphs
Chain-of-Thought Paper — The theoretical foundation of reasoning traces
EU AI Act — Explainability — Regulatory explainability requirements for AI in Europe

← Previous5. Reflection and Self-Correction Next →7. Comparing Reasoning Patterns

6. Reasoning Traces and Explainability

Overview

What a Reasoning Trace Is

More than logs — structured reasoning

Anatomy of a trace

Capturing Traces in the State

Adding trace fields to the AgentState

Nodes that record their reasoning

Pattern: a trace wrapper decorator

Structured Trace Format

A Pydantic model for production

Factory and serialization

Debugging with Traces

The trace as a map of the error

Analysis functions

Comparing successful vs failed runs

LangSmith for Observability

Setup

Custom traces with @traceable

Metadata and monitoring

Explainability for Users

From internal traces to human narratives

Explanations with an LLM

Levels of explainability

Connection to the Project

Troubleshooting

Problem 1: Traces grow too large and eat memory

Problem 2: The trace doesn't capture tool errors

Problem 3: Confidence scores are always the same

Problem 4: LangSmith doesn't show custom functions

Problem 5: Generic traces that don't help you debug

Exercises

Exercise 1: Create a TraceEntry (Easy)

Exercise 2: Diagnose a failed trace (Medium)

Exercise 3: Implement the @traced decorator (Medium)

Exercise 4: Generate a user-friendly explanation (Medium)

Exercise 5: A LangGraph graph with complete tracing (Hard)

Summary

Additional Resources

Custom traces with `@traceable`