LangGraph Integration Guide¶

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

Overview¶

LangGraph uses stateful graphs with nodes and edges to build complex agent workflows. ACGP provides governance at the node execution level, giving you fine-grained control over graph behavior.

Installation¶

PythonTypeScript

pip install acgp-langgraph

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

Basic Usage¶

PythonTypeScript

from langgraph.graph import StateGraph
from acgp_langgraph import GovernedGraph
from acgp import GovernanceSteward

# Your LangGraph graph
workflow = StateGraph(MyState)
workflow.add_node("process", process_node)
workflow.add_edge("__start__", "process")
graph = workflow.compile()

# Add governance
steward = GovernanceSteward(blueprint_file="blueprint.yaml")
governed_graph = GovernedGraph(graph, steward)

# Execute with governance
result = governed_graph.invoke({"input": "data"})

TypeScript Runtime Coming Soon

The TypeScript GovernanceSteward runtime is planned for a future release. The code below shows the planned API. Use the Python SDK for full runtime support.

import { StateGraph } from '@langchain/langgraph';
import { GovernedGraph } from '@acgp-protocol/langgraph';
import { GovernanceSteward } from '@acgp-protocol/sdk';

// Your LangGraph graph
const graph = new StateGraph<MyState>()
  .addNode('process', processNode)
  .addEdge('__start__', 'process')
  .compile();

// Add governance (planned API)
const steward = new GovernanceSteward({ blueprintFile: 'blueprint.yaml' });
const governedGraph = new GovernedGraph({ graph, steward });

// Execute with governance
const result = await governedGraph.invoke({ input: 'data' });

Features¶

Node-Level Governance¶

Evaluate each node before execution:

# Governance checks happen automatically before each node runs
result = governed_graph.invoke({"input": "sensitive data"})

State Tracking¶

Track interventions in graph state:

from acgp_langgraph.state import add_acgp_state_fields

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

# Add ACGP tracking fields
MyState = add_acgp_state_fields(MyState)

# Access intervention history from state
result = governed_graph.invoke({"input": "data"})
print(result['acgp_interventions'])  # List of interventions
print(result['acgp_trust_debt'])  # Current trust debt

Async Support¶

Full async/await support:

from acgp_langgraph import AsyncGovernedGraph

async_governed = AsyncGovernedGraph(async_graph, steward)
result = await async_governed.ainvoke({"input": "data"})

Individual Node Governance¶

Govern specific nodes only:

from acgp_langgraph import GovernedNode

critical_node = GovernedNode(
    node_func=my_critical_function,
    steward=steward,
    node_name="critical_operation"
)

workflow.add_node("critical_operation", critical_node)

Stream Execution¶

# Stream graph execution with governance
for step in governed_graph.stream({"input": "data"}):
    print(f"Step: {step}")

Advanced Patterns¶

Multi-Agent Graphs¶

def researcher_node(state):
    return {"research": "findings"}

def writer_node(state):
    return {"content": "article"}

workflow = StateGraph(MultiAgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_edge("researcher", "writer")

governed_graph = GovernedGraph(workflow.compile(), steward)

Conditional Edges with Governance¶

def route_decision(state):
    return "approve" if state['score'] > 0.8 else "review"

workflow.add_conditional_edges(
    "evaluate",
    route_decision,
    {"approve": "execute", "review": "human_review"}
)

# Governance applies to all paths
governed_graph = GovernedGraph(workflow.compile(), steward)

Advanced Features¶

Override Blocked Nodes¶

# After human approval
result = governed_graph.invoke_with_override(
    trace_id="blocked_trace_id",
    input={"input": "data"}
)

Get Governance Metrics¶

metrics = governed_graph.get_metrics()
print(f"Total interventions: {metrics['total_interventions']}")

Exception Handling¶

from acgp_langgraph import InterventionError, GraphGovernanceError

try:
    result = governed_graph.invoke({"input": "sensitive data"})
except InterventionError as e:
    print(f"Node blocked: {e.message}")
    print(f"Trace ID: {e.trace_id}")

Control State Tracking¶

# Disable state tracking for performance
governed_graph = GovernedGraph(
    graph,
    steward,
    track_in_state=False  # Don't store interventions in state
)

Using State Utilities¶

from acgp_langgraph.state import (
    initialize_acgp_state,
    record_intervention_in_state,
    ACGPStateAnnotation
)

# Initialize state manually
state = initialize_acgp_state({"input": "data"})

# Access ACGP fields
trace_ids = state['acgp_trace_ids']
interventions = state['acgp_interventions']
trust_debt = state['acgp_trust_debt']

Configuration¶

Policy Example¶

acl_tier: ACL-2

rules:
  - condition: "action == 'critical_operation'"
    intervention: escalate
    message: "Critical operation requires approval"

tripwires:
  - name: max_iterations
    threshold: 10
    action: halt

Best Practices¶

Node Granularity

Keep nodes focused on single responsibilities for better governance control.

State Management

Use ACGP state fields to track governance history across node executions.

Performance

Governance adds ~50-100ms per node. Use governance contracts for latency budgets.

Examples¶

Example implementations available in the acgp-langgraph package:

Basic Graph: Simple stateful graph with governance
Async Graph: Async/await graph execution
Multi-Agent Graph: Researcher, writer, and reviewer workflow
State Tracking: Using ACGP state fields for history

Next: AGNO Integration