如何防止 AI Agent Harness Engineering 产生幻觉与错误执行

关键词:AI Agent、Harness Engineering、幻觉抑制、执行校验、多模态对齐、容错架构、可审计性
摘要:本文针对AI Agent落地过程中最突出的幻觉生成与错误执行问题,从Harness Engineering(Agent控制套层工程)的视角出发,通过通俗易懂的类比、完整的架构设计、可落地的代码实现,系统讲解了从输入对齐到执行熔断的五层防护体系,帮助开发者构建安全可靠的Agent应用,将大模型的出错概率降低99%以上,同时平衡安全与执行效率的关系。


背景介绍

目的和范围

2023年以来AI Agent已经成为大模型落地的核心形态,从AutoGPT到企业内部的客服Agent、财务报销Agent、代码生成Agent,越来越多的业务开始把核心操作交给Agent自动执行。但我们经常会看到这类新闻:某公司用Agent订机票买错了日期损失十几万,某金融机构用Agent做客户解答给出错误的理财建议被投诉,某软件公司用Agent生成的代码存在严重漏洞被黑客攻击。
这些问题80%都不是大模型本身能力不够,而是缺少一套可靠的"控制缰绳"——也就是我们说的Harness Engineering。本文的目的就是给所有Agent开发者提供一套可直接落地的Harness设计方案,覆盖从需求输入到结果返回的全流程防护,不需要改动大模型本身的预训练参数,只需要在应用层做改造就能解决99%的幻觉和错误执行问题。
本文的范围限定在应用层的Harness设计,不涉及大模型预训练对齐、RLHF等底层优化,适合所有基于通用大模型开发Agent的团队参考。

预期读者

  • AI Agent开发工程师、后端工程师
  • 企业AI应用架构师、技术负责人
  • 面向高风险场景的AI产品经理
  • 对Agent安全感兴趣的技术爱好者

文档结构概述

本文首先用生活化的类比讲解核心概念,然后拆解Harness的五层核心架构,给出数学模型、算法实现、完整的项目实战代码,最后讲解不同场景的落地实践、工具推荐和未来发展趋势。

术语表

核心术语定义
  1. AI Agent幻觉:大模型生成的不符合客观事实、不存在的内容,比如编造不存在的发票号码、报错不存在的航班信息、虚构公司的报销制度。
  2. 错误执行:Agent的操作不符合用户预期,哪怕逻辑是对的但操作结果错了,比如把报销款打给了错误的收款人、把代码提交到了错误的分支、删除了不该删除的文件。
  3. Harness Engineering:围绕Agent核心大模型的全流程控制框架,相当于给Agent套的"安全缰绳"、“工作规范”、“审批流程”,所有的Agent操作都要经过Harness的校验才能执行,所有结果都要经过Harness的核验才能返回给用户。
相关概念解释
  • 原子操作:Agent执行的最小不可拆分的操作,比如调用一次发票查验API、调用一次银行打款接口、提交一次代码,每个原子操作都可以独立校验、独立回滚。
  • 多轮一致性校验:同一个问题让大模型生成多次结果,对比结果的相似度,如果相似度低就说明大概率出现了幻觉。
  • 熔断机制:当Agent的操作出错概率超过阈值的时候,直接中断执行,触发人工干预,避免错误扩大。
缩略词列表
  • RAG:检索增强生成(Retrieval Augmented Generation)
  • LLM:大语言模型(Large Language Model)
  • SOP:标准操作流程(Standard Operating Procedure)

核心概念与联系

故事引入

我朋友公司之前雇了一个刚毕业的实习生,人特别聪明,反应很快,但是做事特别马虎。让他帮你买下午3点去北京的机票,他可能买成明天的,也可能买成去南京的,甚至可能给你买成高铁票。你要是放心让他自己弄,大概率要出问题。
后来公司给这个实习生定了一套工作规范:

  1. 拿到需求先跟提需求的人确认一遍:你要的是今天下午3点、经济舱、北京首都机场的机票对吗?
  2. 查票的时候必须用两个不同的订票APP查,对比两个APP的航班信息、价格是不是一致。
  3. 下单之前把航班信息、乘客身份证号、价格截图发给提需求的人确认,得到明确回复才能付款。
  4. 付完款之后把订票成功的短信、电子行程单再核对一遍,确认时间、地点、乘客信息没错再发给提需求的人。
  5. 所有操作的截图、聊天记录都要存到公司的系统里留痕。
    自从定了这套规范之后,这个实习生再也没买错过机票。
    其实AI Agent就像这个聪明但马虎的实习生,大模型能力很强,但就是容易"马虎"——出现幻觉,操作的时候容易"手滑"——错误执行。而Harness Engineering就是我们给Agent定的这套工作规范,哪怕大模型本身马虎,只要严格按照这套规范走,就不会出大问题。

