LangGraph

LangGraph builds stateful multi-actor agents on top of LangChain. Because LangGraph runs through LangChain's Runnable interface, Trefur's LangChain callback handler captures every node invocation + tool call + LLM completion automatically.

Install

pip install trefur-observe langgraph langchain-anthropic

Setup

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from trefur_observe import TrefurObserve, patch_langchain

# Initialise observe once at startup.
TrefurObserve.init(
    api_key=os.environ["TREFUR_API_KEY"],
    agent={"name": "support-triage-graph", "framework": "langgraph"},
)
# LangGraph uses LangChain under the hood — the LangChain patch sees
# every node call + tool invocation.
patch_langchain(TrefurObserve.get_instance())

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

llm = ChatAnthropic(model="claude-sonnet-4-20250514")

def classify(state: AgentState):
    return {"messages": [llm.invoke(state["messages"])]}

graph = StateGraph(AgentState)
graph.add_node("classify", classify)
graph.set_entry_point("classify")
graph.add_edge("classify", END)
app = graph.compile()

result = app.invoke({"messages": [{"role": "user", "content": "I lost my password"}]})

What gets traced

  • Node executions — each function node renders as a step with name, inputs, and outputs.
  • Conditional edges — the branch taken is recorded as route_to metadata.
  • Tool calls — every @tool-decorated function or ToolNode invocation gets tool_call step type.
  • LLM completions — model + token counts + prompt + completion text (subject to your redaction level).
  • Checkpoints — checkpointer save events are tagged with the thread id.

Common use case — human-in-the-loop graphs

LangGraph's interrupt_before + interrupt_aftergive you human checkpoints. Trefur records the interrupt as a step with status awaiting_approval so you can build alerts when a checkpoint is bypassed or expires.

Common pitfalls

  • Async + streaming. If you use app.astream, the auto-instrumentation still wraps the run but you must await the final state before the process exits — otherwise the buffer flush is cancelled. Add TrefurObserve.get_instance().flush() at the end of the request if your event loop is shutting down.
  • Multi-thread graphs. Pass thread_id in config={"configurable": {"thread_id": ...}}so checkpoints attribute to the right trace. Trefur reads the thread id from the LangChain context.

See the Python SDK reference for full patch_langchain options, or browse all integrations.