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

> Debug and trace LangGraph workflow execution in production RAG systems. Visualize graphs, use Langfuse, inspect states, and leverage structured logging for seamless observability. Improve your AI agent performance.

- Repository: [jamwithai/production-agentic-rag-course](https://github.com/jamwithai/production-agentic-rag-course)
- Tags: how-to-guide
- Published: 2026-03-23

---

**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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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.

```python
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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/agentic_rag.py)).

### Automatic LLM Instrumentation

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

```python

# 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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/agents/nodes/guardrail_node.py)), while other nodes use `langfuse_tracer.start_span()` from [`src/services/langfuse/client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/langfuse/client.py) (lines 90–104).

```python
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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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.

```python
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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/guardrail_node.py)). Set `LOG_LEVEL=DEBUG` to see internal routing decisions.

```bash
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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/guardrail_node.py)).

## Complete Production Debugging Example

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

```python
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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/guardrail_node.py).
- **Inspect state** – The `AgentState` TypedDict in [`state.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/src/services/langfuse/client.py) and pass it to `AgenticRAGService`. The service automatically creates a `CallbackHandler` (lines 64–76 of [`client.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/client.py)) and injects it into the graph configuration during `_run_workflow` (lines 70–78 of [`agentic_rag.py`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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`](https://github.com/jamwithai/production-agentic-rag-course/blob/main/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.