LangGraph Tutorial (Python): adding observability for intermediate developers
This tutorial shows how to add practical observability to a LangGraph app in Python: tracing node execution, inspecting state transitions, and exporting runs to a real backend. You need this when your graph works locally but you can’t answer basic questions like “which node failed?”, “what state changed?”, or “why did this branch run?”
What You'll Need
- •Python 3.10+
- •
langgraph - •
langchain-core - •
langchain-openai - •A LangSmith account and API key
- •An OpenAI API key
- •Environment variables set for:
- •
LANGSMITH_API_KEY - •
LANGSMITH_TRACING=true - •
OPENAI_API_KEY
- •
Install the packages:
pip install langgraph langchain-core langchain-openai
Step-by-Step
- •Start with a small graph that has two nodes and a conditional branch. Observability is only useful if the graph has enough structure to inspect, so we’ll use a simple router plus an LLM-backed response node.
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
route: str
def route_node(state: State) -> dict:
last = state["messages"][-1].content.lower()
return {"route": "billing" if "invoice" in last or "bill" in last else "general"}
def general_node(state: State) -> dict:
return {"messages": [AIMessage(content="I can help with that.")]}
def billing_node(state: State) -> dict:
return {"messages": [AIMessage(content="I found the billing path.")]}
builder = StateGraph(State)
builder.add_node("route", route_node)
builder.add_node("general", general_node)
builder.add_node("billing", billing_node)
builder.add_edge(START, "route")
builder.add_conditional_edges(
"route",
lambda s: s["route"],
{"general": "general", "billing": "billing"},
)
builder.add_edge("general", END)
builder.add_edge("billing", END)
graph = builder.compile()
- •Turn on tracing with environment variables before you run anything. LangGraph will emit traces automatically when LangSmith tracing is enabled, which gives you node-level visibility without adding custom logging everywhere.
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = os.environ.get("LANGSMITH_API_KEY", "")
os.environ["LANGCHAIN_PROJECT"] = "langgraph-observability-tutorial"
os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY", "")
- •Add a real model-backed node so the trace shows both graph execution and model calls. This is where observability pays off: you can see prompt input, output, latency, and failures per node instead of guessing from application logs.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def llm_general_node(state: State) -> dict:
response = llm.invoke(
[
HumanMessage(
content="Reply briefly and clearly to the user's request: "
+ state["messages"][-1].content
)
]
)
return {"messages": [response]}
builder2 = StateGraph(State)
builder2.add_node("route", route_node)
builder2.add_node("general", llm_general_node)
builder2.add_node("billing", billing_node)
builder2.add_edge(START, "route")
builder2.add_conditional_edges(
"route",
lambda s: s["route"],
{"general": "general", "billing": "billing"},
)
builder2.add_edge("general", END)
builder2.add_edge("billing", END)
graph_with_llm = builder2.compile()
- •Use a streaming run to inspect intermediate state while the graph executes. This is the fastest way to debug production graphs because you can see each step as it happens instead of waiting for the final result.
inputs = {
"messages": [HumanMessage(content="Can you help me understand this invoice?")],
"route": "",
}
for event in graph_with_llm.stream(inputs):
print(event)
- •Add explicit metadata tags so traces are searchable in LangSmith. In real systems, tags are what let you separate customer support flows from underwriting flows or isolate one tenant during incident review.
config = {
"configurable": {},
"tags": ["tutorial", "observability", "billing-flow"],
}
result = graph_with_llm.invoke(
{
"messages": [HumanMessage(content="What does this invoice charge mean?")],
"route": "",
},
config=config,
)
print(result["messages"][-1].content)
Testing It
Run the script once with a billing-related prompt and once with a general prompt. In LangSmith, you should see separate runs for the graph execution and nested spans for each node, including the LLM call inside llm_general_node.
Check that the trace shows the route decision changing between "billing" and "general". If you used tags correctly, filter by billing-flow and confirm that only matching runs appear.
If nothing shows up, verify your environment variables before importing or compiling the graph. The most common failure mode is setting LANGSMITH_TRACING too late or missing LANGSMITH_API_KEY.
Next Steps
- •Add custom metadata per customer or per workflow step using
configurablefields. - •Learn how to attach callbacks for structured logs alongside LangSmith traces.
- •Explore LangGraph checkpoints so you can trace not just execution but also state recovery across retries and human-in-the-loop steps.
Keep learning
- •The complete AI Agents Roadmap — my full 8-step breakdown
- •Free: The AI Agent Starter Kit — PDF checklist + starter code
- •Work with me — I build AI for banks and insurance companies
By Cyprian Aarons, AI Consultant at Topiax.
Want the complete 8-step roadmap?
Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.
Get the Starter Kit