从工具调用到生产护栏:安全构建 AI Agent

2026-07-16 39 预计阅读时间: 1 分钟
来源: docker.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

AI Agent 不只是“会聊天的模型”。它接收目标、观察环境、选择工具、执行动作,再根据结果决定下一步。这样的循环让模型能够处理多步骤任务,也同时放大了提示注入、越权调用、成本失控和不可审计等风险。真正困难的部分不是让 Agent 调用一次 API,而是让它在生产环境中始终待在明确的权限和预算边界内。

Agent 是一个受约束的执行循环

一个典型 Agent 可以拆成五个部分:

  1. 目标:用户希望完成什么,例如查询订单并生成退款建议。
  2. 模型:根据当前上下文提出下一步动作。
  3. 工具:搜索、数据库、HTTP API、代码执行器等外部能力。
  4. 状态:历史动作、工具结果、预算和任务进度。
  5. 终止条件:任务完成、达到步数上限、预算耗尽或触发人工审批。

关键点在于,模型只应该提出动作,应用程序负责验证和执行。不要让模型生成一段任意 shell 命令后直接交给操作系统,也不要把数据库管理员凭据放进 Agent 的运行环境。

可以把每一步抽象为结构化决策:

{
  "action": "lookup_order",
  "arguments": {
    "order_id": "A-1024"
  },
  "reason": "需要确认订单状态"
}

结构化输出并不会自动带来安全性,但它让工具白名单、参数校验、审计和拒绝策略有了稳定的输入边界。

安全边界应放在模型之外

模型提示词可以说明“不要删除数据”,但提示词不是权限系统。来自网页、邮件或文档的内容可能包含提示注入,诱导 Agent 忽略规则或泄露上下文。生产系统需要在模型之外建立强制约束。

最小权限:每个工具使用独立、低权限身份。例如,订单查询工具只能读取必要字段,退款工具只能创建待审批请求,不能直接划款。

参数验证:用 JSON Schema、类型模型或显式代码检查参数。限制路径、域名、金额、记录数量和字符串长度,不能只检查工具名称。

高风险动作审批:付款、删除、发送外部邮件、修改权限等动作应暂停循环,展示明确的执行预览,并要求人类确认。审批对象应包含冻结后的参数,避免批准后被替换。

输入与数据隔离:把工具返回的网页和文档视为不可信数据。不要把其中的指令提升为系统规则,也不要自动读取与任务无关的密钥或文件。

资源预算:设置最大步骤数、超时、模型费用、工具调用次数和结果大小。Agent 卡在循环里时,预算限制就是最后一道确定性边界。

一个可运行的受限 Agent 循环

下面是一个只使用 Python 标准库的最小示例。为了能够直接运行,它用确定性的 mock_model 代替真实模型;接入模型 API 时,只需要替换该函数,并继续返回同样的结构化字典。

将代码保存为 safe_agent.py,然后运行 python safe_agent.py

from dataclasses import dataclass, field
from typing import Any, Callable


ORDERS = {
    "A-1024": {"status": "shipped", "amount": 79.0},
    "A-2048": {"status": "pending", "amount": 249.0},
}


def lookup_order(order_id: str) -> dict[str, Any]:
    if not order_id.startswith("A-") or len(order_id) > 16:
        raise ValueError("invalid order_id")
    return ORDERS.get(order_id, {"status": "not_found"})


def request_refund(order_id: str, amount: float) -> dict[str, Any]:
    if order_id not in ORDERS:
        raise ValueError("unknown order")
    if amount <= 0 or amount > 100:
        raise ValueError("refund requires manual review")
    # Production code should create a pending approval record here.
    return {"status": "approval_required", "order_id": order_id, "amount": amount}


TOOLS: dict[str, Callable[..., dict[str, Any]]] = {
    "lookup_order": lookup_order,
    "request_refund": request_refund,
}


