Skip to main content

​
LangGraph

The LangGraph adapter wraps a compiled graph so every node execution is recorded as a governance event automatically. No per-node instrumentation, no callback wiring — one function call and you’re done.

​
Install

pip install "proofrail[langgraph]"
This installs langgraph if not already present.
Supported LangGraph versions: 0.2.0 - 0.3.x. Install with a LangGraph version in this range; the version constraint is pinned in sdk/pyproject.toml.

​
Quick start

import asyncio
import proofrail
from proofrail.langgraph import govern

from langgraph.graph import StateGraph
from langchain_core.messages import HumanMessage

proofrail.init(api_key="prail_...")

# Build your graph as usual
graph = StateGraph(...)
graph.add_node("research", research_node)
graph.add_node("summarize", summarize_node)
graph.add_edge("research", "summarize")
compiled = graph.compile()

# Wrap it once with govern()
governed = govern(compiled, chain_name="research-workflow")

# Use governed exactly like the original graph
result = await governed.ainvoke({"messages": [HumanMessage(content="Q3 revenue summary")]})
governed has the same .invoke / .ainvoke signatures as the original graph. If any node is denied by policy, ActionDeniedError propagates out and the graph halts at that node.

​
How it works

The adapter tries two strategies in order: Strategy A: astream_events (preferred, LangGraph 0.1+) Consumes the graph’s event stream, captures on_chain_start and on_chain_end events for each node, and fires governance recording for each. The graph’s final output is returned unchanged. Strategy B: LangChain callback fallback If astream_events isn’t available, an AsyncCallbackHandler is injected via the LangGraph config. Same node-level coverage, slightly different code path. You don’t have to choose. The adapter detects which is available and uses it.

​
What gets recorded

For each node execution:
  • node_execution event when the node starts, with the input state as the payload
  • node_result event when the node completes successfully, with the output state
  • node_error event if the node raises, with the exception type and message
Internal LangGraph nodes (__start__, __end__) are filtered out. Only user-defined nodes appear in your audit trail.

​
Policy enforcement

Policy decisions arrive between nodes. If a node’s action is denied: 
from proofrail.exceptions import ActionDeniedError

try:
    result = await governed.ainvoke(initial_state)
except ActionDeniedError as exc:
    print(f"Workflow halted: {exc.policy_name}")
    print(f"Reason: {exc.condition}")
    print(f"Remediation: {exc.remediation}")
The graph halts at the denied node. Earlier nodes that ran successfully are already in the audit trail. For approval-required decisions, record_agent_action blocks until the approver responds. The graph appears to “pause” at that node — your await governed.ainvoke(...) is waiting for the human. On approval the graph resumes; on denial or timeout it raises.

​
Configuring the chain

Pass chain_name and optional metadata: 
governed = govern(
    compiled,
    chain_name="customer-onboarding",
    metadata={
        "team": "platform",
        "version": "2.1.0",
        "environment": "production",
    },
)
chain_name appears in your dashboard and identifies the workflow type. metadata is arbitrary key/value pairs attached to every chain run by this governed graph.

​
Passing LangGraph config

Standard LangGraph config (callbacks, recursion limits, run names) is preserved through the wrapper: 
result = await governed.ainvoke(
    initial_state,
    config={
        "recursion_limit": 50,
        "callbacks": [my_other_callback],
        "run_name": "monthly-batch",
    },
)
ProofRail’s callback is appended to whatever you pass; your existing callbacks continue to fire.

​
Sync invocation

If your code is synchronous: 
result = governed.invoke(initial_state)
Behaves the same way await governed.ainvoke(initial_state) does, but blocking. Raises RuntimeError if called from inside a running event loop (use ainvoke in async contexts).

​
State extraction

The adapter captures the input state passed to each node and the output state returned. State is converted to a plain dict for the audit payload:
  • dict states pass through unchanged
  • Pydantic v2 models call .model_dump()
  • Pydantic v1 models call .dict()
  • NamedTuples call ._asdict()
  • Anything else falls back to dict(state) or str(state) (truncated)
Sensitive fields in state are sanitized using the patterns configured in proofrail.init(). See Configuration Reference for sanitization options.

​
Edge cases

Conditional edges. Routing functions that decide which node runs next don’t produce governance events themselves — only the actual node executions do. The audit trail reflects which path was taken. Subgraphs. Nested graphs work, but each subgraph’s nodes appear as events in the parent chain. There’s currently no automatic parent-child relationship between subgraphs and their containing chain. Streaming. astream and astream_events on the governed wrapper work but the streaming behavior is not customized for governance — events are recorded as they happen, but the stream output is unchanged from the underlying graph. Tool calling within nodes. If your nodes use LangChain tools internally, those tool calls are NOT separately recorded by the LangGraph adapter — only the node-level inputs and outputs are. To record tool calls explicitly, use the LangChain adapter on the underlying executor instead, or record actions manually with the framework-agnostic Chain.

​
Where to go next

LangChain adapter

For non-graph LangChain agents and chains.

MCP adapter

Govern MCP server tool calls.

Human approval guide

How approval gates interact with graph execution.

SDK reference

The complete govern() signature and options.