LangGraph Chatbot
Instrument a LangGraph state-machine chatbot with automatic per-node span capture. Each node invocation becomes a child span under the VeriProof session, giving you full trace visibility into the graph’s execution.
PythonLangGraphPython ≥ 3.10
Prerequisites
pip install veriproof-sdk veriproof-sdk-instrumentation-langgraph langgraph langchain-anthropicEnvironment
VERIPROOF_API_KEY=vp_cust_your-tenant.<key-material>
VERIPROOF_APPLICATION_ID=my-chatbot
ANTHROPIC_API_KEY=sk-ant-...Complete example
import asyncio
import os
import uuid
from typing import Annotated
from typing_extensions import TypedDict
from veriproof_sdk import (
configure_veriproof,
VeriproofClientOptions,
SessionIntent,
)
from veriproof_sdk_instrumentation_langgraph import InstrumentedCompiledStateGraph
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
# 1. Configure VeriProof once at startup
configure_veriproof(
VeriproofClientOptions(
api_key=os.environ["VERIPROOF_API_KEY"],
application_id=os.environ["VERIPROOF_APPLICATION_ID"],
),
service_name=os.environ["VERIPROOF_APPLICATION_ID"],
set_global=True,
)
# 2. Define the LangGraph state and chatbot node
class State(TypedDict):
messages: Annotated[list, add_messages]
llm = ChatAnthropic(model="claude-3-5-haiku-20241022")
async def chatbot_node(state: State) -> State:
response = await llm.ainvoke(state["messages"])
return {"messages": [response]}
# 3. Build the graph
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.set_entry_point("chatbot")
builder.add_edge("chatbot", END)
raw_graph = builder.compile()
# 4. Wrap the compiled graph with VeriProof instrumentation
graph = InstrumentedCompiledStateGraph(raw_graph, graph_name="SimpleChatbot")
async def main():
# 5. Use a stable session ID so spans link to a VeriProof session
session_id = str(uuid.uuid4())
result = await graph.ainvoke(
{"messages": [("user", "What are the key risks of high debt-to-income ratios?")]},
config={"configurable": {"thread_id": session_id}},
)
print("Response :", result["messages"][-1].content)
print("Session ID:", session_id)
asyncio.run(main())What you’ll see in VeriProof
The span hierarchy mirrors the graph structure:
agent_run SimpleChatbot (veriproof.session.id set from thread_id)
└── node chatbot (per-node span from callback handler)
└── chat claude-3-5-haiku-20241022| Span | Key attributes |
|---|---|
agent_run SimpleChatbot | veriproof.session.id, gen_ai.system=langgraph, langgraph.node.count |
node chatbot | langgraph.node.name=chatbot, veriproof.agent.role=primary |
chat claude-3-5-haiku-20241022 | gen_ai.request.model, gen_ai.usage.input_tokens |
Every ainvoke call creates a new VeriProof session. For multi-turn conversations where you want a single session to span multiple turns, manage the session manually with VeriproofSession and pass the session context to the graph via the config dict.
Multi-turn conversations under a single session
import uuid
async def multi_turn():
# Reuse the same session_id across turns — all agent_run spans share
# the same veriproof.session.id in VeriProof.
session_id = str(uuid.uuid4())
messages = []
for user_message in ["Hi", "What are your loan rates?", "Thanks, goodbye"]:
messages.append(("user", user_message))
result = await graph.ainvoke(
{"messages": messages},
config={"configurable": {"thread_id": session_id}},
)
assistant_msg = result["messages"][-1]
messages.append(("assistant", assistant_msg.content))
print(f"Bot: {assistant_msg.content}\n")
print(f"All turns linked under session: {session_id}")
asyncio.run(multi_turn())Next steps
- Multi-Turn Workflow example — orchestrate sequential agents
- Pydantic AI example — simpler single-agent pattern
- Python SDK Core reference — full session builder API
Last updated on