@dataclass
class AgentState:
    goal: str
    history: list[dict[str, Any]] = field(default_factory=list)
    max_steps: int = 4


def mock_model(state: AgentState) -> dict[str, Any]:
    """Replace this function with a model call returning the same schema."""
    if not state.history:
        return {"action": "lookup_order", "arguments": {"order_id": "A-1024"}}

    last = state.history[-1]
    if last["action"] == "lookup_order" and last["result"].get("status") == "shipped":
        return {
            "action": "request_refund",
            "arguments": {"order_id": "A-1024", "amount": 79.0},
        }

    return {"action": "finish", "arguments": {"message": "No further action"}}


def run_agent(goal: str) -> dict[str, Any]:
    state = AgentState(goal=goal)

    for step in range(state.max_steps):
        decision = mock_model(state)
        action = decision.get("action")
        arguments = decision.get("arguments")

        if action == "finish":
            return {"status": "completed", "output": arguments, "history": state.history}
        if action not in TOOLS:
            return {"status": "blocked", "reason": "tool is not allowed"}
        if not isinstance(arguments, dict):
            return {"status": "blocked", "reason": "invalid arguments"}

        try:
            result = TOOLS[action](**arguments)
        except (TypeError, ValueError) as exc:
            return {"status": "blocked", "reason": str(exc), "history": state.history}

        event = {"step": step + 1, "action": action, "arguments": arguments, "result": result}
        state.history.append(event)
        print(event)

        if result.get("status") == "approval_required":
            return {"status": "paused", "approval": event, "history": state.history}

    return {"status": "stopped", "reason": "step budget exceeded", "history": state.history}


if __name__ == "__main__":
    print(run_agent("检查订单 A-1024,并在符合规则时申请退款"))

这个示例刻意保留了几条不能由模型绕过的规则:工具必须出现在 TOOLS 白名单中,参数由具体函数再次验证,退款不会直接执行,循环最多运行四步。接入真实模型后,还应使用提供商支持的结构化输出能力校验响应,并把每次模型请求的 token 用量纳入预算。

上线时要观察动作,而不只是答案

普通聊天应用常关注最终回答是否正确;Agent 系统还必须记录它如何得到答案。建议为每次运行生成 run_id,并记录:

  • 用户目标、模型版本和提示词版本;
  • 每一步的工具名称、经过脱敏的参数、耗时与结果状态;
  • token、费用、重试次数和累计执行时间;
  • 被策略拒绝的动作与拒绝原因;
  • 人工审批者、审批时间和实际执行参数。

日志中不能原样保存访问令牌、完整个人信息或敏感文档。可以记录字段摘要、哈希值或脱敏后的参数,同时把敏感审计数据放入访问受控、保留期明确的存储中。

测试也应覆盖执行轨迹。除了正常任务,还要加入提示注入、超大参数、未知工具、重复调用、工具超时、恶意网页内容和模型返回非法 JSON 等用例。对于产生副作用的工具,使用沙箱或测试租户运行评估,不能让自动化测试操作真实客户数据。

采用前的工程检查清单

在扩大 Agent 权限之前,可以逐项确认:

  • 每个工具是否使用最小权限凭据,并有独立的调用范围;
  • 模型输出是否经过结构化校验,工具参数是否再次验证;
  • 外部内容是否始终按不可信数据处理;
  • 删除、付款、发布和权限修改是否需要人工确认;
  • 是否设置步骤数、超时、费用和并发上限;
  • 是否支持幂等键、重试退避和中断后的安全恢复;
  • 是否能够通过 run_id 重建完整动作轨迹;
  • 是否可以立即禁用某个工具或撤销 Agent 凭据。

AI Agent 的价值来自自主执行,但生产可靠性来自确定性的控制层。让模型负责建议,让代码负责权限、校验、预算和审批,才能在保留自动化能力的同时,把错误限制在可观察、可暂停、可恢复的范围内。


相关推荐