LangGraph

The LangGraph integration wraps a compiled graph in an Arclasp Chain and records supported node lifecycle events through the execution hooks exposed by LangGraph.

Install

pip install "arclasp[langgraph]"

The current package constraint is langgraph>=0.2.0,<0.4.0.

Add Arclasp to an existing graph

The following is an integration fragment. It assumes your application already builds and compiles a LangGraph graph.

import arclasp
from arclasp.langgraph import govern

arclasp.init(
    api_key="prail_...",
    backend_url="https://api.arclasp.com",
)

compiled = graph.compile()

governed = govern(
    compiled,
    chain_name="research-workflow",
    metadata={"environment": "production"},
)

result = await governed.ainvoke(initial_state)

The wrapper preserves the graph’s ordinary invocation interface while associating the run with an Arclasp Chain.

How it works

The current adapter supports two integration paths depending on what the wrapped graph/runtime exposes:

LangGraph event streaming through astream_events(version="v2");
callback injection through LangChain-compatible async callbacks as a fallback.

The integration records supported node lifecycle activity into the containing Arclasp Chain.

Consequential side effects

Do not assume that generic node observation is a universal pre-side-effect barrier for arbitrary code inside a node. When a node contains an irreversible operation that must not run until Arclasp has allowed or approved it, instrument that exact boundary inside the node:

async def send_payment_node(state):
    async with arclasp.Chain("payment-boundary") as chain:
        await chain.record_agent_action(
            agent_name="payment-agent",
            action_type="tool_call",
            action_name="send_payment",
            payload={"amount_usd": state["amount_usd"]},
        )

        # Customer code executes only after governance resolves successfully.
        result = await send_payment(state["amount_usd"])

    return {"payment_result": result}

If the broader graph is already wrapped with govern(...), choose your Chain boundaries deliberately so the evidence model matches the workflow you actually want to govern. Do not create duplicate instrumentation merely to increase event counts.

Policy errors

Policy and availability errors raised by Arclasp can propagate out of the governed graph.

try:
    result = await governed.ainvoke(initial_state)
except arclasp.ActionDeniedError as exc:
    print(f"Governed workflow blocked: {exc}")

Earlier governed evidence remains associated with the Chain according to the normal Chain lifecycle.

Sync and async

Prefer:

result = await governed.ainvoke(initial_state)

Synchronous invocation is also supported for synchronous applications:

result = governed.invoke(initial_state)

Do not call the synchronous wrapper from inside an already running async event loop.

Boundaries

Conditional routing functions are not automatically consequential governance points.
Tool calls nested inside a node are not automatically separate governed tool events unless you instrument them separately.
Nested/subgraph execution does not imply a separate automatic parent-child Arclasp Chain hierarchy.
Streaming behavior remains primarily the underlying graph’s behavior; Arclasp does not redefine the stream protocol.

Get Started

Concepts

Framework Integration

Guides

Other

LangGraph

LangGraph

Install

Add Arclasp to an existing graph

How it works

Consequential side effects

Policy errors

Sync and async

Boundaries

​LangGraph

​Install

​Add Arclasp to an existing graph

​How it works

​Consequential side effects

​Policy errors

​Sync and async

​Boundaries

LangGraph

Install

Add Arclasp to an existing graph

How it works

Consequential side effects

Policy errors

Sync and async

Boundaries