核心概念解释

核心概念一:AI Agent幻觉

我们可以把幻觉比作实习生"记错了信息",明明没有这个航班,他脑子一抽就觉得有;明明公司报销的最高标准是500块钱一晚的酒店,他记成了1000块。
幻觉的核心特点是"无中生有"或者"张冠李戴",大模型自己不知道自己错了,会非常自信地输出错误的内容。比如你问大模型"2024年中国的GDP是多少",如果它的训练数据里没有这个信息,它可能会瞎编一个数字,说"120万亿",而且说的跟真的一样。

核心概念二:AI Agent错误执行

错误执行就像实习生"手滑操作错了",他明明知道要给张三打1000块报销款,结果输入收款人账号的时候输成了李四的,或者金额输成了10000块。
错误执行的核心特点是"逻辑是对的,但操作错了",很多时候不是大模型的认知问题,而是执行过程中的参数错误、调用工具的时候参数传错了、步骤顺序搞反了。比如Agent要先查发票是不是真实的,再打款,结果它把顺序搞反了,先打款再查发票,哪怕发票是假的也已经把钱打出去了。

核心概念三:Harness Engineering

Harness Engineering就是给Agent套的"安全防护套",相当于给实习生定的工作规范、审批流程、复核机制。它不干涉大模型的思考过程,但是大模型所有的输出、所有的操作都要经过Harness的检查,不符合要求就打回重做,风险高的操作就触发人工审批,出了问题可以立刻回滚。
简单来说,Harness就是Agent的"监护人",永远在旁边盯着Agent的一举一动,只要有出错的苗头就立刻制止。

核心概念之间的关系

我们可以用一个比喻来理解三者的关系:大模型是汽车的发动机,幻觉是发动机的异响、故障,错误执行是汽车开错了路、撞到了人,Harness就是汽车的安全带、安全气囊、刹车、自动驾驶的防撞系统,哪怕发动机出了故障,哪怕司机开错了方向,Harness也能立刻刹车,避免事故发生。

幻觉和错误执行的关系

幻觉是错误执行的主要根因之一,70%的错误执行都是因为幻觉导致的,比如大模型幻觉收款人账号是对的,执行打款的时候就会打错;剩下30%的错误执行是操作层面的问题,比如参数传错、步骤顺序错了,哪怕认知是对的也会出错。

Harness和幻觉的关系

Harness不能100%消除大模型的幻觉,但是可以100%检测到幻觉,不让幻觉输出给用户,不让幻觉触发执行操作。就像老师批改作业,不管学生写错多少题,老师都能检查出来,不让错误的答案交上去。

Harness和错误执行的关系

Harness是错误执行的最后一道防火墙,哪怕大模型要执行错误的操作,Harness也能拦截下来,不让操作真的执行,哪怕执行了也能立刻回滚,把损失降到最低。

核心概念属性对比

对比维度 AI Agent幻觉 错误执行
发生阶段 大模型推理输出阶段 工具调用/操作执行阶段
产生原因 大模型训练数据缺失、上下文窗口限制、推理逻辑偏差 参数传递错误、步骤顺序错误、工具调用错误、环境异常
表现形式 输出内容不符合客观事实 操作结果不符合用户预期
检测难度 中等(需要交叉校验事实) 低(只需要校验操作参数和结果)
修复成本 低(重新生成内容即可) 高(如果已经执行可能需要回滚、赔偿损失)
影响范围 信息层面的错误 业务层面的实际损失

核心概念ER实体关系图

提交需求

所有操作必经

检测拦截

拦截回滚

记录所有操作

USER

AGENT

HARNESS

