智能体系统中的人类反馈与容错机制

Human Feedback and Fault Tolerance in Agentic Systems

智能体系统能够自主处理查询库存、管理购物车等复杂任务，但同时也对系统可靠性以及行为是否符合人类预期提出了更高要求。单步错误假设、虚构的工具调用或突发的外部接口故障，都可能引发多步连锁错误。除了实时的人机协同（Human-in-the-Loop，HITL），生产级智能体还需要两类核心机制：人类反馈循环（Human Feedback Loops）与容错机制（Fault Tolerance）。来自终端用户、开发者或领域专家的反馈能够提供纠偏信号与优化方向；而重试、回滚撤销等容错模式则使系统能够妥善应对运行异常，维持工作流稳定。二者结合，有助于管控智能体行为的不确定性并驱动持续迭代。

智能体系统中的人类反馈循环

1. 人类反馈的作用

即便能力较强的智能体，仍可能误解指令、检索到无关信息、误用工具或产出低质量回答。引入人类判断对智能体的操作与输出进行审查、纠偏或打分，是许多实际 AI 应用保障质量的关键环节，有助于确保智能体决策有据可依、切实可用且始终处于受控范围内。人类反馈按时间维度可分为两个层面：在执行期间或刚结束时立即介入纠正，以及利用长期积累的反馈数据迭代优化模型或提示词。对于需要持续学习与适应的智能体而言，这两个层面缺一不可。

2. 人类反馈的主要形式

智能体系统中的人类反馈通常包括以下形式。

  • 交互式用户反馈：用户在交互过程中通过点赞、点踩、星级评分或直接输入纠正指令（如告诉购物助手“这与我的意思不符”，或从多个候选中点选更合适的一项）提供实时反馈。此外，当用户被迫重新措辞提问或补充说明“我问的是蓝色保护套，颜色不对”时，这种隐式信号同样表明首轮意图理解存在偏差。系统应完整记录这些显式与隐式反馈——凡是迫使用户主动纠偏的执行轨迹，都是值得重点复盘的理解失败样本。
  • 转交人工处理（Escalation）：当智能体对下一步操作把握不足，或识别到即将执行的动作影响重大（如购物助手准备发放高额折扣，或无法有把握地解答复杂的商品兼容问题）时，主动请求人工批准或将对话转交人工客服。设置“请求人工帮助”分支是一道常用的安全防线，既能在低置信度或模糊场景下保障用户体验，人工客服给出的标准解答还可沉淀为后续的训练样本。
  • 离线标注与日志审核：在任务执行结束后，由开发者或分析人员结合可观测性平台记录的对话日志、工具调用参数与逐步决策轨迹开展离线复盘，标记推理偏差或不当操作（例如指出某次工具调用纯属多余，或最终回答遗漏了核心条件）。结构化的离线审核标注与评分可直接转化为高质量的基准评估数据集。

3. 支持反馈的记录机制与可观测性

在课程工程链路中，本地 JSONL 文件用于保存测试输入、参考答案和授权资料 ID；自托管的 Langfuse 则通过 dataset items 和 experiments 对比不同版本表现，并记录脱敏后的运行追踪、人工反馈与错误详情。实验归档时，数据集的 input / expected_output、提示词整数版本号（version）与底层语料 hash 均需分别独立保存。

4. 利用反馈持续改进系统

收集反馈之后，需要将反馈用于改进系统。常见方法包括以下几种。

  • 模型微调与奖励学习（RLHF / 策略优化）：利用反映人类偏好的排序或评分数据训练奖励模型，进而通过基于人类反馈的强化学习（RLHF）优化智能体决策，或直接筛选购物助手中用户满意与不满意的交互记录微调 LLM 及独立策略模型。经过训练后，智能体可逐步学会优先选择能带来高评分的动作（例如当历史数据显示遇到模糊提问直接作答普遍得分偏低时，学会先向用户发起澄清提问）。
  • 调整提示词与工具逻辑：大量问题无需重训模型，仅靠优化提示词工程或补齐工具即可解决。例如，若反馈显示用户频繁要求“并排对比两款商品”而现有工具不支持，团队即可新增商品对比工具或补充规格检索步骤；若日志暴露出智能体经常误解某类特定句式，则可在系统提示词中针对性补充说明或 Few-shot 示例。
  • 扩充验证集与回归测试：将每一次人工纠偏或标记的线上错误沉淀为固定的回归测试用例（例如发现购物助手推荐了已停产商品后，将该查询及预期行为加入评估集），在每次升级模型版本或修改提示词后自动重跑测试，配合自动化指标与人工抽检形成“收集反馈 → 针对性修复 → 回归测试 → 部署上线”的持续演进闭环。

