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_live_...
VERIPROOF_APPLICATION_ID=my-chatbot
ANTHROPIC_API_KEY=sk-ant-...

Complete example


import asyncio
import os
from typing import Annotated
from typing_extensions import TypedDict
 
from veriproof_sdk import (
    configure_veriproof,
    VeriproofClientOptions,
    SessionIntent,
    SessionOutcome,
    StepType,
    StepOutcome,
    RiskLevel,
)
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",
    application_id=os.environ["VERIPROOF_APPLICATION_ID"],
    session_intent=SessionIntent.CONVERSATIONAL,
)
 
 
async def main():
    # 5. Invoke the graph — a VeriProof session is created automatically
    result = await graph.ainvoke(
        {"messages": [("user", "What are the key risks of high debt-to-income ratios?")]},
        session_name="chatbot-turn-1",
    )
 
    print("Response:", result["messages"][-1].content)
    print("Session ID:", graph.last_session_id)
    print("Merkle root:", graph.last_merkle_root)
 
 
asyncio.run(main())

What you’ll see in VeriProof

The span hierarchy mirrors the graph structure:


session: chatbot-turn-1
  └── agent_run SimpleChatbot          (graph root span)
        └── node chatbot               (per-node span)
              └── ChatAnthropic        (model call)

Span	Key attributes
`session chatbot-turn-1`	`veriproof.session.intent=CONVERSATIONAL`, `gen_ai.system=langgraph`
`agent_run SimpleChatbot`	`veriproof.graph.name=SimpleChatbot`
`node chatbot`	`veriproof.node.name=chatbot`, execution time
`ChatAnthropic`	`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


from veriproof_sdk import VeriproofSession, SessionIntent, SessionOutcome
 
async def multi_turn():
    async with VeriproofSession(
        application_id=os.environ["VERIPROOF_APPLICATION_ID"],
        session_name="customer-support-chat",
        intent=SessionIntent.CONVERSATIONAL,
    ) as session:
        messages = []
 
        for user_message in ["Hi", "What are your loan rates?", "Thanks, goodbye"]:
            messages.append(("user", user_message))
 
            result = await graph.ainvoke(
                {"messages": messages},
                # Pass the active VeriProof session so graph spans nest within it
                config={"callbacks": [session.langchain_callback()]},
            )
            assistant_msg = result["messages"][-1]
            messages.append(("assistant", assistant_msg.content))
            print(f"Bot: {assistant_msg.content}\n")
 
        session.set_outcome(SessionOutcome.SUCCESS)

Next steps

Multi-Turn Workflow example — orchestrate sequential agents
Pydantic AI example — simpler single-agent pattern
Python SDK Core reference — full session builder API