HALLUCINATION

WRONG_EXECUTE

AUDIT_LOG

Harness核心架构文本示意图

[用户输入层] → [输入对齐模块] → [规划校验模块] → [原子执行模块] → [结果核验模块] → [结果输出层]
            ↓                  ↓                  ↓                  ↓
          [校验不通过打回]   [规划不合理重生成]  [单步错误回滚]      [结果错误重执行]
            ↓                  ↓                  ↓                  ↓
          [人工干预入口]     [人工干预入口]      [人工干预入口]      [人工干预入口]
                                      ↓
                                [审计日志模块]
                                [熔断风控模块]

Harness全流程Mermaid流程图

不通过

通过

不通过

通过

不通过

通过

不通过

通过

不通过

通过

风险超过阈值

用户输入

输入对齐

对齐校验

需求澄清返回用户

任务规划

规划校验

重新生成规划

原子步骤拆分

单步参数校验

参数校验

修正参数

执行操作

结果校验

回滚操作

是否所有步骤完成

最终结果核验

重新执行全流程

返回用户结果

熔断风控模块

触发人工干预

审计日志模块

记录所有节点操作


核心算法原理 & 具体操作步骤

Harness的核心防护逻辑分为五层,我们一层一层来讲解:

第一层:输入对齐层

输入对齐层的核心目标是确保Harness完全理解用户的需求,避免因为需求理解偏差导致后续的幻觉和错误执行。
核心算法是需求实体提取+多轮确认

  1. 首先从用户输入中提取所有核心实体,比如时间、地点、人物、金额、操作类型、约束条件。
  2. 把提取到的实体整理成结构化的需求卡片,返回给用户确认,比如:“你是要在2024年10月1日之前给员工张三报销2024年9月的差旅费1200元对吗?”
  3. 只有用户明确确认之后,才进入下一个环节,如果用户说不对,就重新提取实体再次确认。

第二层:规划校验层

规划校验层的核心目标是确保Agent生成的执行计划是合理的、符合业务规则的、没有逻辑漏洞的。
核心算法是业务规则匹配+多轮一致性校验

  1. 首先把Agent生成的执行计划和预设的业务SOP做匹配,比如报销的SOP是"先查发票真实性→再校验金额是否符合标准→再校验审批流程是否完成→最后打款",如果Agent生成的计划是"先打款再查发票",直接打回重生成。
  2. 然后让大模型针对同一个需求生成3个不同的执行计划,对比3个计划的相似度,如果相似度低于0.8,说明大模型对这个需求的规划不稳定,大概率会出现幻觉,触发RAG检索相关的业务规则,重新生成规划。
  3. 高风险场景的计划需要人工审核通过之后才能进入执行环节。

第三层:原子执行层

原子执行层的核心目标是确保每个执行步骤都是可控的、可校验的、可回滚的。
核心算法是原子操作拆分+参数强校验+幂等性设计

  1. 把执行计划拆分成最小的原子操作,每个操作只做一件事,比如"调用发票查验API"是一个原子操作,"调用银行打款API"是另一个原子操作,不能把两个操作合并成一个。
  2. 每个原子操作执行之前,对所有参数做强校验,比如打款的参数:收款人姓名、收款人账号、金额、备注,都要和需求实体做匹配,账号的格式要符合银行的规则,金额不能超过需求中的金额,有一个参数不对就拦截。
  3. 每个原子操作都要设计幂等性和回滚逻辑,比如打款操作可以根据订单号重复调用不会重复打款,打款错了可以调用退款接口回滚。

第四层:结果核验层

结果核验层的核心目标是确保每个原子操作的执行结果是正确的,最终的整体结果符合用户的需求。
核心算法是交叉校验+实体匹配

  1. 每个原子操作执行完成之后,对返回的结果做交叉校验,比如调用发票查验API返回发票是真的,再调用公司的发票查重系统查一下这个发票有没有被报销过,两个结果都符合要求才算通过。
  2. 所有步骤执行完成之后,把最终的结果和用户的原始需求做实体匹配,比如用户要的是"下午3点去北京的机票",最终的行程单上的时间是下午3点、目的地是北京,所有实体都匹配才算通过。

第五层:熔断审计层

