> ## Documentation Index
> Fetch the complete documentation index at: https://docs.proofrail.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# LangGraph

> Govern compiled LangGraph workflows with one line

# 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

```bash theme={null}
pip install "proofrail[langgraph] @ git+https://github.com/TOAAiV/proofrail"
```

This installs `langgraph` if not already present.

<Note>
  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`.
</Note>

## Quick start

```python theme={null}
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:

```python theme={null}
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`:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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](/reference/configuration) 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](/frameworks/langchain) on the underlying executor instead, or record actions manually with the [framework-agnostic Chain](/quickstart).

## Where to go next

<CardGroup cols={2}>
  <Card title="LangChain adapter" icon="link" href="/frameworks/langchain">
    For non-graph LangChain agents and chains.
  </Card>

  <Card title="MCP adapter" icon="plug" href="/frameworks/mcp">
    Govern MCP server tool calls.
  </Card>

  <Card title="Human approval guide" icon="user-check" href="/guides/human-approval">
    How approval gates interact with graph execution.
  </Card>

  <Card title="SDK reference" icon="code" href="/reference/sdk-api">
    The complete `govern()` signature and options.
  </Card>
</CardGroup>
