開発者ガイド
LangGraph エージェントに事前行為チェックを追加する
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となるため、実行ノードを選択できません。