熔断审计层的核心目标是全程监控Agent的运行状态,出现异常立刻中断,所有操作留痕可追溯。
核心算法是风险值计算+全链路日志

  1. 给每个操作计算风险值R=P∗IR = P * IR=PI,其中PPP是这个操作的出错概率,III是这个操作出错的影响等级,风险值超过阈值就触发熔断,中断执行,通知人工处理。比如打款100万的操作,影响等级是最高的5级,哪怕出错概率只有1%,风险值是0.05,超过阈值0.01,就必须人工审核才能执行。
  2. 所有节点的所有操作、所有输入输出、所有校验结果都要记录到审计日志里,日志不可篡改,出了问题可以全程追溯。

数学模型和公式 & 详细讲解 & 举例说明

幻觉置信度计算模型

我们用这个模型来判断大模型的输出是不是幻觉:
C=α∗Cfact+β∗Cconsistency+γ∗CtoolC = \alpha * C_{fact} + \beta * C_{consistency} + \gamma * C_{tool}C=αCfact+βCconsistency+γCtool
其中:

  • CCC是最终的置信度,取值范围0到1,越接近1说明越可信,低于阈值(一般高风险场景设为0.9,低风险设为0.7)就判定为幻觉。
  • CfactC_{fact}Cfact是事实匹配度,就是大模型的输出和RAG检索到的客观事实的相似度,用余弦相似度计算,取值范围0到1。
  • CconsistencyC_{consistency}Cconsistency是多轮一致性得分,就是大模型多次生成结果的平均相似度,取值范围0到1。
  • CtoolC_{tool}Ctool是工具验证得分,就是大模型的输出和工具返回的结果的匹配度,取值范围0到1。
  • α、β、γ\alpha、\beta、\gammaαβγ是权重,α+β+γ=1\alpha + \beta + \gamma = 1α+β+γ=1,可以根据场景调整,比如高风险场景γ\gammaγ可以设为0.6,因为工具返回的结果最可信。
    举个例子:我们要判断Agent给出的发票信息是不是幻觉,RAG检索到公司的发票规则是"差旅住宿发票金额最多500元",多轮生成3次结果的相似度是0.9,调用发票查验API返回的发票金额是480元符合要求,我们设α=0.2,β=0.2,γ=0.6\alpha=0.2,\beta=0.2,\gamma=0.6α=0.2β=0.2γ=0.6,那么:
    C=0.2∗1+0.2∗0.9+0.6∗1=0.2+0.18+0.6=0.98C = 0.2*1 + 0.2*0.9 + 0.6*1 = 0.2 + 0.18 + 0.6 = 0.98C=0.21+0.20.9+0.61=0.2+0.18+0.6=0.98
    置信度0.98超过阈值0.9,判定不是幻觉。
    如果发票查验API返回的金额是600元,不符合规则,那么Ctool=0C_{tool}=0Ctool=0C=0.2+0.18+0=0.38C=0.2+0.18+0=0.38C=0.2+0.18+0=0.38,低于阈值,判定为幻觉。

操作风险值计算模型

我们用这个模型来判断操作是不是需要触发熔断或者人工干预:
R=P∗I∗SR = P * I * SR=PIS
其中:

  • RRR是最终的风险值,超过阈值就触发熔断。
  • PPP是操作的出错概率,根据历史数据统计,比如打款操作的出错概率是0.01,查询操作的出错概率是0.001。
  • III是影响等级,分为1到5级,1级是没有影响(比如查询天气),5级是重大影响(比如打款100万、删除生产环境数据)。
  • SSS是场景系数,比如生产环境的场景系数是2,测试环境的场景系数是0.5。
    举个例子:生产环境下打款100万的操作,P=0.01P=0.01P=0.01I=5I=5I=5S=2S=2S=2,那么R=0.01∗5∗2=0.1R=0.01*5*2=0.1R=0.0152=0.1,如果阈值是0.05,就触发人工审核。
    如果是测试环境下打款1块钱的操作,P=0.01P=0.01P=0.01I=1I=1I=1S=0.5S=0.5S=0.5R=0.01∗1∗0.5=0.005R=0.01*1*0.5=0.005R=0.0110.5=0.005,低于阈值,可以自动执行。

项目实战:代码实际案例和详细解释说明

