Integrating Observability Into Your LangChain Pipeline

Execution comes with a LangChain agent, but a structured account of each step does not arrive automatically. You still need to record which tool ran, the context the LLM received, its output, span duration, and whether that output was correct. LangChain includes logging for development debugging, but production calls require a different view.

LangChain's callback system provides the needed mechanism. Your agent does not need restructuring: attach a callback handler, which can observe every event across the chain lifecycle.

How LangChain Callbacks Work

LangChain emits events throughout agent execution. LLM calls produce on_llm_start and on_llm_end; tools produce on_tool_start and on_tool_end; and the completed chain produces on_chain_end. A callback handler overrides the events you need, and it can be attached at chain level, LLM level, or globally.

Here is a minimal trace handler showing the shape of a run:

from langchain.callbacks.base import BaseCallbackHandler
from datetime import datetime, timezone

class TraceHandler(BaseCallbackHandler):
    def __init__(self, trace_id: str):
        self.trace_id = trace_id
        self.spans = []

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.spans.append({
            "trace_id": self.trace_id,
            "event": "llm_start",
            "ts": datetime.now(timezone.utc).isoformat(),
            "input_preview": prompts[0][:300] if prompts else ""
        })

    def on_llm_end(self, response, **kwargs):
        output = ""
        if response.generations:
            output = response.generations[0][0].text
        self.spans.append({
            "trace_id": self.trace_id,
            "event": "llm_end",
            "ts": datetime.now(timezone.utc).isoformat(),
            "output": output
        })

The result is a timestamped record of each LLM event, associated with a trace ID created when the invocation begins.

Attaching the Handler to Your Agent

Run configuration is the cleanest attachment point. It leaves agent initialization unchanged and allows handlers to vary by environment.

import uuid
from langchain.agents import AgentExecutor

def run_agent(user_input: str):
    trace_id = str(uuid.uuid4())
    handler = TraceHandler(trace_id=trace_id)

    result = agent_executor.invoke(
        {"input": user_input},
        config={"callbacks": [handler]}
    )

    persist_trace(handler.spans)
    return result

A key caveat applies when tools invoke LLMs themselves: those inner calls also emit callbacks, usually as desired. A tool that separately initializes a sub-chain may not inherit the parent's callbacks, however. LangChain's propagation rules changed between v0.1 and v0.2. If inner calls are missing from spans, verify that propagation was enabled when the sub-chain was built or pass an explicit handler during its initialization.

What to Capture in Each Span

Plain text capture is seldom enough to debug production failures. After adding callback hooks, decide deliberately what each span type should record.

LLM spans: Record the complete prompt, not merely a preview, to audit injected context; the complete response, including any chain-of-thought when reasoning prompts are used; response metadata token counts for tracking cost alongside quality; and latency computed from start and end timestamps.

Tool spans: Record the tool name, exact input string, output with truncation and a length flag when very large, and whether an exception occurred, including its exception class.

Chain-level events: Record the final agent output, the ordered tool-call sequence, and any intermediate reasoning steps. The sequence shows whether the agent followed the expected path or took an unexpected one.

A queryable table for this structured data supports searches that plain text logs cannot: find runs calling a particular tool more than twice, locate outputs containing a phrase, or compare token counts across two time windows for prompt bloat.

Adding Evaluation Hooks

Trace data becomes much more useful when evaluation scores accompany raw events. Add them in on_chain_end, after the final output exists and can be sent to a rubric-based scorer.

import os
from futureagi import Client as FAGIClient

fagi = FAGIClient(api_key=os.environ["FAGI_API_KEY"])

def on_chain_end(self, outputs, **kwargs):
    final_output = outputs.get("output", "")

    eval_result = fagi.evaluate(
        response=final_output,
        rubric="factual_accuracy",
        context=self._last_retrieved_context
    )

    self.spans.append({
        "trace_id": self.trace_id,
        "event": "evaluation",
        "score": eval_result.score,
        "flags": eval_result.flags,
        "ts": datetime.now(timezone.utc).isoformat()
    })

    if eval_result.score < 0.6:
        trigger_alert(self.trace_id, eval_result)

Evaluation runs within the callback rather than as a separate offline job, so every production call receives a near-real-time score. When scores decline, the change appears within minutes instead of surfacing through a support ticket three days later.

Persisting and Querying Traces

Collected spans matter only when they are queryable. A common starting point is a Postgres table with a JSONB span payload and indexed trace_id, event, and ts columns. This supports queries such as:

SELECT trace_id, span_payload->>'score' as score
FROM agent_spans
WHERE event = 'evaluation'
  AND (span_payload->>'score')::float < 0.7
  AND ts > NOW() - INTERVAL '24 hours'
ORDER BY score ASC
LIMIT 50;

The query identifies runs from the last 24 hours with the lowest evaluation scores. You can then retrieve each full trace and inspect the LLM's input and output. Structured traces make this debugging workflow possible, unlike flat log lines.

What This Does Not Cover

Callbacks observe events within the chain, but they do not catch transport failures such as network timeouts, LLM provider rate limit errors, or JSON parsing failures that occur before an LLM call starts. Those events never reach on_llm_start. Use ordinary exception handling around agent invocation and log transport errors separately.

Callbacks also provide no aggregate view without a query layer. Writing spans to a file and never examining them means you logged everything but learned nothing. Regular trace queries matter more than storage sophistication. A morning cron job that runs SQL and sends the day's score distribution to Slack can be more useful than a complex dashboard nobody opens.

This approach also adds latency to each agent run because the example executes evaluation synchronously in on_chain_end. For latency-sensitive applications, send evaluation to a background task queue, mark the span pending, and fill in the score asynchronously. The trace remains complete; its score arrives a few seconds later.

Agent restructuring is unnecessary. The agent and prompt templates remain unchanged. You are consuming telemetry that callbacks already expose and routing it to a place where your team can act.