soulpolicy-langgraph

5 行给你的 LangGraph agent 加漂移监测。

安装

pip install soulpolicy soulpolicy-langgraph

用法

from langgraph.graph import StateGraph, END
from typing import TypedDict
from soulpolicy import Client
from soulpolicy_langgraph import drift_check_node, should_block_on_drift

class State(TypedDict, total=False):
    messages: list
    drift: dict | None

sp = Client(api_key="sk_live_xxx")

graph = StateGraph(State)
graph.add_node("agent", my_agent_step)
graph.add_node("drift", drift_check_node(sp, baseline="bl_xxx"))
graph.set_entry_point("agent")
graph.add_edge("agent", "drift")
graph.add_conditional_edges(
    "drift",
    should_block_on_drift("warn"),     # 路由谓词
    {"continue": "agent", "block": END},
)

5 行：1 个 client，1 个 node，1 个谓词，2 条边。其余照常写。

文本提取

默认从 state["messages"] 取最后一条 AI/assistant 消息。也支持 state["output"] / state["last_message"] 字符串。要自定义：

graph.add_node("drift", drift_check_node(
    sp,
    baseline="bl_xxx",
    state_text=lambda s: s["agent_reply"],
))

失败容忍

drift_check_node(sp, baseline="bl_xxx", on_error="skip")
# 调用失败时写入 state["drift"]=None + state["drift_error"]=str(e)
# 默认 "raise" 让 LangGraph retry/handle

路由

should_block_on_drift("warn") —— severity ≥ warn 路由到 "block"。可改：

should_block_on_drift("critical", continue_target="next", block_target="rollback")