这些改进需要结合人的判断与自动评估。自动指标可能发现回答存在事实错误，人工还可能发现表述令人困惑；两方面都可以用于下一次训练或调整。

智能体系统中的容错机制

1. 挑战与失败类型

传统软件的许多失败类型已经得到充分研究，AI 智能体的概率性行为和复杂流程又引入新的类型。多步骤计划中的任何步骤都可能失败，且其方式可能不确定，难以在测试时预测。例如，购物助手的向量查询没有返回有用商品，属于信息检索失败；LLM 编造不存在的 API 调用，属于语义错误；真实 API 调用崩溃或超时，属于执行错误。这些失败还可能互相影响：一次问题可能使智能体进入错误路径，误认为工具已成功，从而让后续步骤依据错误输入和假设运行。外部 API 故障、速率限制等依赖问题也可能随时发生，超出智能体控制范围。因此，设计时需要预期模型推理或外部因素可能出错，并规定可控制的处理方式。

2. 核心容错策略

智能体容错可将经典软件工程的高可用模式与 AI 特有的校验手段相结合：

  • 指数退避重试与熔断（Exponential Backoff & Circuit Breaker）：针对偶发网络抖动、接口限流或工具短暂不可用，可在工具封装层自动捕获异常并按指数递增间隔（如等待 1 秒、2 秒、4 秒）重试，避免密集重试加剧下游负载或再次触发限流；若重试成功，用户仅感知到轻微延迟。当多次重试仍失败时，则转入备选降级路径。此外还可配置熔断器：当某个外部服务持续报错时暂时切断对该服务的调用，直接走降级策略或请求人工介入，防止单点故障拖垮整个智能体。
  • 撤销操作与状态回滚（Rollback & State Recovery）：当多步计划会修改外部业务状态或内部工作状态时，需具备类似数据库事务的回滚能力。例如，购物助手已将商品锁定在购物车并预扣库存，但随后的支付环节失败，此时必须撤销锁定并释放库存。工程上可在每个关键步骤成功后保存智能体记忆状态或外部状态快照（Checkpoint），遇到严重错误时回滚至最近的检查点；或者采用暂存提交模式——先在临时数据结构中记录中间变更，待全部前置步骤确认成功后再一次性提交至数据库或购物车。核心原则是：在确认整条操作链成功前，谨慎执行不可逆的外部副作用。
  • 安全的替代降级机制（Graceful Fallback）：当首选工具不可用，或模型多次重试后仍无法输出合法格式时，平滑切换至更简单的备选方案、返回阶段性部分结果或返回清晰的错误说明。例如：商品向量检索服务故障时临时降级为关键词搜索；LLM 持续无法生成合法的 API JSON 参数时返回预置的模板回复；或者在尝试一种提示词表述无果后切换备选提示词模板；当实时库存接口超时时向用户说明“暂按有货处理，稍后将为您二次确认”。
  • 前置防护与验证（Guardrails & Validation）：在智能体输出或动作真正执行前设置校验关卡。例如要求生成 SQL 查询时，先通过语法解析器或只读 Dry-run 检查合法性；一旦校验失败，立即在进入下一步前触发内部纠错循环或切换工具，尽早拦截看似合理实则错误的语义输出，降低后续回滚的代价。
  • 人工接手兜底（Human Takeover）：当所有自动恢复手段均告失败，或智能体在同一问题上反复报错时，及时将任务转交人工。聊天机器人可将完整上下文移交人工客服跟进（如多次下单失败后由客服人工完成订单），后台自主任务则可暂停后续步骤并向维护人员发送告警。每次人工接手的过程都应记录归档，用于反哺后续的自动化容错改造。

3. 失败的可观测性与监控

容错机制的有效运转高度依赖实时可观测性。系统需通过结构化错误日志、错误率指标与告警准确识别故障位置（例如明确记录究竟是哪个工具调用超时，还是哪条 Pydantic 校验规则未通过），并由外围确定性的流程控制代码（而非单纯依赖 LLM 自身判断）来调度重试或降级分支。

例如，当监控发现购物助手的 retrieve_context工具错误率骤升时，工程师可快速排查是下游库存服务宕机，还是近期提示词变更导致模型传入了多余的非法字段；设置阈值告警（如查询失败率超过 5% 时自动通知值班人员）可避免静默故障扩大。

