How to track AI agent conversation state with structured logging and correlation IDs

What this does

This guide adds structured JSON logging with correlation IDs to an AI agent, enabling operators to reconstruct the full conversation state for any user session across distributed services. Each incoming request receives a unique correlation ID that propagates through all agent reasoning steps, tool calls, and model invocations. The resulting log stream can be queried by conversation ID to produce a complete transcript and decision trace.

Steps

1. Install the structured logging library:

   pip install python-json-logger
   

   Expected output: Successfully installed python-json-logger-2.0.7.

2. Configure the JSON logger in a logging_setup.py module:

   import logging
   from pythonjsonlogger import jsonlogger
   
   handler = logging.StreamHandler()
   formatter = jsonlogger.JsonFormatter("%(asctime)s %(name)s %(levelname)s %(message)s")
   handler.setFormatter(formatter)
   logger = logging.getLogger("ai-agent")
   logger.addHandler(handler)
   logger.setLevel(logging.INFO)
   

3. Generate correlation IDs at the request boundary. In the FastAPI middleware or Flask before-request hook:

   import uuid
   @app.middleware("http")
   async def add_correlation_id(request, call_next):
       correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
       request.state.correlation_id = correlation_id
       response = await call_next(request)
       response.headers["X-Correlation-ID"] = correlation_id
       return response
   

4. Inject the correlation ID into all log statements within the agent loop:

   logger.info("agent_step_started", extra={
       "correlation_id": ctx.correlation_id,
       "conversation_id": ctx.conversation_id,
       "step": step_number,
       "action": action_name,
   })
   

5. Log state transitions including tool invocations and model responses:

   logger.info("tool_called", extra={
       "correlation_id": ctx.correlation_id,
       "tool_name": tool.name,
       "input_snippet": str(tool_input)[:200],
       "duration_ms": elapsed_ms,
   })
   

6. Log the final agent response with full conversation metadata:

   logger.info("conversation_complete", extra={
       "correlation_id": ctx.correlation_id,
       "total_steps": step_count,
       "tokens_used": total_tokens,
       "outcome": "completed" if success else "error",
   })
   

7. Query logs by correlation ID. With jq on a log file:

   cat agent.log | jq 'select(.correlation_id == "abc-123")'
   

   Expected output: all log lines for that conversation, ordered by timestamp.

Verification

cat agent.log | jq -s 'group_by(.correlation_id) | map({id: .[0].correlation_id, steps: length})' | jq '.[0].steps'

Expected output: an integer showing the number of log entries for the first correlation ID.

Common failures

• Correlation ID missing from logs — ensure extra dict keys match the formatter fields. Add correlation_id to the formatter's reserved_attrs or use %(correlation_id)s in the format string.
• JSON parsing fails on log lines — non-JSON messages from library loggers pollute the stream. Set logging.getLogger("urllib3").setLevel(logging.WARNING) and use JSON handlers only for the agent logger.
• State reconstruction incomplete — missing intermediate logs indicate the agent raised an exception before logging. Wrap the agent loop in a try/finally block that logs conversation_aborted with the correlation ID.

Related guides

• Build a structured prompt/response logging pipeline with Fluentd
• Set up centralized logging for distributed AI agents using the ELK stack
• Log conversational context windows for AI agent debugging