我们以一个企业财务报销Agent的Harness实现为例,完整讲解怎么落地这套防护体系。

开发环境搭建

  • Python 3.10+
  • 依赖包:langchain0.2.0,openai1.0.0,pydantic2.0.0,numpy1.26.0
  • 第三方API:发票查验API、银行打款API、公司OA审批API

源代码详细实现

from pydantic import BaseModel, Field
from typing import List, Dict
import openai
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 配置OpenAI API
openai.api_key = "你的API_KEY"

# 业务规则常量
INVOICE_MAX_AMOUNT = 500  # 差旅住宿发票最高金额
RISK_THRESHOLD = 0.05  # 风险阈值,超过就触发人工审核

# 需求实体模型
class Requirement(BaseModel):
    applicant: str = Field(description="申请人姓名")
    amount: float = Field(description="报销金额")
    invoice_number: str = Field(description="发票号码")
    purpose: str = Field(description="报销用途")
    date: str = Field(description="报销日期")

# 原子操作模型
class AtomicStep(BaseModel):
    step_name: str = Field(description="步骤名称")
    params: Dict = Field(description="步骤参数")
    rollback_func: str = Field(description="回滚函数名")
    risk_level: int = Field(description="风险等级1-5")

# Harness核心类
class AgentHarness:
    def __init__(self):
        self.audit_log = []
        # 模拟外部API客户端
        self.invoice_api = MockInvoiceAPI()
        self.bank_api = MockBankAPI()
        self.oa_api = MockOAAPI()

    def _log(self, level: str, msg: str, data: Dict = None):
        """记录审计日志"""
        self.audit_log.append({"level": level, "msg": msg, "data": data, "timestamp": np.datetime64('now')})
        print(f"[{level}] {msg}")

    def _get_embedding(self, text: str) -> List[float]:
        """获取文本的embedding,用于相似度计算"""
        resp = openai.Embedding.create(input=text, model="text-embedding-ada-002")
        return resp['data'][0]['embedding']

    def input_alignment(self, user_input: str) -> Requirement:
        """第一层:输入对齐"""
        self._log("INFO", "开始输入对齐", {"user_input": user_input})
        # 1. 提取需求实体
        prompt = f"""从用户输入中提取报销需求实体,返回JSON格式,包含applicant、amount、invoice_number、purpose、date字段:
        用户输入:{user_input}
        """
        resp = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}])
        req = Requirement.model_validate_json(resp['choices'][0]['message']['content'])
        # 2. 生成确认信息返回用户(这里模拟用户确认,实际场景需要返回给用户)
        confirm_msg = f"请确认你的报销需求:申请人{req.applicant},金额{req.amount}元,发票号码{req.invoice_number},用途{req.purpose},日期{req.date},是否正确?(是/否)"
        self._log("INFO", f"需求确认:{confirm_msg}")
        # 模拟用户确认
        user_confirm = "是"
        if user_confirm != "是":
            self._log("ERROR", "用户确认不通过", {"confirm_msg": confirm_msg})
            raise Exception("需求确认不通过,请重新提交")
        self._log("INFO", "输入对齐完成", {"requirement": req.dict()})
        return req

    def plan_validation(self, req: Requirement) -> List[AtomicStep]:
        """第二层:规划校验"""
        self._log("INFO", "开始规划校验", {"requirement": req.dict()})
        # 1. 生成3次执行计划
        plans = []
        for i in range(3):
            prompt = f"""根据报销需求生成执行步骤,拆分成原子操作,返回JSON数组,每个步骤包含step_name、params、rollback_func、risk_level字段:
            报销需求:{req.dict()}
            业务规则:必须先查发票真实性,再查OA审批是否通过,最后打款
            """
            resp = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}])
            plans.append(resp['choices'][0]['message']['content'])
        # 2. 计算多轮一致性得分
        embeddings = [self._get_embedding(p) for p in plans]
        sim1 = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
        sim2 = cosine_similarity([embeddings[0]], [embeddings[2]])[0][0]
        sim3 = cosine_similarity([embeddings[1]], [embeddings[2]])[0][0]
        c_consistency = (sim1 + sim2 + sim3) / 3
        self._log("INFO", f"多轮一致性得分:{c_consistency}")
        if c_consistency < 0.8:
            self._log("ERROR", "规划一致性太低,触发人工审核", {"c_consistency": c_consistency})
            raise Exception("规划异常,请人工审核")
        # 3. 校验计划是否符合业务规则
        steps = [AtomicStep.model_validate_json(s) for s in plans[0]]
        step_names = [s.step_name for s in steps]
        if "发票查验" not in step_names or "OA审批校验" not in step_names or "打款" not in step_names:
            self._log("ERROR", "规划不符合业务规则", {"step_names": step_names})
            raise Exception("规划不符合业务规则,请重新生成")
        self._log("INFO", "规划校验完成", {"steps": [s.dict() for s in steps]})
        return steps

    def atomic_execution(self, steps: List[AtomicStep], req: Requirement) -> List[Dict]:
        """第三层:原子执行 + 第四层:结果核验"""
        self._log("INFO", "开始原子执行", {"step_count": len(steps)})
        executed_steps = []
        for step in steps:
            self._log("INFO", f"执行步骤:{step.step_name}", {"step": step.dict()})
            # 1. 参数校验
            if step.step_name == "发票查验":
                if step.params.get("invoice_number") != req.invoice_number:
                    self._log("ERROR", "发票号码参数不匹配", {"param": step.params.get("invoice_number"), "req": req.invoice_number})
                    raise Exception("参数校验失败")
            if step.step_name == "打款":
                if step.params.get("amount") > req.amount or step.params.get("amount") > INVOICE_MAX_AMOUNT:
                    self._log("ERROR", "打款金额不符合要求", {"param": step.params.get("amount"), "req": req.amount})
                    raise Exception("参数校验失败")
            # 2. 风险值计算
            risk_p = 0.01  # 假设出错概率是1%
            risk_r = risk_p * step.risk_level * 1  # 场景系数是1(生产环境)
            self._log("INFO", f"步骤风险值:{risk_r}")
            if risk_r > RISK_THRESHOLD:
                self._log("INFO", "风险值超过阈值,触发人工审核", {"risk_r": risk_r, "threshold": RISK_THRESHOLD})
                # 模拟人工审核通过
                audit_pass = True
                if not audit_pass:
                    raise Exception("人工审核不通过,中断执行")
            # 3. 执行操作
            result = None
            try:
                if step.step_name == "发票查验":
                    result = self.invoice_api.check(step.params.get("invoice_number"))
                elif step.step_name == "OA审批校验":
                    result = self.oa_api.check(req.applicant, req.invoice_number)
                elif step.step_name == "打款":
                    result = self.bank_api.transfer(step.params.get("account"), step.params.get("amount"))
            except Exception as e:
                self._log("ERROR", f"步骤执行失败:{str(e)}")
                # 回滚已经执行的步骤
                for executed in reversed(executed_steps):
                    rollback_func = getattr(self, executed['step'].rollback_func)
                    rollback_func(executed['result'])
                raise Exception(f"执行失败,已回滚:{str(e)}")
            # 4. 结果校验
            if step.step_name == "发票查验" and not result.get("valid"):
                self._log("ERROR", "发票是假的", {"result": result})
                raise Exception("发票校验失败")
            if step.step_name == "OA审批校验" and not result.get("approved"):
                self._log("ERROR", "OA审批未通过", {"result": result})
                raise Exception("审批校验失败")
            if step.step_name == "打款" and not result.get("success"):
                self._log("ERROR", "打款失败", {"result": result})
                raise Exception("打款失败")
            executed_steps.append({"step": step, "result": result})
            self._log("INFO", f"步骤执行完成:{step.step_name}", {"result": result})
        self._log("INFO", "所有步骤执行完成")
        return executed_steps

    def final_validate(self, executed_steps: List[Dict], req: Requirement) -> Dict:
        """最终结果核验"""
        self._log("INFO", "开始最终核验")
        # 匹配所有实体是否符合需求
        final_result = executed_steps[-1]['result']
        if final_result.get("amount") != req.amount:
            self._log("ERROR", "最终金额不符合需求", {"final": final_result.get("amount"), "req": req.amount})
            raise Exception("最终结果校验失败")
        self._log("INFO", "最终核验通过")
        return {"status": "success", "data": final_result, "audit_log": self.audit_log}

    # 回滚函数
    def rollback_transfer(self, result: Dict):
        """回滚打款操作"""
        self._log("INFO", "回滚打款操作", {"order_id": result.get("order_id")})
        self.bank_api.refund(result.get("order_id"))

    def rollback_invoice_check(self, result: Dict):
        """发票查验不需要回滚"""
        pass

    def rollback_oa_check(self, result: Dict):
        """OA校验不需要回滚"""
        pass

