← Back to Blog
Jun 30 2026 · 12 min read

Build a Stateful Agent with LangGraph and Trace It with Langfuse

PythonLangGraphLangfuseAI AgentsObservabilityOpenAITutorial

Most agent demos die the moment a user asks a follow-up question. The model forgets what it just said, the tool calls collide, and you have no idea which step burned the budget. The fix is a graph that holds state between steps, plus tracing that shows you what actually happened.

This article builds exactly that: a LangGraph agent that remembers the conversation, calls a real tool, and ships a full trace to Langfuse. You will end with a working file you can run.

Prerequisites

  • Python 3.10 or newer

  • An OpenAI API key (OPENAI_API_KEY) or any other chat model you can wrap with LangChain

  • A free Langfuse account at cloud.langfuse.com to grab LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY

Install the four packages you actually need:

pip install "langgraph>=0.2" "langchain-openai>=0.2" "langfuse>=3" python-dotenv

At the time of writing, langgraph was on 0.6.x and langfuse on 3.x. Pin above those and the import paths in this article stay correct.

What a LangGraph Agent Actually Is

LangGraph models an agent as a directed graph. Each node is a Python function. The state is a typed dict that flows through edges. A conditional edge picks the next node based on the state, which is how you decide whether to call a tool, ask the user, or stop.

The minimum useful agent has four pieces:

  • A state schema that holds the message history

  • A model node that calls the LLM

  • A tools node that runs functions the model requested

  • A conditional edge that loops back to the model until it stops calling tools

Once you have those, every other pattern (human-in-the-loop, parallel tool calls, subgraphs) is a variation on this shape.

Step 1: Define the State

LangGraph uses Python type hints. Annotated with a reducer tells the graph how to merge updates.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

The add_messages reducer appends new messages and overwrites by id. Without it, every node would replace the whole list, and your agent would lose context after one turn. This is the single most common bug in beginner LangGraph code.

Step 2: A Real Tool

A tool is just a function with a docstring and a type hint. The model reads the docstring to decide when to call it.

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city. Use this when the user asks about weather."""
    # Stub. Swap for a real API call.
    lookup = {
        "jakarta": "31C, partly cloudy",
        "singapore": "29C, thunderstorms",
        "tokyo": "22C, clear",
    }
    return lookup.get(city.lower(), f"No data for {city}")

Keep tool docstrings short and specific. A model that is unsure whether to call a tool will skip it, and you will spend an afternoon debugging prompts when the real issue is a vague description.

Step 3: Wire the Model and the Graph

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

load_dotenv()

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [get_weather]
model_with_tools = model.bind_tools(tools)

def call_model(state: State) -> dict:
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

graph = StateGraph(State)
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")

agent = graph.compile()

ToolNode is the prebuilt node that reads tool_calls off the last message, runs the right tool, and returns a ToolMessage. You can write this by hand, but the prebuilt version handles parallel tool calls and error wrapping for free.

The should_continue function is the conditional edge. It returns the name of the next node. If the model just made a tool call, we go to tools. Otherwise we hit END and the graph returns the final answer.

Step 4: Run It

from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="What is the weather in Jakarta and Singapore?")]
})

for msg in result["messages"]:
    msg.pretty_print()

Because the model can call tools in parallel, you should see one AIMessage with two tool_calls, two ToolMessage results, and a final AIMessage that combines the answer. The total graph steps: model, tools, model. Three node visits, one round trip.

Step 5: Trace It with Langfuse

This is where most teams stop. They ship the agent, hit a bug in production, and have no way to see which step broke. Langfuse is an open source tracing tool that hooks into LangGraph with one line.

Set three env vars (put them in .env):

LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.com

Then enable the callback:

from langfuse import get_client
from langfuse.langchain import CallbackHandler

langfuse = get_client()
handler = CallbackHandler()

result = agent.invoke(
    {"messages": [HumanMessage(content="What is the weather in Jakarta?")]},
    config={"callbacks": [handler]},
)

Open the Langfuse dashboard. You will see a trace with three spans: one for the model call, one for the tool execution, one for the final model call. Each span shows the prompt, the completion, token usage, latency, and cost.

That single trace turns "the agent gave a weird answer" into "the second model call hallucinated the temperature." Same data, but you can actually act on it.

Step 6: Add an Evaluation Smoke Test

Tracing is reactive. You also want a quick check that the graph still behaves correctly after you change a prompt or swap a model. Langfuse ships a dataset-and-runner setup for this.

from langfuse import get_client

langfuse = get_client()
dataset = langfuse.create_dataset(name="weather-agent-smoke")

dataset.create_item(
    input={"messages": [HumanMessage(content="Weather in Tokyo?")]},
    expected_output="Tokyo",
)

def run_agent(item):
    return agent.invoke(item.input, config={"callbacks": [handler]})

result = langfuse.run_evaluation(
    dataset_name="weather-agent-smoke",
    task=run_agent,
    evaluators=[],  # add a custom evaluator for stricter checks
)

Wire this into CI and you catch prompt regressions before they reach users. The evaluators=[] argument is a list of callables; pass any function that takes a (input, output, expected_output) tuple and returns a score.

When to Use LangGraph vs Other Options

LangGraph is the right choice when you need explicit control over flow: loops, branches, human approval steps, persistence. The state object is the source of truth, and you can inspect it mid-execution with a checkpointer.

Skip it for simple single-turn tool use. LangChain's create_tool_calling_agent is shorter and runs the same model calls without the graph boilerplate. Reach for LangGraph when the demo stops fitting on one turn.

For multi-agent systems where agents hand off to each other, LangGraph's subgraph pattern is cleaner than CrewAI's role-based setup. The cost is more code; the benefit is that every handoff is visible in the trace.

Common Pitfalls

Forgetting the reducer. If you declare messages: list[BaseMessage] without Annotated[..., add_messages], every node replaces the list. The graph still runs. The agent just forgets everything after step one.

Returning the wrong shape from a node. A node must return a dict whose keys are fields in the state. Returning a bare list or a single message works for some nodes and silently breaks others. Always return {"messages": [...]}.

Tracing only the first call. If you build the agent once and reuse it across requests, register the callback inside your request handler, not at module load. Module-level callbacks can leak state between users.

Mixing async and sync nodes. If you call ainvoke on the graph, every node function must be async. A sync call_model in an async graph will block the event loop. Pick one style and stick to it.

Where to Go Next

The four directions worth your time, in order of payoff:

  • Add a MemorySaver checkpointer so the agent remembers across requests in the same session.

  • Swap ChatOpenAI for a local model through Ollama to cut costs on simple queries.

  • Connect the Langfuse trace to your alerting stack when token cost per trace crosses a threshold.

  • Replace the tool stub with a real API and add an evaluator that checks tool input shape.

Need Help Implementing This?

I help teams design and build scalable cloud infrastructure, DevOps pipelines, and production-grade systems.

Book a Free Consultation