LangChain Integration Guide¶

Integrate ACGP with LangChain for governance of chains, tools, and modern Runnable-based agents.

Overview¶

ACGP integrates with LangChain through callback hooks. This foundation-maintained community adapter is tuned around the supported LangChain 1.2 Runnable surface while preserving a compatibility path for classic run() chains.

Installation¶

pip install acgp-langchain

For production runtime posture in the underlying Python steward:

pip install "acgp-sdk[postgres]"

For legacy chain projects that still rely on the classic package split:

pip install "acgp-langchain[classic]"

Basic Usage¶

from langchain.agents import create_agent

from acgp import GovernanceSteward, PostgresStateStorage
from acgp_langchain import GovernedChain

agent = create_agent("openai:gpt-5", tools=[your_tool])

steward = GovernanceSteward.production(
    blueprint_file="blueprint.yaml",
    state_storage=PostgresStateStorage(connection_string="postgresql://runtime/acgp"),
)
governed_agent = GovernedChain(
    agent,
    steward,
    agent_id="urn:acgp:agent:assistant:prod:7f4c9d2a",
    agent_name="assistant",
)

result = governed_agent.invoke(
    {"messages": [{"role": "user", "content": "Process this request"}]}
)

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

Production-oriented behavior¶

Stable principal required: the adapter fails closed if agent_id is missing.
Runnable-first execution: invoke, ainvoke, stream, astream, batch, and abatch are supported.
Per-run callback bookkeeping: callback reasoning and result caches are keyed by LangChain run IDs with bounded retention.
Context-local override mode: override flows use execution-local state instead of mutating the shared callback object.
Payload hygiene: reasoning and tool parameters are normalized and truncated before submitting a trace.
Direct config export: callback_config() returns a Runnable-compatible config block you can merge into your own orchestration.
Recourse surface: the wrapper exposes get_hitl_request(), get_appeal_request(), complete_hitl_review(), and complete_operator_appeal().

Advanced Features¶

Override Raised Interventions¶

result = governed_agent.invoke_with_override(
    trace_id="blocked_trace_id",
    input={"messages": [{"role": "user", "content": "Previously blocked query"}]},
)

Batch Execution¶

results = governed_agent.batch([
    {"messages": [{"role": "user", "content": "Request A"}]},
    {"messages": [{"role": "user", "content": "Request B"}]},
])

Get Governance Metrics¶

metrics = governed_agent.get_metrics()
print(metrics)

Read Pending Recourse¶

try:
    governed_agent.invoke({"messages": [{"role": "user", "content": "Risky query"}]})
except Exception:
    pass

recourse = governed_agent.get_last_recourse()
if recourse and recourse["kind"] == "human_review":
    pending = governed_agent.get_hitl_request(recourse["request_id"])

Complete HITL Review¶

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

Complete Operator Appeal¶

result = governed_agent.complete_operator_appeal(
    request_id="appeal-123",
    trace_id="trace-123",
    outcome="allow",
    operator_id="reviewer-2",
    justification="Exception approved by operator",
)

Exception Handling¶

from acgp_langchain import InterventionError

try:
    result = governed_agent.invoke(
        {"messages": [{"role": "user", "content": "Risky query"}]}
    )
except InterventionError as exc:
    print(f"Action blocked: {exc.message}")
    print(f"Trace ID: {exc.trace_id}")

Direct Callback Wiring¶

from acgp_langchain import ACGPCallback

callback = ACGPCallback(
    steward,
    agent_id="urn:acgp:agent:assistant:prod:7f4c9d2a",
    agent_label="assistant",
    raise_on_intervention=True,
)

result = chain.invoke(
    {"messages": [{"role": "user", "content": "Query"}]},
    config={"callbacks": [callback]},
)

Or export a prebuilt config from the wrapper:

config = governed_agent.callback_config({"tags": ["prod"]})
result = agent.invoke({"messages": [...]}, config=config)

Notes¶

The adapter requires a stable agent_id and fails closed if it is missing.
Governance is strongest on components that emit standard LangChain tool and agent callback events.
Local runtime evaluation, pending HITL state, and appeal completion still come from the core Python GovernanceSteward implementation.

LangGraph Integration