LangGraph Integration

Full observability and governance for LangGraph stateful agents with automatic node tracing.

langgraph >= 0.0.20StateGraphMessageGraph

Installation

Terminal
pip install turingpulse_sdk turingpulse_sdk_langgraph langgraph

Quick Start

1. Initialize TuringPulse

config.py
from turingpulse_sdk import init, TuringPulseConfig

init(TuringPulseConfig(
    api_key="sk_live_your_api_key",
    workflow_name="my-project",
))

2. Instrument Your Graph

graph.py
from langgraph.graph import StateGraph, END
from turingpulse_sdk_langgraph import instrument_langgraph
from typing import TypedDict

# Define your state
class AgentState(TypedDict):
    messages: list
    next_step: str

# Create your graph
graph = StateGraph(AgentState)

# Add nodes
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# Add edges
graph.add_edge("agent", "tools")
graph.add_conditional_edges("tools", should_continue)

# Compile the graph
compiled = graph.compile()

# Instrument with TuringPulse
instrumented_graph = instrument_langgraph(
    compiled,
    name="langgraph-agent",
    labels={"framework": "langgraph"},
)

3. Run Your Agent

run.py
# Run the instrumented graph
result = instrumented_graph.invoke({
    "messages": [{"role": "user", "content": "What's the weather?"}],
})

# Traces are automatically sent to TuringPulse
# Each node execution is captured as a span

What Gets Traced

The LangGraph integration automatically captures:

  • Graph Execution — Full trace for each invoke/stream
  • Node Spans — Each node execution as a child span
  • State Transitions — State before and after each node
  • Conditional Edges — Which path was taken
  • Tool Calls — Tool invocations within nodes
  • LLM Calls — Nested LLM calls with tokens
  • Errors — Exceptions with stack traces
ℹ️
Automatic LLM Tracing
If you're using LangChain LLMs within your nodes, they're automatically traced as nested spans.

With Governance

governance.py
from turingpulse_sdk import GovernanceDirective
from turingpulse_sdk_langgraph import instrument_langgraph

instrumented_graph = instrument_langgraph(
    compiled,
    name="langgraph-agent",
    governance=GovernanceDirective(
        hitl=True,
        hatl=True,
        reviewers=["admin@company.com"],
        escalation_channels=["slack://approvals"],
        severity="high",
    ),
)

With KPIs

kpis.py
from turingpulse_sdk import KPIConfig
from turingpulse_sdk_langgraph import instrument_langgraph

instrumented_graph = instrument_langgraph(
    compiled,
    name="langgraph-agent",
    kpis=[
        KPIConfig(
            kpi_id="latency_ms",
            use_duration=True,
            alert_threshold=10000,
            comparator="gt",
        ),
        KPIConfig(
            kpi_id="node_count",
            value=lambda ctx: len(ctx.metadata.get("nodes_executed", [])),
            alert_threshold=20,
            comparator="gt",
        ),
    ],
)

Streaming Support

streaming.py
# Streaming is fully supported
for event in instrumented_graph.stream({
    "messages": [{"role": "user", "content": "Hello"}],
}):
    print(event)
    # Each event is captured in the trace

# Async streaming
async for event in instrumented_graph.astream({
    "messages": [{"role": "user", "content": "Hello"}],
}):
    print(event)

Full Configuration

full-config.py
from turingpulse_sdk import GovernanceDirective, KPIConfig
from turingpulse_sdk_langgraph import instrument_langgraph

run = instrument_langgraph(
    compiled,
    name="langgraph-agent",
    model="gpt-4o",
    provider="openai",
    governance=GovernanceDirective(
        hatl=True,
        reviewers=["admin@company.com"],
    ),
    kpis=[
        KPIConfig(kpi_id="latency_ms", use_duration=True, alert_threshold=10000),
    ],
    metadata={"team": "ml-ops"},
)

Next Steps