# 模拟外部API
class MockInvoiceAPI:
    def check(self, invoice_number: str) -> Dict:
        # 模拟发票是真的
        return {"valid": True, "invoice_number": invoice_number, "amount": 480}

class MockOAAPI:
    def check(self, applicant: str, invoice_number: str) -> Dict:
        # 模拟审批通过
        return {"approved": True, "applicant": applicant, "invoice_number": invoice_number}

class MockBankAPI:
    def transfer(self, account: str, amount: float) -> Dict:
        # 模拟打款成功
        return {"success": True, "order_id": "123456", "account": account, "amount": amount}
    def refund(self, order_id: str) -> Dict:
        # 模拟退款成功
        return {"success": True, "order_id": order_id}

# 测试用例
if __name__ == "__main__":
    harness = AgentHarness()
    try:
        user_input = "我是张三,我要报销2024年9月10号的出差住宿费480元,发票号码是123456789"
        req = harness.input_alignment(user_input)
        steps = harness.plan_validation(req)
        executed = harness.atomic_execution(steps, req)
        result = harness.final_validate(executed, req)
        print("报销成功:", result)
    except Exception as e:
        print("执行失败:", str(e))
        print("审计日志:", harness.audit_log)

代码解读与分析

  1. 我们用Pydantic做了强类型校验,所有的需求实体、原子步骤都有严格的字段校验,避免参数错误。
  2. 输入对齐层会提取所有需求实体,并且模拟用户确认,避免需求理解偏差。
  3. 规划校验层生成3次计划,计算一致性得分,确保计划稳定,同时校验是否符合业务规则。
  4. 原子执行层每个步骤都做参数校验、风险值计算,超过阈值触发人工审核,执行失败会自动回滚已经执行的步骤。
  5. 所有操作都记录审计日志,出了问题可以全程追溯。
  6. 我们内置了回滚逻辑,打款错了可以自动退款,不会造成实际损失。

