为 LangGraph 智能体添加操作前检查

检查的作用

检查返回 permit、 deny或 escalate ，并附上决定结果的条款和签名回执。检查仅提供建议，真正拒绝执行的是图路由。托管检查在内存中评估请求，不保留请求内容。

代码

这是经过审计的示例，与以下位置发布的内容完全一致： public/examples/frameworks/.

"""LangGraph route that checks HANRIA before its execution node."""

from __future__ import annotations

import json
from typing import Any, TypedDict
from urllib.error import URLError
from urllib.request import Request, urlopen

from langgraph.graph import END, START, StateGraph

CHECK_URL = "https://check.hanria.ai/v1/check"
MANDATE = {
    "schema_version": "0.2-draft",
    "mandate_id": "read-text-files",
    "purpose": "Permit reads of text files under the example directory.",
    "default": "deny",
    "clauses": [
        {
            "id": "permit-example-read",
            "effect": "permit",
            "match": {
                "kind": ["file"],
                "verb": ["read"],
                "target_prefix": ["/tmp/example/"],
            },
        }
    ],
}


class ToolState(TypedDict, total=False):
    action: dict[str, Any]
    check_url: str
    decision: dict[str, Any]
    executed: bool
    result: str


def check_action(action: dict[str, Any], check_url: str = CHECK_URL) -> dict[str, Any]:
    """Return a HANRIA decision, mapping every transport or response failure to error."""
    body = json.dumps({"mandate": MANDATE, "action": action}).encode()
    request = Request(check_url, data=body, headers={"content-type": "application/json"})
    try:
        with urlopen(request, timeout=10) as response:
            decision = json.load(response)
    except (OSError, URLError, ValueError, json.JSONDecodeError) as error:
        return {"outcome": "error", "reason": f"HANRIA check failed: {error}"}
    if not isinstance(decision, dict) or decision.get("outcome") not in {
        "permit", "deny", "escalate", "error"
    }:
        return {"outcome": "error", "reason": "HANRIA returned an invalid decision"}
    return decision


def check_node(state: ToolState) -> ToolState:
    return {"decision": check_action(state["action"], state.get("check_url", CHECK_URL))}


def route_decision(state: ToolState) -> str:
    return "execute" if state["decision"].get("outcome") == "permit" else "refuse"


def execute_node(state: ToolState) -> ToolState:
    target = state["action"]["operation"]["target"]
    return {"executed": True, "result": f"tool would run for {target}"}


def refuse_node(state: ToolState) -> ToolState:
    decision = state["decision"]
    return {
        "executed": False,
        "result": f"tool refused: {decision.get('outcome', 'error')}: "
        f"{decision.get('reason', 'no reason returned')}",
    }


def build_graph():
    graph = StateGraph(ToolState)
    graph.add_node("check", check_node)
    graph.add_node("execute", execute_node)
    graph.add_node("refuse", refuse_node)
    graph.add_edge(START, "check")
    graph.add_conditional_edges("check", route_decision, {"execute": "execute", "refuse": "refuse"})
    graph.add_edge("execute", END)
    graph.add_edge("refuse", END)
    return graph.compile()


def example_action() -> dict[str, Any]:
    return {
        "schema_version": "0.1-draft",
        "requested_by": {"agent": "langgraph-example"},
        "operation": {"kind": "file", "verb": "read", "target": "/tmp/example/note.txt"},
        "justification": "Read the example note.",
    }


if __name__ == "__main__":
    graph = build_graph()
    print("Constructed HANRIA graph without invoking it.")

运行

保存 langgraph.py，然后在保存该文件的文件夹中运行以下命令。

uv run --with langgraph==1.2.12 python langgraph.py

该命令构造图，但不会调用它。在你的应用程序中调用 build_graph().invoke({"action": example_action()})。操作字典放在图状态的 "action" 键下。检查节点必须始终位于执行节点之前。

处理每一种结果

• permit: 路由选择执行节点，因为授权书条款许可了拟议操作。
• deny: 路由选择拒绝节点。不要通过其他路径尝试该操作。
• escalate: 图路由选择拒绝节点。请在该图路由之外请求运营者作出决定。
• error: 路由选择拒绝节点。将其视为 deny，报告原因，并且不要盲目重试。

传输故障、格式错误的响应或未知结果都会变为 error，因此无法选择执行节点。

参考资料

试用检查 · 验证签名回执 · OpenAPI 文档 · 面向智能体的摘要