Default LangChain verbosity dumps human-readable text to stderr, which is useless for grep or log aggregation. Adding structured logging langchain agents converts every model call, tool invocation, and intermediate decision into queryable JSON events. This guide walks through a concrete implementation you can drop into a production service.
Step 1: Install and configure a JSON logger
Start with a logging stack that emits one JSON object per line. pythonjsonlogger is lightweight and works with the standard library. Avoid pretty-printing in production; you want machine-parseable records.
pip install pythonjsonlogger langchain-core langchain-openaiConfigure the root logger for your agent process:
import logging
from pythonjsonlogger import jsonlogger
def setup_logger() -> logging.Logger:
log = logging.getLogger("agent")
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
"%(asctime)s %(levelname)s %(name)s %(message)s"
)
handler.setFormatter(formatter)
log.addHandler(handler)
log.setLevel(logging.INFO)
return log
logger = setup_logger()Every log call now produces a flat JSON document. Keep the schema stable: fixed keys (event, trace_id, run_id) make downstream querying trivial.
Step 2: Implement a LangChain callback handler
LangChain exposes BaseCallbackHandler from langchain_core.callbacks. Subclass it and override the hooks you care about. At minimum, capture LLM starts/ends, chain starts, agent actions, and tool ends. Push each event through the JSON logger with structured extra fields.
from langchain_core.callbacks import BaseCallbackHandler
import uuid
from contextvars import ContextVar
trace_id_var: ContextVar[str | None] = ContextVar("trace_id", default=None)
class StructuredLoggingHandler(BaseCallbackHandler):
def __init__(self, logger: logging.Logger):
self.logger = logger
def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id, **kwargs):
if trace_id_var.get() is None:
trace_id_var.set(str(uuid.uuid4()))
self.logger.info("chain_start", extra={
"event": "chain_start",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"parent_run_id": str(parent_run_id) if parent_run_id else None,
"inputs": inputs,
})
def on_llm_start(self, serialized, prompts, *, run_id, parent_run_id, **kwargs):
self.logger.info("llm_start", extra={
"event": "llm_start",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"model": serialized.get("kwargs", {}).get("model", "unknown"),
"prompt_count": len(prompts),
})
def on_llm_end(self, response, *, run_id, parent_run_id, **kwargs):
token_usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
self.logger.info("llm_end", extra={
"event": "llm_end",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"token_usage": token_usage,
})
def on_agent_action(self, action, *, run_id, parent_run_id, **kwargs):
self.logger.info("agent_action", extra={
"event": "agent_action",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"tool": action.tool,
"tool_input": action.tool_input,
"log": action.log,
})
def on_tool_end(self, output, *, run_id, parent_run_id, **kwargs):
self.logger.info("tool_end", extra={
"event": "tool_end",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"output": output,
})
def on_chain_error(self, error, *, run_id, parent_run_id, **kwargs):
self.logger.error("chain_error", extra={
"event": "chain_error",
"trace_id": trace_id_var.get(),
"run_id": str(run_id),
"error": str(error),
})Attach this handler to any executor or chain via the callbacks=[...] argument. The handler receives the same run_id LangChain uses internally, which lets you reconstruct a single agent run from disjoint log lines.
Step 3: Propagate a correlation ID across the agent run
The trace_id in Step 2 is set once per chain start using a ContextVar. This survives across async boundaries if you use contextvars correctly. Set it at the entrypoint of a request so all nested LangChain activity shares one identifier.
import uuid
from contextvars import ContextVar
trace_id_var = ContextVar("trace_id", default=None)
def handle_request(user_query: str):
trace_id_var.set(str(uuid.uuid4()))
# pass agent_executor with StructuredLoggingHandler(logger)
return agent_executor.invoke({"input": user_query})If you run multiple agents in a service, include trace_id in every log line. This is the single most useful field for debugging structured logging langchain agents in distributed systems.
Step 4: Log tool calls with explicit schemas
Tools are where agents silently fail. Wrap your tools so inputs and outputs are serialized cleanly. Use Pydantic or typed dicts; never log raw objects with circular references.
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Return current weather for a city."""
# pretend network call
return f"Sunny in {city}, 22C"
# In the handler, on_agent_action already logs tool + tool_input.
# on_tool_end logs the output. Add redaction if needed:
def redact(output: str) -> str:
if "password" in output.lower():
return "REDACTED"
return outputFor production, add a logging filter that scans extra for sensitive keys. A filter is cleaner than scattering redaction logic in each callback.
class RedactFilter(logging.Filter):
def filter(self, record):
if isinstance(getattr(record, "inputs", None), dict):
record.inputs.pop("api_key", None)
return True
logger.addFilter(RedactFilter())Step 5: Route LLM traffic through a unified endpoint
If your agent calls multiple model providers, you lose token accounting when each SDK logs differently. Route ChatOpenAI (or compatible) through a single OpenAI-compatible gateway. For example, n4n.ai exposes one endpoint across 240+ models with automatic fallback when a provider is rate-limited or degraded, and emits per-token usage metering. Because it honors client routing directives and forwards provider cache-control hints, your structured logging langchain agents can pair local traces with gateway-side token accounting using the same trace_id.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
base_url="https://api.n4n.ai/v1",
api_key="your-gateway-key",
callbacks=[StructuredLoggingHandler(logger)],
)
agent = AgentExecutor.from_agent_and_tools(
agent=build_agent(llm),
tools=[get_weather],
callbacks=[StructuredLoggingHandler(logger)],
)The on_llm_end token usage from LangChain will match the gateway’s metering, giving you a cross-check.
Step 6: Verify end-to-end logging
Run a minimal script that triggers a tool call and inspect stdout.
if __name__ == "__main__":
trace_id_var.set("test-trace-123")
agent = AgentExecutor.from_agent_and_tools(
agent=build_agent(llm),
tools=[get_weather],
callbacks=[StructuredLoggingHandler(logger)],
)
agent.invoke({"input": "What is the weather in Berlin?"})Successful verification means:
- You see
chain_start,llm_start,agent_action,tool_end,llm_end, andchain_end(addon_chain_endsimilarly) lines. - Every line contains
"trace_id": "test-trace-123". - The
llm_endline includes a non-emptytoken_usageobject. - No
chain_errorappears unless the tool intentionally fails.
Pipe to jq to confirm structure:
python agent.py 2>&1 | jq 'select(.event=="agent_action")'You should get a JSON object with tool and tool_input. If that works, your structured logging langchain agents pipeline is correctly instrumented.
Production considerations
Log volume scales with agent steps. Sample low-value llm_start events if cost is a concern, but always keep agent_action and chain_error. Use an async handler (AsyncBaseCallbackHandler) when your agent runs in an event loop to avoid blocking on network log shipping. Finally, standardize field names across services: trace_id, run_id, event, timestamp. That turns scattered logs into a queryable timeline.