一、全自动执行让幻觉的代价变了

大模型幻觉在对话场景中表现为“说错话”,在 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 表达这条拦截链路,结构会更清晰:

无来源 / 空值

有来源

冲突

一致 / 单源

低置信

高置信

是

否

批准

拒绝 / 超时

模式性异常

正常

Agent 执行引擎
规划 → 选工具 → 组装参数 → 调用

第一层:事实校验
Grounding Check

拦截:标记数据缺失

第二层:多源比对
Cross-Source Validation

降置信度 → 转人工

第三层:置信度门控
Confidence Gate

拦截:不执行

是否不可逆动作?

第四层:HITL 人工确认
超时默认拒绝

执行

第五层:事件驱动熔断
异常事件流 → 模式检测

自动熔断:暂停并升级

完成并发布审计事件

三、第一层:事实校验——不让“编造”进入执行链路

事实校验的核心是:模型输出的每一个事实性结论,都必须有可验证的来源。没有来源,不进入执行。

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 约束模型“不要编”。

事实校验的判定路径可以概括为下面这张图:

是

否

是

否

是

否

工具调用返回结果

接口是否报错 / 超时?

UNGROUNDED
标记数据缺失,禁止推断补全

必填字段是否为空?

PARTIAL
记录 missing_fields

数值是否超出合理范围?

PARTIAL
记录 unverified_fields

GROUNDED
允许进入下游

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 可以建议“哪个更可信”,但最终判断由人做出。

多源比对的决策分支如下:

0 个

1 个

≥2 个

一致

冲突

同一 query 向多个数据源查询

可用来源数量

no_source
直接拒绝

single_source
置信度 0.6 → 标记复核

各源结果是否一致?

consistent
置信度 0.95 → 放行

conflict
置信度 0.3 → 转人工,不自动裁决

五、第三层:置信度门控——用阈值决定“走不走”

置信度门控是执行链路上的“闸门”:高置信度放行,中置信度标记,低置信度拦截。

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。同一个置信度,在不同风险等级的动作下,决策完全不同。

置信度门控的阈值分级可以用一张图看清:

read_only

write_low

write_high

irreversible

置信度 confidence

动作风险等级

proceed ≥ 0.7
review ≥ 0.5

proceed ≥ 0.85
review ≥ 0.6

proceed ≥ 0.95
review ≥ 0.8

proceed ≥ 0.99
review ≥ 0.9

PROCEED / FLAG / REVIEW / BLOCK

六、第四层: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 的当前状态不可信,应该暂停而不是继续执行。

熔断器的状态流转可以这样表示:

记录异常事件
failure_count + 1

failure_count ≥ threshold

发布 agent.hallucination_breaker_open
暂停自动执行并升级

人工介入修复后重置

Closed

Open

八、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 执行链路中设置了哪些“安全刹车”?是事实校验、多源比对,还是置信度门控?欢迎在评论区交流你们的拦截策略和踩坑经验。

Logo

更多推荐