How to Debug and Trace LangGraph Workflow Execution in Production RAG Systems

You can debug LangGraph workflows by visualizing the graph topology with Mermaid diagrams, enabling Langfuse tracing for end-to-end observability, inspecting the AgentState at each node, and leveraging structured Python logging throughout the execution pipeline.

The jamwithai/production-agentic-rag-course repository implements an Agentic RAG service using LangGraph’s StateGraph to orchestrate guardrail validation, document retrieval, grading, and answer generation. Understanding how to debug and trace LangGraph workflow execution is essential for production deployments, as it reveals bottlenecks, routing decisions, and state mutations across the multi-step pipeline.

Visualize the Workflow Topology

Before running the graph, inspect its static structure using the visualization methods in src/services/agents/agentic_rag.py (lines 388–462). These helpers expose the compiled graph’s nodes and edges without invoking the LLM.

• get_graph_mermaid() – Returns a Mermaid syntax string (line ~388) you can paste into markdown viewers or mermaid.live.
• get_graph_ascii() – Generates an ASCII representation for terminal debugging.
• get_graph_visualization() – Produces PNG bytes (line ~406) for documentation or static diagrams.

from src.services.agents.agentic_rag import AgenticRAGService

service = AgenticRAGService(
    opensearch_client=os_client,
    ollama_client=ollama,
    embeddings_client=jina,
)

# Mermaid diagram for documentation

print(service.get_graph_mermaid())

# ASCII for quick terminal checks

print(service.get_graph_ascii())

# PNG for architectural reviews

with open("rag_workflow.png", "wb") as f:
    f.write(service.get_graph_visualization())

Enable End-to-End Tracing with Langfuse

The repository integrates Langfuse for distributed tracing. When you pass a LangfuseTracer instance to AgenticRAGService, the service automatically injects a CallbackHandler into the graph execution configuration (lines 70–78 of agentic_rag.py).

Automatic LLM Instrumentation

The CallbackHandler wraps every LLM call within the graph. It is attached in the _run_workflow method:

# Excerpt from src/services/agents/agentic_rag.py (lines 70-78)

config = {"thread_id": f"user_{user_id}_session_{int(time.time())}"}
if self.langfuse_tracer and trace:
    callback_handler = CallbackHandler()  # Langfuse v3 auto-links to current span

    config["callbacks"] = [callback_handler]
result = await self.graph.ainvoke(state_input, config=config, context=runtime_context)

Create Custom Spans in Nodes

Individual nodes create explicit spans for granular observability. For example, the guardrail node uses the legacy Langfuse v2 SDK to track validation scores (lines 59–75 of src/services/agents/nodes/guardrail_node.py), while other nodes use langfuse_tracer.start_span() from src/services/langfuse/client.py (lines 90–104).

from src.services.langfuse.client import LangfuseTracer
from src.config import Settings

tracer = LangfuseTracer(Settings())  # Reads LANGFUSE_* env vars

service = AgenticRAGService(os_client, ollama, jina, langfuse_tracer=tracer)

# Wrap execution in a trace (v3 style)

with tracer.start_as_current_span(name="agentic_rag_request", user_id="alice") as trace:
    response = await service.ask(
        query="Explain Retrieval-Augmented Generation",
        user_id="alice"
    )
    # Trace contains spans for guardrail, retrieval, grading, and generation

Inspect State and Enable Debug Logging

Dump AgentState at Runtime

The shared state is defined as a TypedDict in src/services/agents/state.py (lines 9–72). Since every node receives and returns a slice of this state, you can dump it between steps to trace data flow.

import json
from pprint import pprint

# Inspect final state after execution

result = await service.ask("What is retrieval-augmented generation?")
print(json.dumps(result, indent=2))

# Contains: answer, sources, reasoning_steps, etc.

# Inside a custom node, log the incoming state

logger.debug(f"State keys before grading: {list(state.keys())}")

Configure Python Logging Levels

All nodes use logging.getLogger(__name__) to emit decision points and fallback warnings (e.g., lines 13–38 of guardrail_node.py). Set LOG_LEVEL=DEBUG to see internal routing decisions.

export LOG_LEVEL=DEBUG

Typical output from the guardrail node shows the threshold comparison and routing decision:

INFO:guardrail_node:Guardrail score: 92, threshold: 80
INFO:guardrail_node:Decision -> continue

If a node falls back due to an LLM error, you’ll see a warning with the fallback score (lines 15–22 of guardrail_node.py).

Complete Production Debugging Example

Combine visualization, tracing, and state inspection for full observability:

import logging
from src.services.langfuse.client import LangfuseTracer
from src.config import Settings
from src.services.agents.agentic_rag import AgenticRAGService

logging.basicConfig(level=logging.DEBUG)

tracer = LangfuseTracer(Settings())
service = AgenticRAGService(os_client, ollama, jina, langfuse_tracer=tracer)

# 1. Visualize before execution

print(service.get_graph_mermaid())

# 2. Execute with tracing and state inspection

with tracer.start_as_current_span(name="debug_run", user_id="dev") as trace:
    result = await service.ask("How does vector similarity work?", user_id="dev")
    print("Final state keys:", result.keys())

Summary

• Visualize first – Use get_graph_mermaid() or get_graph_visualization() from src/services/agents/agentic_rag.py to verify graph topology before debugging runtime behavior.
• Trace everything – Inject CallbackHandler via the Langfuse tracer in _run_workflow (lines 70–78) to capture LLM spans automatically.
• Log decisions – Enable DEBUG level logging to see threshold scores and routing decisions in nodes like guardrail_node.py.
• Inspect state – The AgentState TypedDict in state.py (lines 9–72) provides a structured view of data flow; dump it after service.ask() calls to verify transformations.

Frequently Asked Questions

How do I visualize a LangGraph workflow structure without executing it?

Call get_graph_mermaid(), get_graph_ascii(), or get_graph_visualization() on an initialized AgenticRAGService instance. These methods access the compiled graph’s internal representation (lines 388–462 of src/services/agents/agentic_rag.py) and return static formats you can inspect without invoking any LLM calls or retrieval operations.

What is the best way to track state changes between nodes in LangGraph?

Dump the AgentState TypedDict—which is defined in src/services/agents/state.py (lines 9–72)—either inside node implementations using logger.debug(state) or by inspecting the dictionary returned from service.ask(). Since each node returns a partial update to this state, comparing keys before and after execution reveals exactly how data mutates through the pipeline.

How do I configure distributed tracing for LangGraph workflows?

Instantiate LangfuseTracer from src/services/langfuse/client.py and pass it to AgenticRAGService. The service automatically creates a CallbackHandler (lines 64–76 of client.py) and injects it into the graph configuration during _run_workflow (lines 70–78 of agentic_rag.py). Wrap your ask() calls in tracer.start_as_current_span() to create parent traces that capture all node spans and LLM generations.

How can I debug node-specific logic in a LangGraph application?

Raise the Python logging level to DEBUG and examine the logs in individual node files such as src/services/agents/nodes/guardrail_node.py. Each node logs its input parameters, decision thresholds (e.g., guardrail scores), and fallback paths (lines 13–38). You can also add custom Langfuse spans using the tracer’s start_span() method to isolate specific logic blocks within a node for detailed performance analysis.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}