首页 / 文章

langgraphagentsguardrailsfastapillmsafety

在执行之前，利用Jev和LangGraph对智能体的工具调用进行防护。

为LangGraph智能体添加执行前的安全检查与执行后的完成审核机制，利用Jev类型的决策来阻止诸如大额退款等高风险操作。

发布者：REACTAPP.TOP阅读约 8 分钟1817 词

来源: JEV AI: Adding Pre-Execution Safety and Post-Execution Verification to LangGraph Agentic Framework.

那些能够支出资金或删除数据的智能体需要在工具运行前设置一个检查点，而在任务 supposedly 完成后还需再设置一个检查点。本示例将这两个检查点都集成到 LangGraph 电商智能体中，由 TypeSafe AI 开发的名为 Jev 的决策模型作为控制机制。

Jev 的输出内容

Jev 属于“System One”类决策模型。与 GPT-4 这类通用大型语言模型以开放式文字回答不同，Jev 能接收结构化或非结构化的状态信息，并返回格式化的决策结果、置信度分数以及概率分布。它提供了三种基本功能。

  • Choice 会从固定集合中选择一个标签，并附带置信度值及完整的概率分布返回。它适用于工具权限判断（允许、拒绝、确认、审核）以及意图分类。
  • Score将某事物置于有序等级中，并附带置信度值。它适用于金融或政策领域的风险评估（低、中、高、极严重）以及严重程度评级。
  • Noul会给出条件为真的概率，范围在0.0到1.0之间。它可用于布尔值判断、政策合规性检查以及任务完成度的评估。

项目设置

该代理运行在已安装了这些依赖项的全新Python虚拟环境中。

fastapi==0.115.6
uvicorn==0.32.1
pydantic==2.10.4
langgraph==0.2.61
langchain==0.3.13
langchain-openai==0.3.1
langchain-typesafe==0.1.0
httpx==0.27.2
python-dotenv==1.0.1

安全层：jev_utils.py

该文件充当验证中间件，是智能体与其工具之间的防护屏障，负责处理两个独立的问题。

执行前：guard_tool_call

自主智能体可能会在无人监督的情况下触发敏感操作，如支付、删除或退款。此函数位于每次工具调用之前，负责对工具名称、参数及相关安全策略进行结构化分类，从而生成决策（允许、确认、审核或拒绝）并给出风险评分。

执行后：review_completion

大型语言模型往往无法判断多步骤目标是否真的已经达成，还是仍有任务未完成。该功能会将最近的状态历史与用户最初设定的目标进行比较，进而将任务标记为已完成或未完成，并附上相应的置信度值。

import os
from typing import Any, Dict, List
from langchain_typesafe import Choice, Score, TypeSafeClassifier

if not os.getenv("TYPESAFE_API_KEY"):
    raise RuntimeError("TYPESAFE_API_KEY not set")

classifier = TypeSafeClassifier()

def guard_tool_call(tool: str, args: Dict[str, Any], policy: List[str], side_effects: List[str]) -> Dict[str, Any]:
    questions = {
        "decision": Choice(
            instructions="Choose the safest next action for this tool call.",
            criteria={
                "allow": "Execute immediately.",
                "confirm": "Proceed after user confirmation.",
                "review": "Require human review.",
                "deny": "Do not execute.",
            },
        ),
        "risk": Score(
            instructions="Score the financial and policy risk.",
            criteria=["Low", "Moderate", "High", "Critical"],
        ),
    }

    res = classifier.invoke({
        "state": {"tool": tool, "arguments": args, "policy": policy, "side_effects": side_effects},
        "questions": questions,
    })

    return {
        "decision": res.choices["decision"].choice,
        "confidence": res.choices["decision"].confidence,
        "risk_score": res.scores["risk"].score,
    }

def review_completion(state_summary: str, goal: str) -> Dict[str, Any]:
    questions = {
        "complete": Choice(
            instructions="Is the stated goal fully achieved?",
            criteria={
                "complete": "Goal fully achieved.",
                "incomplete": "Goal not achieved.",
            },
        )
    }

    res = classifier.invoke({
        "state": {"state_summary": state_summary, "goal": goal},
        "questions": questions,
    })

    return {
        "status": res.choices["complete"].choice,
        "confidence": res.choices["complete"].confidence,
    }

协调器：agent_graph.py

该文件用于构建具有状态且呈循环结构的LangGraph工作流，承担三项职责。

  • 决定下一步操作的功能在agent_node中实现。它会读取消息历史和当前目标，然后通过OpenAI的工具调用API来选择回复内容或工具调用。
  • 安全执行的功能在tools_node中实现。它会拦截模型提出的工具调用，并通过guard_tool_call对其进行检查。诸如500美元以上的退款等高风险操作会被阻止。经过审批的工具才会被执行，其执行结果会被记录到tool_outputs中，从而使最终回复包含结构化的信息快照。
  • 生命周期与路由属于build_agent_graph功能。该功能会从起始节点设置条件分支，先指向智能体，再要么指向工具，要么指向完成检查，最后到达终点。循环会不断在推理与工具使用之间切换，直至任务完成。
    • from typing import Annotated, Any, Dict, List, Optional
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, BaseMessage, SystemMessage, ToolMessage
from langchain_openai import ChatOpenAI
from jev_utils import guard_tool_call, review_completion