健康状态仪表板可以展示成功率、每项任务平均尝试次数、转交人工的频率等指标，帮助观察可靠性。例如，失败查询比例突然超过 5% 时，可以自动通知维护人员，减少未被发现的失败。开发期间也必须测试容错能力，例如主动触发工具异常，或提供难以理解的查询，观察系统如何处理。如果智能体无法继续，或者生成错误回答，就说明恢复逻辑存在不足。容错需要预期错误、发现错误，并通过受控的重试、撤销操作或人工介入进行处理。

结合人类反馈与容错

在实际生产系统中，人类反馈与容错机制相辅相成：

  • 反馈帮助改进错误处理：人工反馈经常揭示开发者未预料到的失败类型。例如，审核人员发现某种商品描述会使智能体反复执行错误操作，就可以增加验证规则、专门的替代路径，或调整提示。人工检查能够识别自动检查遗漏的问题。许多自动容错策略也来自对人工处理方式的观察，例如分析人员总在修正文本价格解析结果，就可以增加对应函数，将有效处理方法纳入系统逻辑。
  • 容错有助于获得更有效的反馈：系统妥善处理失败后，用户和标注人员能够评估更丰富的结果。如果智能体遇到问题立即终止，反馈可能仅剩“没有结果”；如果它恢复后提供部分答案或请求帮助，用户可以提供线索或澄清，标注人员也能更具体地指出不足。受控地降低功能能够提高纠正意见和评分的参考价值。
  • 把本次人工反馈与触发它的具体请求、资料版本及处理结果关联起来，供下一轮回归评估使用。
  • 持续改进循环：容错使智能体能够自行处理部分问题并继续服务用户；交互产生显式或隐式反馈，进一步改善能力和可靠性。有些人工纠正可以逐渐自动化，例如用户经常纠正某个事实，就更新检索知识库以减少重复错误。新场景带来新的失败类型时，监控系统可以发现问题并安排人工审核。观察、自动或人工修复、学习和更新构成持续循环：系统处理已有能力覆盖的任务，人类处理系统尚不能完成的部分，并为后续改进提供指导。

人类参与下的可靠智能体

智能体系统具有动态执行和处理复杂任务的能力，也容易遇到新的失败类型。生产环境中的设计需要同时支持接受人类指导和承受执行错误。人工监督与容错相结合，可以让智能体在每次执行中不断改进和适应，并将错误转化为改善系统的机会。

对需要人工介入的失败，保存可复现的输入和状态，并记录处理结论；恢复执行前重新核验当前权限与任务状态。

通过脱敏记录定位失败

在自托管的 Langfuse 中，按 request ID 串联查看资料读取、向量检索、权限校验与回答生成各阶段状态，并将人工反馈精确关联到当时的具体资料版本。遇到 HTTP 429 限流与临时性 5xx 错误时允许执行有限次数的退避重试；遇到权限校验失败时则必须立即终止请求，严禁重试绕过。异常通知队列仅传递错误摘要与排查链接，通知发送端与知识库读取器必须分别独立配置。

练习：记录本课所用配置、输入、预期结果与失败分支，并用脱敏数据复现。

将脱敏告警写入通知发送端

在工程实现中，告警通知发送端与知识源读取器采用各自独立的凭据与配置：飞书自定义机器人使用秒级 timestamp 与请求体内的 JSON sign 字段；钉钉机器人则使用毫秒级 timestamp 与 URL 查询参数签名。Webhook URL 与签名密钥仅保存在后端发送服务中。

examples/notifications.pyPython
import base64
import hashlib
import hmac
import json
import sqlite3
import time
from urllib.parse import parse_qs, urlparse

import httpx


class DeliveryUnknown(RuntimeError):
    pass


class NotificationOutbox:
    def __init__(self, connection):
        self.db = connection
        self.db.execute('CREATE TABLE IF NOT EXISTS notifications (event_id TEXT PRIMARY KEY, state TEXT NOT NULL)')

    def reserve(self, event_id):
        with self.db:
            found = self.db.execute('SELECT state FROM notifications WHERE event_id=?', (event_id,)).fetchone()
            if found:
                if found[0] == 'sent':
                    return False
                raise DeliveryUnknown('notification_requires_reconciliation')
            self.db.execute('INSERT INTO notifications VALUES (?, ?)', (event_id, 'unknown'))
        return True

    def sent(self, event_id):
        with self.db:
            self.db.execute('UPDATE notifications SET state=? WHERE event_id=?', ('sent', event_id))


