LangGraph Integration Guide

Integrate ACGP with LangGraph stateful graphs for node-level governance.

---

Overview

LangGraph builds workflows from nodes and shared state. This foundation-maintained community adapter is strongest when you wrap nodes before compile(). A compiled-graph wrapper is available as a fallback when the compiled graph still exposes a governable nodes mapping.

---

Installation

PythonTypeScript

pip install acgp-langgraph

For production runtime posture in the underlying Python steward:

pip install "acgp-sdk[postgres]"

npm install @acgp-protocol/langgraph @acgp-protocol/sdk

---

Recommended Pattern: Wrap Nodes Before compile()

PythonTypeScript

from typing import TypedDict
from langgraph.graph import END, StateGraph

from acgp import GovernanceSteward, PostgresStateStorage
from acgp_langgraph import GovernedNode


class MyState(TypedDict):
    input: str
    result: str


def process_node(state: MyState) -> MyState:
    return {"result": f"processed {state['input']}"}


steward = GovernanceSteward.production(
    blueprint_file="blueprint.yaml",
    state_storage=PostgresStateStorage(connection_string="postgresql://runtime/acgp"),
)

workflow = StateGraph(MyState)
workflow.add_node(
    "process",
    GovernedNode(
        process_node,
        steward,
        node_name="process",
        agent_id="urn:acgp:agent:assistant:prod:7f4c9d2a",
        agent_name="assistant",
    ),
)
workflow.set_entry_point("process")
workflow.add_edge("process", END)
graph = workflow.compile()

result = graph.invoke({"input": "data"})

TypeScript Runtime Coming Soon

The TypeScript GovernanceSteward runtime is planned for a future release.

---

Compiled Graph Wrapper

Use GovernedGraph or AsyncGovernedGraph only when your compiled graph still exposes a mutable nodes mapping.

from acgp_langgraph import GovernedGraph

compiled = workflow.compile()
governed_graph = GovernedGraph(
    compiled,
    steward,
    agent_id="urn:acgp:agent:assistant:prod:7f4c9d2a",
    agent_name="assistant",
)
result = governed_graph.invoke({"input": "data"})

If the compiled graph does not expose mutable nodes, the wrapper raises a configuration error instead of silently skipping governance.

Use the general GovernanceSteward(...) constructor for local development or advanced bootstrap flows. Production bootstrap should use GovernanceSteward.production(...).

---

Production-oriented behavior

Node-Level Governance As The Default

The safe default is wrapping individual nodes before compile(). That avoids depending on compiled-graph internals that can change across LangGraph releases.

State Preservation

from acgp_langgraph.state import add_acgp_state_fields

class MyState(TypedDict):
    messages: list[str]

MyState = add_acgp_state_fields(MyState)

Recorded fields include trace IDs, intervention history, evaluation counts, the last evaluated trace, trust-debt totals when available, extracted reasoning, pending HITL / appeal requests, and the last governed node error.

Async Support

from acgp_langgraph import AsyncGovernedNode

async def analyze_node(state):
    ...

workflow.add_node(
    "analyze",
    AsyncGovernedNode(
        analyze_node,
        steward,
        node_name="analyze",
        agent_id="urn:acgp:agent:assistant:prod:7f4c9d2a",
    ),
)

Error Tracking

If a governed node raises, the adapter records structured error details into shared state before the exception is re-raised.

Pending Recourse Helpers

result = governed_graph.complete_hitl_review(
    request_id="hitl-123",
    trace_id="trace-123",
    outcome="approve",
    operator_id="reviewer-1",
    justification="Verified by human reviewer",
)

The same helper surface is available for operator appeals via complete_operator_appeal().

Stream Execution

for step in governed_graph.stream({"input": "data"}):
    print(step)

---

Exception Handling

from acgp_langgraph import GraphGovernanceError, InterventionError

try:
    result = graph.invoke({"input": "sensitive data"})
except InterventionError as exc:
    print(exc.message)
except GraphGovernanceError as exc:
    print(f"Integration misconfigured: {exc.message}")

---

Notes

• The adapter requires a stable agent_id and fails closed if it is missing.
• Wrapping nodes before compile() is the preferred path because it does not depend on compiled-graph internals.
• Local runtime evaluation, pending HITL state, and appeal completion still come from the core Python GovernanceSteward implementation.

---

Examples

Example implementations available in the acgp-langgraph package:

• Basic Graph: Simple stateful graph with governed nodes
• Async Graph: Async node governance with AsyncGovernedNode
• Multi-Agent Graph: Multiple governed nodes in one workflow