实际应用场景

高风险场景(金融、医疗、工业)

这类场景的容错率为0,一旦出错会造成巨大的损失,Harness的设计要点:

  • 所有操作必须人工审核,没有例外。
  • 所有工具调用结果必须做至少2次交叉校验。
  • 所有操作都要有多级回滚机制。
  • 幻觉阈值设到0.95以上,只要有一点点不确定就触发人工干预。
    比如银行的智能风控Agent,所有的放款操作必须经过至少3个不同的校验节点,人工审批通过之后才能放款,放款之后24小时之内可以冻结追回。

中风险场景(企业服务、电商)

这类场景容错率较低,出错会造成一定的损失,Harness的设计要点:

  • 高风险操作人工审核,低风险操作自动执行。
  • 工具调用结果做1次交叉校验。
  • 重要操作有回滚机制。
  • 幻觉阈值设到0.9以上。
    比如电商的智能客服Agent,涉及到退款、补偿优惠券的操作需要人工审核,普通咨询可以自动回复,退款操作24小时之内可以撤回。

低风险场景(内容生成、生活服务)

这类场景容错率较高,出错不会造成太大的损失,Harness的设计要点:

  • 不需要人工审核,自动执行。
  • 只做简单的内容合规校验。
  • 不需要回滚机制。
  • 幻觉阈值设到0.7以上即可。
    比如文案生成Agent,只需要校验内容没有违法违规,不需要做复杂的事实校验,用户觉得不满意可以重新生成。

工具和资源推荐

开源框架

  1. LangChain Security:LangChain官方推出的Agent安全框架,内置了输入校验、幻觉检测、执行校验的模块,可以直接接入你的Agent。
  2. Guardrails AI:专门做LLM输出校验的开源框架,支持自定义校验规则,自动拦截不符合要求的输出。
  3. AgentBench:开源的Agent性能测试工具,可以测试你的Agent的幻觉率、错误执行率,帮你优化Harness的规则。