class LocalQuota:
    def __init__(self, second_limit, minute_limit, clock=time.monotonic):
        self.second_limit, self.minute_limit, self.clock = second_limit, minute_limit, clock
        self.calls = []

    def check(self):
        now = self.clock()
        self.calls = [t for t in self.calls if now - t < 60]
        if len(self.calls) >= self.minute_limit or (self.second_limit and sum(now - t < 1 for t in self.calls) >= self.second_limit):
            raise RuntimeError('notification_local_quota')
        self.calls.append(now)


def feishu_signature(timestamp, secret):
    key = f'{timestamp}\n{secret}'.encode()
    return base64.b64encode(hmac.new(key, b'', hashlib.sha256).digest()).decode()


def dingtalk_signature(timestamp, secret):
    value = f'{timestamp}\n{secret}'.encode()
    return base64.b64encode(hmac.new(secret.encode(), value, hashlib.sha256).digest()).decode()


class WebhookNotifier:
    def __init__(self, client, provider, webhook_url, secret, outbox, clock=time.time):
        parts = urlparse(webhook_url)
        expected = {'feishu': 'open.feishu.cn', 'dingtalk': 'oapi.dingtalk.com'}
        if provider not in expected or parts.scheme != 'https' or parts.hostname != expected[provider]:
            raise ValueError('notification_endpoint_not_allowed')
        if not secret or 'REPLACE_' in secret:
            raise ValueError('notification_secret_required')
        self.query = {k: v[0] for k, v in parse_qs(parts.query).items()}
        if provider == 'dingtalk' and ('access_token' not in self.query or 'REPLACE_' in self.query['access_token']):
            raise ValueError('dingtalk_bot_token_required')
        self.client, self.provider, self.url, self.secret = client, provider, webhook_url, secret
        self.outbox, self.clock = outbox, clock
        self.quota = LocalQuota(5, 100) if provider == 'feishu' else LocalQuota(None, 20)

    def send(self, event_id, summary):
        if not event_id or len(summary) > 1000:
            raise ValueError('notification_summary_invalid')
        if self.provider == 'feishu':
            timestamp = str(int(self.clock()))
            body = {'msg_type': 'text', 'content': {'text': summary}, 'timestamp': timestamp,
                    'sign': feishu_signature(timestamp, self.secret)}
            params = None
        else:
            timestamp = str(int(self.clock() * 1000))
            body = {'msgtype': 'text', 'text': {'content': summary}, 'at': {'isAtAll': False}}
            params = {**self.query, 'timestamp': timestamp, 'sign': dingtalk_signature(timestamp, self.secret)}
        encoded = json.dumps(body, ensure_ascii=False).encode()
        if self.provider == 'feishu' and len(encoded) > 20 * 1024:
            raise ValueError('notification_payload_too_large')
        self.quota.check()
        if not self.outbox.reserve(event_id):
            return {'state': 'deduplicated'}
        try:
            result = self.client.post(self.url, params=params, content=encoded,
                                      headers={'Content-Type': 'application/json'})
            if result.status_code != 200:
                raise DeliveryUnknown(f'notification_http_status={result.status_code}')
            data = result.json()
            code = data.get('code', data.get('StatusCode')) if self.provider == 'feishu' else data.get('errcode')
            if code != 0:
                raise DeliveryUnknown(f'notification_business_code={code}')
        except httpx.HTTPError:
            raise DeliveryUnknown('notification_transport_outcome_unknown') from None
        self.outbox.sent(event_id)
        return {'state': 'sent'}

演示代码中发送的告警仅包含脱敏后的 request ID、失败类别与审核入口链接，严禁夹带知识库正文、密钥或个人敏感信息。平台配额方面，飞书单个自定义机器人限制为 100 次/分钟且 5 次/秒、单次请求体不超过 20KB；钉钉自定义机器人限制为 20 次/分钟。注意，示例中的 LocalQuota 仅约束单进程内调用，多实例部署时需替换为分布式共享限流器。

此外，SQLite NotificationOutbox 按 event_id 持久化记录发送状态：仅在收到明确的成功响应后才更新为 sent；一旦发生网络超时、非 200 HTTP 状态码、业务错误码非零或响应无法解析，状态将保留为 unknown（待核查），下次遇到相同 event_id 时抛出 DeliveryUnknown('notification_requires_reconciliation')。这是因为飞书与钉钉的自定义 Webhook 机器人接口本身不支持服务端幂等键，在网络超时等投递结果未知的情况下，必须由人工或业务对账确认是否已送达后再决定是否重发，防止重复轰炸告警群。

练习：重复提交同一个 event_id，并模拟网络读取超时，检查实际 HTTP 发送次数与 outbox 表中的状态变化；同时验证发出的告警文本中不包含任何知识库原文内容。

搜索课程

正在加载课程目录…