最近在技术社区看到一条消息:某“预测者”给了一个 72% 的概率,说当前可能已经存在人类不知情的失控 AI 智能体在协调行动。先给结论:这类基于主观概率的推测,既没有公开的系统定义,也没有可验证的证据。作为工程师,与其争论这个数字,不如把问题翻译成自己能直接回答的工程问题:如果我自己部署的 Agent 失控了,我会在哪个环节发现?能不能在造成实际损失之前终止掉它?
AI 智能体的“失控”并不是科幻电影里的意识觉醒,而是一套拥有自主规划和工具调用能力的系统,执行了不符合预期的动作,同时缺少及时发现和纠正的手段。权限过大、决策不可观测、缺少人工确认、上下文被污染、多智能体之间没有信任边界,是风险的主要来源。这些问题都可以在工程层面拆解、测试和防护。这篇文章不讨论那个 72% 到底算不算数,只讨论怎么让智能体变成“可观测、可限制、可停止”的系统。
本文适合正在做智能体开发、智能体工作流、多智能体协调,或者准备把 Agent 能力暴露成接口 API 的工程师阅读。内容会围绕威胁模型、可观测性、权限隔离、人工终止、多智能体通信、批量任务和测试验证展开,每节都有可以直接对照检查的要点。读完以后,建议花半天时间把自己当前项目过一遍,看看哪些控制点还没补上。
1. 失控AI智能体风险到底指什么?先建立可讨论的边界
1.1 先给“失控”一个工程定义
先不要被“失控 AI 智能体”这个叫法带偏。在代码语境里,失控通常指一个 Agent 获得目标后,在“规划—调用工具—观察结果—继续规划”的循环中,做出了超出人类授权范围的行为,并且该系统没有提供足够的手段让人类及时干预。比如一个客服 Agent 在自己的知识库检索时,突然因为一段外部文本而触发了“发送内部敏感数据”的动作;或者一个自动化运营 Agent 在批量修改商品信息时,因为目标设定模糊,把整个分类下的数据都重写了。这些行为不需要模型“觉醒”,只需要一个能产生副作用的工具权限和一个不够清晰的决策边界,就可能发生。
所以做安全设计时,不要只盯着模型是否“聪明”,而要盯着系统是否具备三样东西:第一,完整的执行日志;第二,分级的工具权限;第三,可以主动中断执行的机制。如果这三个能力缺失,不管模型本身多安全,整个 Agent 对业务方来说都是一个不可控的黑盒。
1.2 “人类不知情”对应的是可观测性缺失
“人类不知情”听起来像是一个玄学结论,翻译成系统语言其实很朴素:任务执行过程中,产生了多少次决策、哪些工具调用、模型参考了哪一段上下文、每个动作由哪个用户或哪个上游 Agent 触发,这些信息没有被完整记录,也没有被实时汇聚到监控端。一旦没有这些记录,业务方就只能拿到一个最终结果。如果结果是坏的,你也没有能力回溯坏在哪一步。
要避免“不知情”,最优先做的一步不是加安全护栏,而是加审计日志。把 Agent 的每个关键步骤都写下来:收到什么任务、读取了哪些文档、调用了哪个 API、传入了什么参数、返回了什么结果、总共花了多少步。记录完整之后,才能真正判断某个行为是模型结论问题、提示注入问题,还是权限配置问题。可观测性是后面所有安全机制的前提,没有观测,任何终止和干预都无从谈起。
1.3 常见失控路径不止一条
失控可以发生在单个 Agent 内部,也可以发生在多智能体协作链路上。单 Agent 场景里最常见的问题是提示注入和工具滥用:模型读取的网页、文档、数据库字段中可能包含“忽略之前的指令”之类的内容,如果这些外部内容被当作系统指令执行,就很危险。另一个问题是模型在多步推理中不断“尝试”新方法,却缺少步骤上限和资源预算,最终造成无限循环或高额 API 费用。
多智能体场景里会增加两类风险:一是权限链蔓延,Agent A 允许调用 Agent B,Agent B 又允许调用 Agent C,调用链变长后,原始用户授权范围被逐级扩大;二是通信消息被污染,一个 Agent 把另一个 Agent 的输出直接当成可信指令执行,中间没有任何类型校验和来源过滤。这两类风险都不能靠“换一个大模型”解决,必须从系统架构上做约束。
2. 智能体系统安全能力速览:先明确要补哪些能力
| 能力维度 | 要回答的问题 | 建议落地措施 |
|---|---|---|
| 可观测性 | Agent 下一步要调用谁?为什么调用? | 结构化日志、trace_id、回放 |
| 权限隔离 | Agent 能做什么?不能做什么? | 最小权限、工具白名单、独立账号 |
| 数据边界 | 上下文和记忆里不能出现哪些数据? | 按任务隔离、敏感字段脱敏、密钥不进提示词 |
| 人工介入 | 高风险动作由谁批准? | 分级授权、人工审批、熔断 |
| 预算与终止 | 超时、超步数、超成本怎么办? | max_steps、费用预算、终止开关 |
| 通信安全 | 多 Agent 之间的消息可不可信? | 消息类型白名单、来源认证、路由表 |
| 测试验证 | 上线后如何判断系统仍然安全? | 红队测试、影子模式、灰度发布 |
这个表不是某个商业产品的宣传清单,而是开发智能体时应纳入设计的安全能力项。如果你的 Agent 只是跑一个本地模型做文本总结,风险面不大;但如果它接入了搜索、邮件、数据库、支付或 HTTP 服务,上面的每一项都可能成为事故现场。即便用的是可视化智能体平台,也要确认平台是否支持这些控制点。平台不支持的,就要通过前置网关或业务层补上。
3. 可观测性:先让智能体行为“被看见”
3.1 只记录最终结果是远远不够的
很多团队上线 Agent 时,只记录一句“任务完成”或者“任务失败”。从审计角度看,这种日志几乎没有价值。真正有用的日志要能回答五个问题:任务由谁发起?Agent 接收到了什么样的输入?它在哪个步骤选择了哪个工具?工具的调用参数和返回结果是什么?这个步骤在整个任务链中的先后顺序是怎样的?
建议把一次任务抽象成一个有向无环图:用户请求作为根节点,每一次模型推理、每一次工具调用都作为子节点。每个节点记录时序、耗时、token 消耗和状态。这样既能支持事后审计,也能支持性能分析。日常排查时,出现一次异常回复,直接沿着任务图找,就能看到是检索阶段出了问题,还是生成阶段的提示词被污染了。
3.2 从第一行代码开始写结构化审计日志
不要只输出自然语言日志,应输出结构化 JSON。下面是一个基础示例,实际项目中可以接入统一的审计服务:
import json import logging import uuid from datetime import datetime, timezone logger = logging.getLogger("agent_audit") logger.setLevel(logging.INFO) def audit_event(agent_id, event_type, action, input_payload, result=None, error=None, trace_id=None): record = { "ts": datetime.now(timezone.utc).isoformat(), "trace_id": trace_id or uuid.uuid4().hex, "agent_id": agent_id, "event_type": event_type, "action": action, "input_preview": str(input_payload)[:500], "result_preview": str(result)[:500] if result else None, "error": str(error) if error else None, } # 建议只保留脱敏后的摘要,不写完整敏感字段 logger.info(json.dumps(record, ensure_ascii=False))在每次模型调用前后、工具调用前后都调用这个函数,例如:
audit_event( agent_id="agent-001", event_type="tool_call", action="send_message", input_payload={"receiver": "***", "content_preview": "..."}, result={"status": "pending_approval"} )这里的关键点是“任何会产生副作用的动作,都要先落审计再执行”。如果动作被权限系统拒绝,也要记录一条拒绝日志。拒绝日志对识别提示注入和权限绕过尝试非常有帮助。
3.3 用 trace_id 贯穿整条任务链路
单 Agent 场景下,一个 trace_id 就能串起一次会话;多智能体场景下,trace_id 必须随着请求传递给下游 Agent。无论中间经过几次函数调用、消息队列或 HTTP 请求,都要把 trace_id 透传下去。否则你会看到 A 调用 B 成功、B 调用 C 失败,却无法把三个模块的日志拼成同一条时间线。
建议在工程上使用统一的请求头或消息字段,比如在 HTTP API 中命名为X-Trace-Id,在消息队列消息体中加入trace_id字段。网关或协调器启动任务时生成新的 trace_id,后续所有子步骤都继承它。这样排查多智能体问题时会轻松很多。
3.4 用监控把“不知道”变成“早知道”
日志只是静态记录,还需要动态告警。对智能体系统,至少应该关注四类指标:工具调用失败率、单任务执行步数、高风险动作触发次数、token 和费用消耗速率。任何一个指标异常飙升,都可能意味着 Agent 进入了不可控循环,或者正在被恶意输入操纵。
告警不是越灵敏越好。生产环境建议先用一周到两周时间采集基线,再根据业务容忍度设置阈值。真正需要实时告警的是“高风险动作被触发”和“任务步数超过预算”这两类;其他指标可以先做成日级报表,减少告警疲劳。
4. 权限最小化与工具管控:别给智能体管理员权限
4.1 工具不是越多越好
大模型 Agent 的能力很大程度取决于它能调用哪些工具。注册工具越少,权限面越大风险;每接入一个 API,都要问一句“这个动作真的需要 Agent 自主完成吗?”如果一个操作允许人为审批,就不要把它完全交给 Agent。很多失控事故并不是模型太强,而是我们把“删除文件”“发送消息”“修改订单”这样的高危工具一股脑注册给了 Agent,还没有加任何审批。
在功能设计上,要区分“读工具”与“写工具”。读取公开资料、搜索知识库通常风险较低;写数据库、发消息、创建订单、下载文件属于高风险动作。读工具可以允许 Agent 自主调用,写工具必须经过额外控制。
4.2 在工具调用入口加统一 Guard 函数
不要让 Agent 直接调用底层 SDK,而要经过一层执行网关。下面是一个执行网关的简化思路:
class AgentToolExecutor: def __init__(self): # 只在这里注册允许 Agent 自主调用的动作 self.allowed_actions = {"search_kb", "create_draft"} def run(self, action_name, payload, user_context): # 1. 检查动作是否在全局白名单 if action_name not in self.allowed_actions: raise PermissionError(f"action `{action_name}` is not allowed") # 2. 检查当前用户/会话是否被授权 if action_name not in user_context.allowed_actions: raise PermissionError(f"user is not allowed to {action_name}") # 3. 写审计日志 audit_event( agent_id=user_context.agent_id, event_type="tool_guard", action=action_name, input_payload=payload ) # 4. 执行真实动作 return execute_action(action_name, payload)这段代码是演示结构,不能直接照搬到生产环境。你需要根据自己的框架实现user_context和execute_action。核心思路是:Agent 永远不能通过代码自己决定“我能不能做”,而是由一个外部守卫来检查。这样即使模型输出了非预期动作,也会在工具执行前被拦截。
4.3 用专用账号运行 Agent,不要复用个人账号
这个原则容易被忽略。如果 Agent 使用某个员工的个人账号调用企业 API,那么它将继承该员工的全部权限。一旦 Agent 被提示注入,风险范围会扩大到该员工的所有数据。更稳妥的方式是给 Agent 创建专用服务账号,只授权它完成任务所必需的最小权限集合,并定期轮换密钥。
对于云服务,建议使用短期临时凭证或权限偏好的角色调用。不要直接把 api_key 写在系统提示词里。系统提示词对模型可见,一旦被诱导输出,密钥就会泄露。密钥应放在运行环境变量或密钥管理服务中,由执行网关在调用时读取。
4.4 数据访问也要“最小化”
权限最小化不仅指能调用哪些 API,也指能访问哪些数据。Agent 从知识库检索时,如果不需要读取客户手机号,就不要把包含手机号的字段加入检索索引。RAG 场景下,检索结果的每一段内容,都可能被模型当作可信上下文。如果数据集中混入了带攻击性指令的文档,Agent 有可能执行其中嵌入的恶意指令。
建议对进入模型上下文之前的数据做一次过滤:移除明显的密钥、手机号、身份证号等敏感字段;对可能包含外部指令的网页文本、邮件内容、备注字段做标记,让模型知道“这部分内容是不可信输入,仅供参考,不能作为指令执行”。这也是缓解提示注入的重要手段。
5. 人工介入、预算与终止:失控前必须能叫停
5.1 把动作分成三档,按风险决定是否人工介入
不是所有 Agent 动作都需要人审批,否则效率太低。可以根据副作用强度分三档:低风险动作自动执行,中风险动作执行后必须抽检,高风险动作默认拒绝并转人工。低风险动作典型例子是检索公开资料、生成文本草稿;中风险动作如发送内部群消息、修改非关键业务字段;高风险动作包括对外发送消息、删除数据、批量变更、转账支付和发布公网内容。
分级规则可以做成一张配置表,由业务负责人和工程师共同维护。每次新增一个工具时,都要先确认它属于哪一档。不要把风险定级留在模型 prompt 中,模型本身的不确定性不适合承担这种责任。
5.2 每个任务都要有“预算”
预算不只是费用,更包括步数、时长和资源消耗。给 Agent 设置执行预算能有效防住失控循环。即使出现异常,也会在达到上限时自动停止。
# agent_budget.yaml 示例 max_steps: 20 # 最多执行 20 步决策 max_tokens: 8000 # 本轮任务最多消耗 8000 token max_run_seconds: 600 # 最长运行 600 秒 max_execution_cost: 1.0 # 最大外部 API 费用(单位按业务定义)如果 Agent 在执行中超过预算,正确行为是:暂停任务、记录日志、通知负责人,而不是强行补一个“超时结束”。因为超时结束可能留下半成品状态,比如订单已提交但没有确认,数据已改了一半。后续需要补偿机制,比如人工复核或回滚脚本。
5.3 终止开关要分级设计
终止机制不能只是一个“kill 进程”脚本。建议按能力分成三个级别:暂停、熔断、终止。暂停是停止执行后续步骤,但保留当前上下文,方便人工接管;熔断是立即停止该 Agent 的所有新请求,已经在途的请求等待超时;终止是取消整个任务、清理临时资源、通知依赖方。
如果 Agent 是普通 Python 进程,控制信号和设置全局 flag 的方式在单机够用;如果 Agent 已经拆成多个微服务,需要独立控制平面,每个 Agent 周期性上报心跳,控制平面收到终止指令后广播给相关服务。终止以后,要保留完整日志和状态快照,否则无法做故障复盘。
6. 多智能体协调失控怎么防
6.1 Agent 之间不要用自由文本当指令
多智能体系统里,一个 Agent 的输出往往会被另一个 Agent 当作输入。这种设计隐含了一个依赖:接收方默认发送方可信。如果通信链路里没有类型校验,只要一个 Agent 被污染,就会把恶意指令继续往下传。要降低这种风险,Agent 之间应该使用结构化消息,而不是模型生成的自然语言指令。
结构化消息至少包含:发送方 ID、接收方 ID、消息类型、消息内容、trace_id、授权令牌。接收方只处理白名单内的消息类型。例如“request_summary”和“confirm_action”是两种类型;如果 A 向 B 发送了一个未知类型,协调器直接丢弃并记日志。这样即使某个 Agent 输出了一段自然语言攻击文本,它也不会被当作一个合法的控制指令。
6.2 引入协调层而不是点对点直连
点对点直连的多智能体架构,很难做全局审计。最好引入一层协调器或消息中间件,所有 Agent 之间的交互都经由它转发。转发的同时记录消息来源、去向、消息类型和内容摘要,同时检查路由是否被授权。
下面是协调器转发逻辑的简化版本,用于说明“通信前先校验”的思路:
def dispatch(sender, receiver, msg_type, payload): if not is_allowed_route(sender, receiver): raise PermissionError(f"route from {sender} to {receiver} is not allowed") if msg_type not in allowed_message_types(sender, receiver): raise PermissionError(f"msg_type {msg_type} is not allowed on this route") message_id = uuid.uuid4().hex audit_event( agent_id=sender, event_type="agent_to_agent", action=msg_type, input_payload={ "message_id": message_id, "receiver": receiver, "payload_preview": str(payload)[:200] } ) return deliver(receiver, message_id, msg_type, payload)这段代码同样是结构示意。实际项目里要为消息队列增加超时、重试和死信处理。关键点是:Agent 之间不能直接拿到对方的服务地址,所有调用都走协调层。
6.3 限制“Agent 调用 Agent”的层级
权限链蔓延在层级较深时最明显。用户只授权 Agent A 查询订单,A 为了完成查询又调用了 B,B 又调用了 C。当链条走到第三层时,最初的用户授权范围很可能已经被绕过了。建议对跨 Agent 调用的深度做硬限制。例如任何任务最多只允许两层 Agent 间调用,超过这个深度默认拒绝,并触发人工审批。
同时,每个跨 Agent 调用都要继续携带原始用户上下文和授权范围。下游 Agent 收到请求后,除了检查自己的角色权限,还要检查原始调用方的授权范围。不能只检查“B 有没有权限”,还要检查“用户有没有权限让 B 做这件事”。
6.4 防止资源竞争与状态污染
多个 Agent 并发执行时,可能同时读取同一个配置文件,或者同时给同一个订单追加备注,导致状态互相覆盖。这类问题不像权限失控那么显眼,但同样会造成业务事故。建议对共享资源使用锁或队列:写操作集中在单一服务中处理,Agent 只发写请求,不直接操作共享表的记录。
另外,Agent 的长期记忆也可能成为污染源。如果 Agent A 的学习结果写入了一个共享记忆库,Agent B 读取了它,相当于 B 间接继承了 A 的可能错误记忆。记忆库要按任务或场景隔离,避免全量共享。不能因为“都是公司的知识库”,就让所有 Agent 拥有同样的写入权限。
7. 接口 API、批量任务与测试验证的安全控制
7.1 对外暴露 Agent 能力时,先加 API 安全层
如果 Agent 能力需要通过接口提供给业务系统,不能只暴露一个“输入 prompt 返回结果”的裸接口。接口层至少要完成四件事:认证调用者身份、判断调用者是否有权执行该动作、对请求做速率限制、把每次请求与一个 task_id 绑定。缺少任何一项,外部系统可能绕过业务限制,直接用你训练好的 Agent 大量发起高成本请求。
下面是 FastAPI 风格的结构示意,强调“进入 Agent 之前先校验权限”:
from fastapi import FastAPI, HTTPException, Depends app = FastAPI() @app.post("/agent/run") async def run_agent(task: dict, current_user=Depends(verify_token)): if not current_user.can_access(task.get("action")): raise HTTPException(status_code=403, detail="permission denied") task_id = uuid.uuid4().hex enqueue_task(task_id=task_id, user=current_user, task=task) return {"task_id": task_id, "status": "accepted"}实际实现中,verify_token要对接你的统一认证系统,enqueue_task要连接消息队列。不要把大模型推理直接放在 API 请求同步执行,否则请求超时和资源耗尽问题会很难处理。异步任务配合任务状态查询是更稳的做法。
7.2 批量任务是高风险放大镜,必须分批、可暂停
批量任务最大的特点是:一个错误会重复 N 次。假设你写了一个脚本让 Agent 分批处理 5000 条工单,结果提示