企业官网建设流程全解析

如果你的 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_level=HIGH且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 风险信号从哪里来

交易场景中的风险信号来源很多，常见的有：

不同来源的信号字段可能完全不同，但进入 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: str

TradeAction.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( decision=Decision.REJECT, reason=f"单笔金额 {action.amount:.2f} 超过上限 {self.max_amount:.2f}", rule_name=self.name, request_id=action.request_id, ) return DecisionResult( decision=Decision.ALLOW, reason="金额校验通过", rule_name=self.name, request_id=action.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( decision=Decision.REJECT, reason="风险等级为 CRITICAL，禁止自动执行", rule_name=self.name, request_id=action.request_id, ) if signal.risk_level == RiskLevel.HIGH: return DecisionResult( decision=Decision.REQUIRE_APPROVAL, reason="风险等级为 HIGH，需人工审批", rule_name=self.name, request_id=action.request_id, ) return DecisionResult( decision=Decision.ALLOW, reason="风险等级在允许范围内", rule_name=self.name, request_id=action.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( decision=Decision.REQUIRE_APPROVAL, reason=f"信号 {signal.signal_type} 已触发阈值，Agent 仍在买入，需要人工确认", rule_name=self.name, request_id=action.request_id, ) return DecisionResult( decision=Decision.ALLOW, reason="信号与动作匹配", rule_name=self.name, request_id=action.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( decision=Decision.ALLOW, reason="人工审批通过", rule_name="ApprovalService", request_id=action.request_id, ) return DecisionResult( decision=Decision.REJECT, reason="人工审批拒绝", rule_name="ApprovalService", request_id=action.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 实战：搭建一个可运行的 RiskGuard

4.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( timestamp=time.time(), request_id=action.request_id, symbol=action.symbol, side=action.side, amount=action.amount, signal_type=signal.signal_type, signal_level=signal.risk_level.value, decision=result.decision, reason=result.reason, ) self.audit_records.append(record) @staticmethod