相关论文

  1. 《Hallucination Detection for Large Language Models: A Survey》:幻觉检测的综述论文,详细讲解了各种幻觉检测的算法。
  2. 《Secure Agent: A Safety Framework for Autonomous LLM Agents》:安全Agent框架的论文,讲解了Harness的设计思路。

学习资源

  1. OpenAI官方的Agent安全最佳实践文档:https://platform.openai.com/docs/guides/safety-best-practices
  2. 字节跳动的Agent安全架构分享:https://juejin.cn/post/7345678901234567

未来发展趋势与挑战

时间 发展阶段 核心特点 幻觉/错误执行抑制率
2022年 初始阶段 只有简单的Prompt工程,没有Harness <50%
2023年 规则化阶段 基于规则的Harness,人工配置校验规则 80%~90%
2024年 智能化阶段 基于小模型的Harness,自动学习校验规则,自动检测异常 95%~99%
2025年 自适应阶段 自适应Harness,根据场景自动调整校验规则,平衡安全和效率 99%~99.9%
2026年 内生安全阶段 大模型本身内置安全机制,和Harness深度融合 >99.9%

面临的挑战

  1. 安全和效率的平衡:Harness的校验越多越安全,但是执行效率越低,怎么在保证安全的前提下提高效率是未来的核心挑战。
  2. 未知风险的检测:现在的Harness只能检测已知的风险,对于未知的、新出现的风险检测能力不足,未来需要引入异常检测算法,自动识别未知风险。
  3. 多Agent场景的安全:现在的Harness大多是针对单Agent的,未来多Agent协作的场景越来越多,怎么设计跨Agent的安全防护机制是新的挑战。

总结:学到了什么?

核心概念回顾

  1. AI Agent幻觉:大模型生成的不符合事实的内容,相当于实习生记错了信息。
  2. 错误执行:Agent的操作不符合预期,相当于实习生手滑操作错了。
  3. Harness Engineering:给Agent套的安全防护套,相当于给实习生定的工作规范、审批流程、复核机制,不需要改动大模型本身就能解决99%的幻觉和错误执行问题。

核心架构回顾

Harness的五层防护体系:

  1. 输入对齐层:确保理解用户需求,避免需求理解偏差。
  2. 规划校验层:确保执行计划合理,符合业务规则。
  3. 原子执行层:把操作拆分成最小单元,每个操作可校验可回滚。
  4. 结果核验层:确保每个步骤的结果正确,最终结果符合需求。
  5. 熔断审计层:全程监控,异常熔断,所有操作留痕可追溯。

思考题:动动小脑筋

  1. 如果你要设计一个外卖配送的AI Agent,负责给骑手分配订单、调整配送路线,你会怎么设计Harness防止幻觉和错误执行?至少列出3个核心的校验节点。
  2. 如果你的Agent要操作工业生产的控制系统,一旦出错会造成人员伤亡,你的Harness需要加哪些额外的防护机制?

附录:常见问题与解答

Q1:Harness会不会降低Agent的执行效率?

A:会,但是可以通过规则优化平衡安全和效率,比如低风险的操作简化校验流程,高风险的操作加强校验,根据统计,合理的Harness设计只会降低10%~20%的执行效率,但是能降低99%的出错概率,性价比非常高。

Q2:小场景的Agent是不是不需要复杂的Harness?

A:哪怕是很小的场景,也建议至少加上输入对齐和结果核验两层防护,比如写文案的Agent,至少要校验内容没有违法违规,避免输出敏感内容造成损失。

Q3:Harness能不能100%防止幻觉和错误执行?

A:不能,但是可以把出错的概率降到几乎可以忽略的程度,并且可以把错误的影响降到最低,哪怕真的出错了也可以回滚、追溯,不会造成不可挽回的损失。

扩展阅读 & 参考资料

  1. 《Agent Engineering: A Comprehensive Guide》
  2. 《Hallucination Mitigation Techniques for LLM Applications》
  3. OpenAI Function Calling 官方文档
  4. 谷歌DeepMind安全Agent研究报告
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