# --- Typed State ---

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    goal: Optional[str]
    tool_outputs: List[Dict[str, Any]]

# --- Tools Definitions & Logic ---

TOOL_POLICIES = {
    "checkout": {
        "policy": ["Payment captured upon order creation."],
        "side_effects": ["Charges customer"],
    },
    "issue_refund": {
        "policy": ["Refunds above $500 require approval."],
        "side_effects": ["Moves funds"],
    },
}

def get_catalog_info(query: str) -> Dict[str, Any]:
    return {"result": f"Catalog info for: {query}"}

def add_to_cart(user_id: str, product_id: str, qty: int) -> Dict[str, Any]:
    return {"status": "ok", "user_id": user_id, "product_id": product_id, "qty": qty}

def checkout(user_id: str, cart_id: str, address: Dict[str, str], **_) -> Dict[str, Any]:
    return {
        "status": "ok",
        "order_id": "ord_12345",
        "user_id": cart_id,
        "items": [{"product_id": "prod_777", "qty": 2}],
        "ship_to": address,
    }

def issue_refund(order_id: str, amount_usd: float, reason: str) -> Dict[str, Any]:
    return {"status": "ok", "refund_id": "ref_999", "order_id": order_id, "amount_usd": amount_usd}

TOOLS = {
    "get_catalog_info": get_catalog_info,
    "add_to_cart": add_to_cart,
    "checkout": checkout,
    "issue_refund": issue_refund,
}

TOOL_SCHEMAS = [
    {
        "name": "get_catalog_info",
        "description": "Get product/catalog information.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
    {
        "name": "add_to_cart",
        "description": "Add product to cart.",
        "parameters": {
            "type": "object",
            "properties": {"user_id": {"type": "string"}, "product_id": {"type": "string"}, "qty": {"type": "integer"}},
            "required": ["user_id", "product_id", "qty"],
        },
    },
    {
        "name": "checkout",
        "description": "Place an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string"},
                "cart_id": {"type": "string"},
                "address": {
                    "type": "object",
                    "properties": {"line1": {"type": "string"}, "city": {"type": "string"}, "state": {"type": "string"}, "country": {"type": "string"}, "pin": {"type": "string"}},
                    "required": ["line1", "city", "state", "country", "pin"],
                },
            },
            "required": ["user_id", "cart_id", "address"],
        },
    },
    {
        "name": "issue_refund",
        "description": "Issue order refund.",
        "parameters": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}, "amount_usd": {"type": "number"}, "reason": {"type": "string"}},
            "required": ["order_id", "amount_usd", "reason"],
        },
    },
]

llm = ChatOpenAI(model="gpt-4.1", temperature=0).bind_tools(TOOL_SCHEMAS)

# --- Graph Nodes ---

def agent_node(state: AgentState) -> Dict[str, Any]:
    sys_msg = SystemMessage(
        content=f"You are an e-commerce assistant. Target Goal: {state.get('goal') or 'N/A'}"
    )
    response = llm.invoke([sys_msg] + state["messages"])
    return {"messages": [response]}

def tools_node(state: AgentState) -> Dict[str, Any]:
    last_msg = state["messages"][-1]
    tool_messages = []
    tool_outputs = list(state.get("tool_outputs", []))

    for tc in getattr(last_msg, "tool_calls", []):
        name, args, call_id = tc["name"], tc["args"], tc["id"]

        # Guard check
        if name in TOOL_POLICIES:
            config = TOOL_POLICIES[name]
            guard = guard_tool_call(name, args, config["policy"], config["side_effects"])

            is_blocked = (
                guard["decision"] == "deny"
                or (name == "issue_refund" and args.get("amount_usd", 0) > 500)
            )

            if is_blocked:
                msg = f"Action `{name}` blocked by policy or high-risk evaluation."
                tool_messages.append(ToolMessage(content=msg, tool_call_id=call_id))
                tool_outputs.append({
                    "status": "blocked",
                    "tool": name,
                    "result": {"error": msg}
                })
                continue

        # Execute safe tool
        res = TOOLS[name](**args)
        tool_messages.append(ToolMessage(content=f"Result: {res}", tool_call_id=call_id))

        # Append structured tool execution record
        tool_outputs.append({
            "status": res.get("status", "ok"),
            "tool": name,
            "result": res
        })

    return {"messages": tool_messages, "tool_outputs": tool_outputs}

def completion_check_node(state: AgentState) -> Dict[str, Any]:
    # Pass-through node; does not pollute message history
    return {}

# --- Builder ---

