如果你的 AI Agent 已经接入了真实交易系统、数据库或运维平台那么下面这个场景一定值得重视Agent 收到一个风险信号解析时漏掉了关键字段仍然生成了交易指令最终真实下单 120 万美元。这个数字不是科幻片里的特效而是 Agent 自主决策链路中非常典型的信任危机。本文要解决的就是这一类事故的工程防线。这类事故很难用下次让模型注意一点来规避。LLM 的推理本身具有概率性只要决策链路里缺少一道确定性的校验闸门模型幻觉、上下文压缩、工具返回格式歧义都可能被放大成真实业务损失。接下来我会从一个交易 Agent 的失败场景出发拆解 AI Agent 为什么会误读风险信号并给出完整的 RiskGuard 风险网关设计与 Python 可运行代码。1. 事故背景AI Agent 执行链路里的信任危机1.1 一类典型的 Agent 事故场景先看一个还原场景某量化交易 Agent 接入了行情服务和风控服务通过 function calling 调用下单工具。风控服务返回了一个风险信号内容类似{ signal: credit_score_drop, risk_level: HIGH, trigger_value: 80, threshold: 50, suggestion: 建议暂停加仓等待风控复核 }按照业务规则risk_levelHIGH且trigger_value threshold时系统应该禁止自动买入或者至少要进入人工审批。但在实际的 Agent 链路里LLM 在长上下文中可能没有把风险信号已触发阈值和禁止买入绑定起来仍然生成了买入指令。如果交易系统直接放行一笔 120 万美元的订单就会这样被合理但错误地执行掉。这不是模型变笨了而是 Agent 系统缺少一个原则模型可以提出意图但不能直接获得执行授权。模型输出是概率性的它的任务是生成自然语言和工具参数而这笔交易能不能执行应该由确定性代码和规则引擎决定。1.2 Agent 决策链路的三个薄弱点从工程角度看这类事故通常有三个共性问题。第一上下文压缩导致关键信息丢失。LLM 的上下文窗口有长度限制而且 Agent 在一次任务中可能叠加了多轮工具返回结果。当工具返回的 JSON 很长时模型可能只提取了与当前动作直接相关的字段忽略了前面的风险等级字段。第二工具返回结果缺少强 schema 约束。如果风控服务返回的数据结构没有被严格校验Agent 只能依赖读自然语言来理解风险信号。一旦字段命名歧义、枚举值不熟悉模型就可能给出看起来合理但错误的解读。第三缺少确定性控制层。传统自动化系统里交易前检查是硬编码逻辑行为可预测。而 Agent 系统往往把决定权全部交给了 LLM没有在工具调用入口加一道独立于模型的规则校验闸门。1.3 为什么风控不能只依赖模型判断风控决策需要可复现、可测试、可解释。给定同样的输入传统规则引擎每次都返回同样的结果而 LLM 的输出受温度、上下文顺序和 token 采样影响即使在相同输入下也可能产生不同结果。更关键的是责任边界问题。当一笔交易因为 Agent 误读信号而亏损时复盘时不能只说模型犯错了因为模型无法为真实资金损失负责。我们需要把风控责任迁移到工程系统上由代码强制执行规则由日志记录决策过程由审批流保留人工确认痕迹。这样即使 Agent 行为错误最终防线仍然可控。2. 总体设计给 Agent 加一道确定性风险网关2.1 架构分层解决思路是引入一个风险网关RiskGuard放在 Agent 与外部执行系统之间。整体链路如下用户/策略 ↓ LLM Agent ↓ 生成工具调用参数 RiskGuard 风险网关 ↓ 规则校验 / 人工审批 / 审计 执行系统交易/数据库/运维平台从 Agent 视角看它仍然是调用一个工具函数但工具函数内部不再直接连交易系统而是先经过 RiskGuard。RiskGuard 的输出有三种放行、拒绝、需要人工审批。只有放行条件下真正的交易执行代码才会被触发。2.2 风险网关核心职责RiskGuard 至少需要承担四个职责。输入标准化不管上游返回的是 JSON、XML 还是纯文本风险信号必须被转换成强类型对象。这一步能提前挡住格式解析类错误。确定性规则判定用规则引擎或普通条件判断实现业务规则比如单笔金额上限、风险等级白名单、信号阈值命中检查。规则不写在 Prompt 里不依赖模型判断。人工审批回路对于中高风险动作或超阈值动作Agent 不能继续执行必须进入人工审批流程。审批通过才放行拒绝则终止。审计与补偿每一次决策请求无论结果是通过、拒绝还是审批都必须记录完整审计日志包括风险信号、交易动作、决策原因、审批人、时间戳。生产环境还需要保留原始上下文供事后复盘。2.3 设计原则三个原则值得贯穿始终。第一模型提请求网关做决策。Agent 负责理解任务、拆分步骤、生成工具参数但能否执行必须由确定性逻辑决定。第二最小权限。Agent 使用的凭证只具备完成业务所需的最小权限即使网关被绕过也尽量限制爆炸半径。第三默认拒绝。拿不准的动作宁可拒绝或转人工也不要放行。风控场景里拒绝一笔错误交易的代价远小于放行一笔错误交易。3. 风险信号标准化与规则引擎设计3.1 风险信号从哪里来交易场景中的风险信号来源很多常见的有风险信号示例格式典型来源市场波动率volatility: 0.85, threshold: 0.7行情服务信用评分下滑credit_score_drop: 80, threshold: 50风控服务最大回撤告警max_drawdown: 12%, threshold: 10%账户服务流动性不足liquidity_depth: 200, threshold: 500订单簿服务风控人工标记manual_flag: pause_trading风控管理后台不同来源的信号字段可能完全不同但进入 RiskGuard 之前必须统一成同一个模型。这一步看似简单却是很多事故的根源如果信号字段进不了标准模型后续规则就无法判断。3.2 标准化模型我使用 Python dataclass 定义风险信号、交易动作、决策结果和审计记录。为了便于理解先创建models.py# 文件路径risk_guard/models.py from dataclasses import dataclass from enum import Enum class RiskLevel(str, Enum): LOW LOW MEDIUM MEDIUM HIGH HIGH CRITICAL CRITICAL class Decision(str, Enum): ALLOW ALLOW REJECT REJECT REQUIRE_APPROVAL REQUIRE_APPROVAL dataclass class RiskSignal: signal_type: str # 信号类型如 volatility / credit_score_drop risk_level: RiskLevel # 风险等级 value: float # 当前触发值 threshold: float # 阈值 source: str # 信号来源 dataclass class TradeAction: symbol: str # 交易标的 side: str # BUY / SELL quantity: float # 数量 price: float # 价格 order_type: str MARKET agent_id: str agent-001 request_id: str # 链路追踪 ID property def amount(self) - float: return self.quantity * self.price dataclass class DecisionResult: decision: Decision reason: str rule_name: str request_id: str dataclass class AuditRecord: timestamp: float request_id: str symbol: str side: str amount: float signal_type: str signal_level: str decision: Decision reason: strTradeAction.amount是一个计算属性方便规则直接判断单笔金额。RiskSignal里的risk_level使用枚举而不是字符串可以避免把HIGH和high当成不同值。3.3 规则引擎设计规则引擎的核心是规则可插拔、结果可合并。我定义了一个抽象基类Rule每个规则只负责一件事比如金额上限、风险等级校验、阈值命中校验。规则返回DecisionResult规则引擎统一收集结果。# 文件路径risk_guard/rules.py from abc import ABC, abstractmethod from typing import List from models import Decision, DecisionResult, RiskLevel, RiskSignal, TradeAction class Rule(ABC): name: str base_rule abstractmethod def evaluate(self, signal: RiskSignal, action: TradeAction) - DecisionResult: pass class MaxAmountRule(Rule): 单笔金额上限规则超过上限直接拒绝。 name MaxAmountRule def __init__(self, max_amount: float): self.max_amount max_amount def evaluate(self, signal: RiskSignal, action: TradeAction) - DecisionResult: if action.amount self.max_amount: return DecisionResult( decisionDecision.REJECT, reasonf单笔金额 {action.amount:.2f} 超过上限 {self.max_amount:.2f}, rule_nameself.name, request_idaction.request_id, ) return DecisionResult( decisionDecision.ALLOW, reason金额校验通过, rule_nameself.name, request_idaction.request_id, ) class RiskLevelRule(Rule): 风险信号等级规则CRITICAL 直接拒绝HIGH 必须审批。 name RiskLevelRule def evaluate(self, signal: RiskSignal, action: TradeAction) - DecisionResult: if signal.risk_level RiskLevel.CRITICAL: return DecisionResult( decisionDecision.REJECT, reason风险等级为 CRITICAL禁止自动执行, rule_nameself.name, request_idaction.request_id, ) if signal.risk_level RiskLevel.HIGH: return DecisionResult( decisionDecision.REQUIRE_APPROVAL, reason风险等级为 HIGH需人工审批, rule_nameself.name, request_idaction.request_id, ) return DecisionResult( decisionDecision.ALLOW, reason风险等级在允许范围内, rule_nameself.name, request_idaction.request_id, ) class ThresholdHitRule(Rule): 阈值命中规则当风险信号已经达到阈值Agent 仍尝试买入时必须走人工审批。 name ThresholdHitRule def evaluate(self, signal: RiskSignal, action: TradeAction) - DecisionResult: if signal.value signal.threshold and action.side BUY: return DecisionResult( decisionDecision.REQUIRE_APPROVAL, reasonf信号 {signal.signal_type} 已触发阈值Agent 仍在买入需要人工确认, rule_nameself.name, request_idaction.request_id, ) return DecisionResult( decisionDecision.ALLOW, reason信号与动作匹配, rule_nameself.name, request_idaction.request_id, ) class RuleEngine: def __init__(self, rules: List[Rule]): self.rules rules def evaluate(self, signal: RiskSignal, action: TradeAction) - List[DecisionResult]: results [] for rule in self.rules: result rule.evaluate(signal, action) results.append(result) if result.decision Decision.REJECT: break return results这里的短路逻辑比较关键一旦某个规则返回REJECT后续规则不再执行因为拒绝是最高优先级的决策。REQUIRE_APPROVAL不会中断循环是为了让多个审批条件都能被记录。3.4 人工审批服务审批是 Agent 风控里最容易形同虚设的环节。生产环境通常对接工单系统、邮件、飞书/钉钉审批流审批人需要看到完整的风险上下文。这里给出一个简化的ApprovalService# 文件路径risk_guard/approval.py import time from models import Decision, DecisionResult, TradeAction class ApprovalService: 人工审批服务。生产环境可以对接工单系统、邮件、飞书/钉钉审批流。 def __init__(self): self.approval_records {} def request_approval(self, action: TradeAction, reason: str) - DecisionResult: print(f[审批] 请求人工审批: {action.request_id}, 原因: {reason}) approved self._wait_for_approval(action, reason) self.approval_records[action.request_id] approved if approved: return DecisionResult( decisionDecision.ALLOW, reason人工审批通过, rule_nameApprovalService, request_idaction.request_id, ) return DecisionResult( decisionDecision.REJECT, reason人工审批拒绝, rule_nameApprovalService, request_idaction.request_id, ) def _wait_for_approval(self, action: TradeAction, reason: str) - bool: # 生产环境应等待审批回调例如轮询工单状态或接收 Webhook。 # 这里用固定返回 模拟延迟来演示流程。 print(f[审批] 已通知审批人等待确认...) time.sleep(1) return True这里_wait_for_approval固定返回True只是用于演示流程。现实中审批人可能拒绝、超时也可能审批未通过网关必须处理这些分支。4. Python 实战搭建一个可运行的 RiskGuard4.1 项目与文件结构完整代码放在同一目录下结构如下risk_guard/ ├── models.py # 数据模型 ├── rules.py # 规则与规则引擎 ├── approval.py # 人工审批服务 ├── guard.py # RiskGuard 门面 └── demo.py # 演示脚本依赖只有 Python 3.10 标准库不需要安装任何第三方包。4.2 RiskGuard 门面guard.py是整个风险网关的入口职责是编排规则引擎、审批服务和审计记录# 文件路径risk_guard/guard.py import time from typing import List from models import AuditRecord, Decision, DecisionResult, RiskSignal, TradeAction from rules import RuleEngine class RiskGuard: def __init__(self, rule_engine: RuleEngine, approval_service): self.rule_engine rule_engine self.approval_service approval_service self.audit_records: List[AuditRecord] [] def evaluate(self, signal: RiskSignal, action: TradeAction) - DecisionResult: results self.rule_engine.evaluate(signal, action) final self._merge_results(results) if final.decision Decision.REQUIRE_APPROVAL: final self.approval_service.request_approval(action, final.reason) self._record(signal, action, final) return final def _record(self, signal: RiskSignal, action: TradeAction, result: DecisionResult): record AuditRecord( timestamptime.time(), request_idaction.request_id, symbolaction.symbol, sideaction.side, amountaction.amount, signal_typesignal.signal_type, signal_levelsignal.risk_level.value, decisionresult.decision, reasonresult.reason, ) self.audit_records.append(record) staticmethod