Skip to Content
ExamplesPythonLangGraph Chatbot

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-anthropic

Environment

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
SpanKey attributes
agent_run SimpleChatbotveriproof.session.id, gen_ai.system=langgraph, langgraph.node.count
node chatbotlanggraph.node.name=chatbot, veriproof.agent.role=primary
chat claude-3-5-haiku-20241022gen_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

Last updated on