def build_agent_graph():
    builder = StateGraph(AgentState)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", tools_node)
    builder.add_node("completion_check", completion_check_node)

    builder.add_edge(START, "agent")
    builder.add_conditional_edges(
        "agent",
        lambda state: "tools" if getattr(state["messages"][-1], "tool_calls", None) else "completion_check",
    )
    builder.add_edge("tools", "agent")
    builder.add_edge("completion_check", END)

    return builder.compile()

    Web层：main.py

    最后一个文件使用FastAPI通过HTTP暴露代理功能。其组成部分包括：

    • 一个/chat REST接口，用于接收消息历史、当前消息以及可选的目标。
    • 将传入的JSON格式消息历史转换为LangChain的HumanMessage和AIMessage对象，以便用于构建图结构。
    • 使用准备好的状态（包括消息、目标以及一个空的tool_outputs列表）调用图结构中的相关功能。
    • 一个类型化的Pydantic响应对象ChatResponse，其中包含代理的最终reply内容，以及一个包含tool_outputs和goal信息的state_snapshot。
    from typing import Any, Dict, List, Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from langchain_core.messages import AIMessage, HumanMessage
from pydantic import BaseModel
from agent_graph import build_agent_graph

load_dotenv(override=True)

app = FastAPI()
agent = build_agent_graph()

class ChatRequest(BaseModel):
    user_id: str
    message: str
    goal: Optional[str] = None
    history: Optional[List[Dict[str, str]]] = None

class StateSnapshot(BaseModel):
    tool_outputs: List[Dict[str, Any]]
    goal: Optional[str] = None

class ChatResponse(BaseModel):
    reply: str
    state_snapshot: StateSnapshot

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    messages = []

    if req.history:
        for m in req.history:
            role, content = m.get("role"), m.get("content", "")
            if role == "user":
                messages.append(HumanMessage(content=content))
            elif role == "assistant":
                messages.append(AIMessage(content=content))

    messages.append(HumanMessage(content=req.message))

    try:
        final_state = agent.invoke({
            "messages": messages,
            "goal": req.goal,
            "tool_outputs": [],
        })
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    # Get the last AI message with non-empty content
    reply_msg = next(
        (m.content for m in reversed(final_state["messages"]) if isinstance(m, AIMessage) and m.content),
        ""
    )

    return ChatResponse(
        reply=str(reply_msg),
        state_snapshot=StateSnapshot(
            tool_outputs=final_state.get("tool_outputs", []),
            goal=final_state.get("goal")
        )
    )

    使用 uvicorn 启动服务器：

    #Run uvicorn server
$uvicorn main:app --reload --port 8000 --host 0.0.0.0

    然后提交一份要求高额退款的申请：

    curl -X 'POST' \
  'http://127.0.0.1:8000/chat' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "user_id": "alice_123",
  "message": "Please issue a $750 refund for order ord_999 because the item arrived broken.",
  "goal": "Issue a $750 refund for order ord_999 to user alice_123."
}'

    回复显示客服正在履行职责。该工作人员拒绝处理750美元的退款，并表示高金额退款可能需要额外的核实或审批。

    {
  "reply": "I'm unable to process the $750 refund for order ord_999 due to a policy restriction or high-risk evaluation. For high-value refunds, additional verification or approval may be required.\n\nWould you like me to connect you with customer support or provide further instructions on how to proceed?",
  "state_snapshot": {
    "tool_outputs": [
      {
        "status": "blocked",
        "tool": "issue_refund",
        "result": {
          "error": "Action `issue_refund` blocked by policy or high-risk evaluation."
        }
      }
    ],
    "goal": "Issue a $750 refund for order ord_999 to user alice_123."
  }
}

    为何退款被拒

    在 agent_graph.py 中的 tools_node 内部，有一个明确的条件用于判断是否拒绝某次请求：

    is_blocked = (
    guard["decision"] == "deny"
    or (name == "issue_refund" and args.get("amount_usd", 0) > 500)
)

    请求的750美元金额超过了500美元的上限，因此is_blocked会立即被设置为true。这一结果背后有两个原因。

    • 策略配置：TOOL_POLICIES规定，金额超过500美元时需经过审批才能执行issue_refund操作。
    • 风险评估：在issue_refund执行之前，jev_utils.py中的guard_tool_call会先对风险进行评估。大额资金转账通常会被判定为High、Critical，或者不属于allow类别，从而导致操作被标记为已阻止。

    设计经验

    Jev起到安全防护层的作用，而非生成器。OpenAI模型（gpt-4.1）负责推理、日常对话以及决定调用哪些工具。Jev则作为一个独立的监管机制，在这些请求执行之前进行拦截，并根据政策、安全规则和金额阈值对它们进行检测。

    相关阅读

  • 你的第一个LangGraph智能体：具备工具与路由功能的足球助手 — 逐步LangGraph教程：共享状态、工具、规划节点、条件边以及Markdown总结，以一个模拟的足球数据助手为例进行讲解。
    • ← 全部文章发送简报