大模型的幻觉拦截:如何为全自动执行的数字员工设置“安全刹车”?
一、全自动执行让幻觉的代价变了
大模型幻觉在对话场景中表现为“说错话”,在 Agent 场景中表现为“做错事”。两者的工程含义完全不同。
对话场景:模型编造一个数字,用户看到后可能质疑、可能忽略。代价是“信息不准确”,风险由人兜底。
全自动执行场景:模型在任务规划中编造一个“数据源不存在”或“字段值为空”的假设,Agent 据此调用工具、写入数据、触发审批。代价是“业务动作错误”,风险直接传导到生产环境。
一个具体的工程场景:Agent 执行“生成上月销售报表”任务,调用 CRM 接口时接口超时。模型没有报错,而是“推断”了一个合理的销售额填进报表。报表发给了管理层,决策基于错误数据做出。这不是“模型不聪明”,而是“执行链路没有刹车”。
Nylas 的调查数据给出了一个参照:1 至 5 个 Agent 的组织,Agent 错误事故率为 45%;超过 20 个 Agent 的组织,事故率升至 70%。Agent 越多,执行链路越长,幻觉传导到业务的机会越多。
二、整体架构:五层安全刹车
┌─────────────────────────────────────────────────────────────────┐
│ Agent 执行引擎 │
│ 任务规划 → 工具选择 → 参数组装 → 调用 → 结果处理 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第一层:事实校验层(Grounding Check) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 数据源存在性 │ │ 字段值完整性 │ │ 数值合理性 │ │
│ │ 接口返回校验 │ │ 空值不补全 │ │ 范围/逻辑 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第二层:多源比对层(Cross-Source Validation) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 交叉验证 │ │ 冲突检测 │ │ 一致性评分 │ │
│ │ 多源比对 │ │ 差异高亮 │ │ 置信度分级 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第三层:置信度门控层(Confidence Gate) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 输出置信评分 │ │ 阈值分级 │ │ 低置信拦截 │ │
│ │ 证据链完整性 │ │ 高/中/低 │ │ 转人工/降级 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第四层:Human-in-the-loop 确认层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 高风险动作 │ │ 不可逆操作 │ │ 金额超阈值 │ │
│ │ 人工确认 │ │ 人工确认 │ │ 人工确认 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第五层:Event-Driven 熔断层(AIOps) │
│ 异常事件采集 → 模式检测 → 自动熔断 → 人工接管 │
└─────────────────────────────────────────────────────────────────┘
核心设计原则:幻觉拦截不是“事后检查”,而是嵌在执行链路中的多层刹车。每一层降低一部分风险,叠加起来才能把“做错事”的概率压到可接受范围。
用 Mermaid 表达这条拦截链路,结构会更清晰:
三、第一层:事实校验——不让“编造”进入执行链路
事实校验的核心是:模型输出的每一个事实性结论,都必须有可验证的来源。没有来源,不进入执行。
3.1 数据源存在性与字段完整性校验
from dataclasses import dataclass
from typing import Optional, List
from enum import Enum
class GroundingStatus(Enum):
GROUNDED = "grounded" # 有可靠来源
PARTIAL = "partial" # 部分字段缺失
UNGROUNDED = "ungrounded" # 无来源,疑似编造
@dataclass
class GroundingResult:
status: GroundingStatus
missing_fields: List[str]
unverified_fields: List[str]
evidence: dict
class GroundingChecker:
"""事实校验器:检查模型输出是否有可验证来源"""
def __init__(self, connector_registry, schema_store):
self.connectors = connector_registry
self.schema = schema_store
def check(self, tool_call: dict, tool_result: dict) -> GroundingResult:
"""校验工具调用的结果是否有事实依据"""
# 1. 接口调用失败:不允许“推断”补全
if tool_result.get("error") or tool_result.get("status") == "failed":
return GroundingResult(
status=GroundingStatus.UNGROUNDED,
missing_fields=[],
unverified_fields=[],
evidence={"error": tool_result.get("error")}
)
# 2. 空值不补全:模型不允许“脑补”缺失值
required_fields = self.schema.get_required_fields(tool_call["action"])
missing = [
f for f in required_fields
if tool_result.get("data", {}).get(f) is None
]
if missing:
return GroundingResult(
status=GroundingStatus.PARTIAL,
missing_fields=missing,
unverified_fields=[],
evidence={"source": tool_call["connector"]}
)
# 3. 数值合理性校验
anomalies = self._check_numeric_anomalies(tool_result)
if anomalies:
return GroundingResult(
status=GroundingStatus.PARTIAL,
missing_fields=[],
unverified_fields=anomalies,
evidence={"source": tool_call["connector"]}
)
return GroundingResult(
status=GroundingStatus.GROUNDED,
missing_fields=[],
unverified_fields=[],
evidence={"source": tool_call["connector"]}
)
def _check_numeric_anomalies(self, result: dict) -> List[str]:
"""数值合理性校验:超出范围的值标记为待验证"""
anomalies = []
for field, value in result.get("data", {}).items():
if isinstance(value, (int, float)):
bounds = self.schema.get_numeric_bounds(field)
if bounds and not (bounds["min"] <= value <= bounds["max"]):
anomalies.append(field)
return anomalies
关键设计:空值不补全。接口超时或返回空值时,正确做法是标记“数据缺失”,而不是让模型“推断一个合理值”。这个规则必须写死在执行链路里,而不是靠 Prompt 约束模型“不要编”。
事实校验的判定路径可以概括为下面这张图:
3.2 事实与推断分栏输出
Agent 的输出必须区分“系统事实”和“模型推断”。推断必须附带证据链,无证据的推断不进入下游动作。
@dataclass
class AgentOutput:
"""Agent输出:事实与推断严格分离"""
facts: list[dict] # 来自工具调用的确定性数据
inferences: list[dict] # 模型推断,每条必须有evidence
confidence: float # 综合置信度 0-1
def is_executable(self, threshold: float = 0.8) -> bool:
"""是否可进入执行阶段"""
# 所有推断必须有证据
if any(not inf.get("evidence") for inf in self.inferences):
return False
# 置信度达标
return self.confidence >= threshold
四、第二层:多源比对——让“不一致”暴露出来
单源数据无法判断真伪。多源比对通过交叉验证,把“数据冲突”暴露出来。
class CrossSourceValidator:
"""多源比对:交叉验证与冲突检测"""
def __init__(self, sources: list):
self.sources = sources # [{"name": "crm", "connector": ...}, ...]
def validate(self, query: dict) -> dict:
"""从多个来源查询同一数据,比对结果"""
results = {}
for src in self.sources:
try:
results[src["name"]] = {
"available": True,
"data": src["connector"].query(query)
}
except Exception as e:
results[src["name"]] = {"available": False, "error": str(e)}
return self._compare(results)
def _compare(self, results: dict) -> dict:
"""比对多源结果"""
available = {k: v for k, v in results.items() if v["available"]}
if not available:
return {"status": "no_source", "action": "reject"}
if len(available) == 1:
return {
"status": "single_source",
"data": list(available.values())[0]["data"],
"confidence": 0.6,
"action": "flag_for_review"
}
# 多源比对
values = {k: v["data"] for k, v in available.items()}
if self._all_equal(values):
return {
"status": "consistent",
"data": list(values.values())[0],
"confidence": 0.95,
"action": "proceed"
}
else:
return {
"status": "conflict",
"values": values,
"confidence": 0.3,
"action": "human_review"
}
关键设计:冲突不自动裁决。多源数据不一致时,系统不“选一个看起来对的”,而是标记冲突、降低置信度、转人工。AI 可以建议“哪个更可信”,但最终判断由人做出。
多源比对的决策分支如下:
五、第三层:置信度门控——用阈值决定“走不走”
置信度门控是执行链路上的“闸门”:高置信度放行,中置信度标记,低置信度拦截。
from enum import Enum
class GateDecision(Enum):
PROCEED = "proceed" # 直接执行
PROCEED_WITH_FLAG = "flag" # 执行但标记
REQUIRE_REVIEW = "review" # 暂停,人工确认
BLOCK = "block" # 拦截,不执行
class ConfidenceGate:
"""置信度门控:按阈值决定执行策略"""
# 不同风险等级的动作,阈值不同
THRESHOLDS = {
"read_only": {"proceed": 0.7, "review": 0.5},
"write_low": {"proceed": 0.85, "review": 0.6},
"write_high": {"proceed": 0.95, "review": 0.8},
"irreversible": {"proceed": 0.99, "review": 0.9}
}
def decide(self, confidence: float, action_risk: str) -> GateDecision:
thresholds = self.THRESHOLDS.get(action_risk,
self.THRESHOLDS["write_low"])
if confidence >= thresholds["proceed"]:
return GateDecision.PROCEED
elif confidence >= thresholds["review"]:
return GateDecision.REQUIRE_REVIEW
elif confidence >= 0.4:
return GateDecision.PROCEED_WITH_FLAG
else:
return GateDecision.BLOCK
关键设计:风险等级决定阈值。只读查询的置信度阈值可以低一些,不可逆操作的阈值必须接近 1.0。同一个置信度,在不同风险等级的动作下,决策完全不同。
置信度门控的阈值分级可以用一张图看清:
六、第四层:Human-in-the-loop——不可逆动作的“人工刹车”
有些动作,AI 不应该单独做。不是因为它做不到,而是因为做错了代价不可逆。
class HITLGuard:
"""Human-in-the-loop守卫:不可逆动作的人工确认"""
IRREVERSIBLE_ACTIONS = {
"send_external_email", "approve_payment",
"delete_record", "modify_contract", "publish_content"
}
AMOUNT_THRESHOLD = 10000 # 金额阈值
def __init__(self, approval_service):
self.approval = approval_service
async def guard(self, action: str, params: dict,
agent_context: dict) -> dict:
"""判断是否需要人工确认"""
needs_confirm = (
action in self.IRREVERSIBLE_ACTIONS
or params.get("amount", 0) > self.AMOUNT_THRESHOLD
or agent_context.get("confidence", 1.0) < 0.9
)
if not needs_confirm:
return {"approved": True, "mode": "auto"}
# 创建审批请求,超时默认拒绝
request_id = self.approval.create(
agent_id=agent_context["agent_id"],
action=action,
params_summary=self._summarize(params),
timeout_hours=4,
default_on_timeout=False # 超时默认拒绝,安全优先
)
approved = await self.approval.wait(request_id)
return {"approved": approved, "mode": "hitl",
"request_id": request_id}
def _summarize(self, params: dict) -> str:
"""生成操作摘要,供人工确认"""
return f"操作参数:{params}"
关键设计:超时默认拒绝。高风险动作的“无人确认”应该被视为“不批准”。这是安全优先的基本逻辑。
七、第五层:Event-Driven 熔断——让异常自己“踩刹车”
单次幻觉可以通过前三层拦截。但“模式性异常”——比如某个 Agent 的失败率突然飙升、某个任务的置信度持续偏低——需要事件驱动的熔断机制。
class HallucinationCircuitBreaker:
"""幻觉熔断器:基于事件模式的自动止损"""
def __init__(self, event_bus, threshold: int = 3):
self.bus = event_bus
self.failure_count = 0
self.threshold = threshold
self.state = "closed"
# 订阅幻觉相关事件
self.bus.subscribe("grounding.ungrounded", self.on_ungrounded)
self.bus.subscribe("crosssource.conflict", self.on_conflict)
self.bus.subscribe("gate.blocked", self.on_blocked)
def on_ungrounded(self, event: dict):
"""无来源输出事件"""
self._record_anomaly(event, "ungrounded_output")
def on_conflict(self, event: dict):
"""多源冲突事件"""
self._record_anomaly(event, "source_conflict")
def on_blocked(self, event: dict):
"""置信度拦截事件"""
self._record_anomaly(event, "confidence_blocked")
def _record_anomaly(self, event: dict, anomaly_type: str):
self.failure_count += 1
if self.failure_count >= self.threshold:
self.state = "open"
# 发布熔断事件:暂停该 Agent 的自动执行
self.bus.publish("agent.hallucination_breaker_open", {
"agent_id": event["agent_id"],
"anomaly_count": self.failure_count,
"last_anomaly": anomaly_type,
"action": "suspend_and_escalate"
})
AIOps 的价值:熔断器订阅的是“异常事件流”,而不是“单次错误”。连续出现无来源输出、多源冲突、置信度拦截,说明这个 Agent 的当前状态不可信,应该暂停而不是继续执行。
熔断器的状态流转可以这样表示:
八、API 调用逻辑:拦截链路的接口设计
# 事实校验
POST /api/v1/grounding/check
Body: {"trace_id": "...", "tool_call": {...}, "tool_result": {...}}
Response: {"status": "grounded|partial|ungrounded", "missing_fields": [], "evidence": {...}}
# 多源比对
POST /api/v1/crosssource/validate
Body: {"query": {...}, "sources": ["crm", "erp"]}
Response: {"status": "consistent|conflict|single_source", "confidence": 0.95, "action": "proceed|review"}
# 置信度门控
POST /api/v1/gate/decide
Body: {"confidence": 0.72, "action_risk": "write_high"}
Response: {"decision": "review", "threshold": 0.95}
# 人工确认
POST /api/v1/hitl/request
Body: {"agent_id": "...", "action": "approve_payment", "params": {...}}
Response: {"request_id": "...", "status": "pending_approval"}
# 熔断状态查询
GET /api/v1/agents/{agent_id}/breaker
Response: {"state": "closed|open", "anomaly_count": 0, "last_anomaly": null}
九、工程实践要点
空值不补全,是幻觉拦截的第一原则。 接口超时或返回空值时,标记“数据缺失”,而不是让模型推断。这条规则要写死在执行链路里,不靠 Prompt 约束。
事实与推断分栏,推断必须带证据。 无证据的推断不进入下游动作。这是“可解释性”的执行层实现,不是文档里的原则。
多源冲突不自动裁决。 冲突时降低置信度、转人工。AI 可以建议“哪个更可信”,但判断由人做出。
置信度阈值按风险分级。 只读查询和不可逆操作,用同一个阈值是不负责任的。风险越高,阈值越接近 1.0。
不可逆动作必须 HITL,超时默认拒绝。 无人确认应被视为“不批准”,而不是“默认通过”。
熔断看模式,不看单次。 连续出现无来源输出、多源冲突、置信度拦截,说明 Agent 状态不可信。熔断器订阅事件流,自动暂停并升级。
拦截链路本身要可观测。 每一次拦截、每一个决策、每一次人工确认,都发布事件。AIOps 消费事件做趋势分析,拦截链路本身也需要被审计。
在行业实践中,类似沈管家这类全自动 AI 数字员工的任务执行引擎,已将状态持久化、异常分级处理和人工确认节点作为执行层的基础能力——任务执行有断点恢复,高风险操作保留人工确认,全链路审计按 trace_id 追溯。这种“编排层管拦截、执行层管隔离”的分工,是幻觉拦截能够真正生效的工程前提:拦截不是某个模型的“自觉”,而是执行链路上的结构性约束。
十、结语
大模型的幻觉无法被“彻底消除”,但可以被“结构性拦截”。
事实校验让“编造”进不了执行链路,多源比对让“不一致”暴露出来,置信度门控让“不确定”停下来,HITL 让“不可逆”由人拍板,事件驱动熔断让“模式性异常”自动止损。五层刹车叠加,才能让全自动执行从“敢演示”走向“敢生产”。
你们在 Agent 执行链路中设置了哪些“安全刹车”?是事实校验、多源比对,还是置信度门控?欢迎在评论区交流你们的拦截策略和踩坑经验。
更多推荐


所有评论(0)