Agent的学习记录--7

来源Github的学习指南Agent Learning Hub的学习：

Stage 2——鲁棒性与防护

在这里，有四种面对不可靠外部环境时的四大核心防御机制

工具执行失败 (Tool Failure)
空结果 (Empty Results)
重复调用与死循环检测 (Loop & Cycle Detector)
幻觉引用检测 (Hallucinated Citations Guard)

这些的目的都是为了防止Agent 在复杂真实生产场景中崩溃、卡死、狂刷 Token 或信口开河

import re
import sys
import json
import sqlite3
from typing import List, Dict, Any, Tuple, Optional

from config import get_llm
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AIMessage

工具执行失败 (Tool Failure) 捕获与自愈反哺和空结果 (Empty Results) 智能降级与防幻觉引导

当工具执行失败的时候，我们可以想象，其肯定是返回不了结果，所以我们可以 try-except去捕获，对于空结果，我们只需要捕获tool_func返回的结果进行判断：

def safe_execute_tool(tool_func, tool_args: Dict[str, Any]) -> str:
    try:
        result = tool_func.invoke(tool_args)
        
        is_empty = False
        if result is None:
            is_empty = True
        elif isinstance(result, str) and not result.strip():
            is_empty = True
        elif isinstance(result, (list, dict)) and len(result) == 0:
            is_empty = True
            
        if is_empty:
            return "[系统提示]: 工具已成功执行，但未匹配到任何结果 (结果集为空)。请向用户如实说明未查到，严禁随意编造虚假信息！"
        
        return str(result)
        
    except Exception as e:
        return f"[工具调用失败诊断]: 执行工具 '{tool_func.name}' 发生异常: {type(e).__name__} - {str(e)}。"

在编写的过程中也遇到一些问题：

LangChain 中 LLM 与 Tool 的返回值差异;
- 在LangChain里，.invoke() 返回的是带有 .content的大模型消息对象；而tool_func返回的是python的原生类型(str / dict / list / None)，所以一旦写了 result.content，模型会直接报错，以为是工具调用失败，这也是异常吞没错误了
空值与弱类型：工具的返回值具有多态性（可能返回 None、""、" "、[]、{} 等），我一开始的时候直接写了 result.strip(),然后直接判断是否为False,这样也会直接崩掉，应该使用isinstance()函数

重复调用与死循环检测器

我们担心工具被重复调用，所以我们需要检测调用次数，防止陷入循环。所以我们对每次模型调用工具给予一个签名，同时记录参数，避免每次都调用同样的参数进入同一个工具。

class LoopCircuitBreaker:
    def __init__(self, max_consecutive_duplicates: int = 2):
        self.max_consecutive_duplicates = max_consecutive_duplicates
        self.call_history: List[str] = []
        
    def check_and_record(self, tool_name: str, tool_args: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
        signature = f"{tool_name}:{json.dumps(tool_args, sort_keys=True, ensure_ascii=False)}"
        time = 0
        for prev in reversed(self.call_history): #只看后面5条消息
            if prev == signature:
                time = time + 1
        if time >= (self.max_consecutive_duplicates - 1) and len(self.call_history) > 0:
            return tuple([True, 
                    f"[死循环熔断]: 检测到连续 {self.max_consecutive_duplicates} 次使用相同参数调用工具 '{tool_name}'！系统已阻断执行。请立即换用其他工具，或直接向用户总结目前已得到的信息。"
                    ])
        self.call_history.append(signature)
        return ([False, None])

这里踩的坑：首先是检测次数的逻辑问题，因为我们的例子里调用工具次数为最多2次，我一开始的判断条件写的是 time >= self.max_consecutive_duplicates这样的话在调用第2次的时候仍然会通过，因为第一次调用工具的时候，次数并不会 +1，只有第二次调用的时候次数 + 1，但这时候我们已经发现不对了，因为它用重复的参数调用2次了，此时就触发了熔断。

还有一个问题是 tuple()的使用，不可以直接写 tuple(False, None)，因为tuple()只有一个参数，这样写等于传了两个参数，会返回类型错误。然后因为我们输出的是 Tuple[bool, Optional[str]]，所以我们可以直接写 return (False, None)或者不加括号或是上面的方式都是可以的

幻觉引用校验器

这里的一个思路是我们先把所有的标签都提取出来，比如[Doc 5]类似的格式，并且看其是否捏造了标签

class CitationGroundingGuard:
    @staticmethod
    def verify_citations(
        generated_answer: str,
        retrieved_docs: List[Dict[str, Any]]
    ) -> Dict[str, Any]:
        total_valid_docs = len(retrieved_docs) # Total docs
         citation_mark = re.findall(r"\[(?:Doc|文档)\s*(\d+)\]", generated_answer, flags=re.IGNORECASE) # flags参数忽略大小写
        cited_ids = set(int(m) for m in citation_mark)
        hallucinated_doc_ids = []
        warnings = []
        for cid in sorted(cited_ids):
            if cid < 1 or cid > total_valid_docs:
                hallucinated_doc_ids.append(cid)
        if not hallucinated_doc_ids:
            is_valid = True
        else:
            is_valid = False
            warnings.append(f"检测到虚假引用 Doc 编号: {hallucinated_doc_ids}，检索库中文档上限仅为 {total_valid_docs}！")
        return {"is_valid": is_valid, 
                "cited_ids": sorted(list(cited_ids)), 
                "hallucinated_doc_ids": hallucinated_doc_ids, 
                "warnings": warnings}

在这里，我们实现了第一步检查模型回答中的 [Doc X] 标签是否存在捏造，但没有实现引用句是否在原文片段中有是实质性的关键词支撑：

这里的做法有很多种：

词重叠 / 关键词 Jaccard 相似度：
1. 用标点符号将回答拆分成单句；
2. 提取带有 [Doc X] 的单句，并去除停用词后提取名词/关键词；
3. 计算该句与 Doc X 原文的关键词交集率（Jaccard 相似度）或 ROUGE-1 分数；
4. 若重合比例低于阈值（如 < 20%），判定为无实质依据。
NLI（自然语言推理）小模型：
1. 使用轻量级 Cross-Encoder 模型（如 DeBERTa-v3-small 或 BGE-Reranker）；
2. 前提（Premise）：Doc X 的原文片段；
3. 假设（Hypothesis）：模型生成的带有引用的那句话；
4. 如果模型判断为 Entailment（蕴含），说明有充分依据；若为 Contradiction（矛盾）或 Neutral（无关），则警告。
LLM-as-a-Judge：
异步调用一个小参数量的大模型（如 Qwen2.5-7B），Prompt 传入原文与生成句，让它只回答 YES/NO。

我自己的思考就是在解决幻觉问题的时候，是不是能够使用之前提到的supervisor-worker的工作流。这种具有自我反思的功能，通过Multi-agent实现。

其闭环状态机如下：

[用户提问]
    │
    ▼
[Worker Agent] ──▶ 检索文档并生成带 [Doc X] 引用的草稿
    │
    ▼
[Supervisor / Critic Agent] ──▶ 严格对比【草稿句子】与【原文内容】
    │
    ├─►【通过 (PASS)】──▶ 输出最终答案给用户
    │
    └─►【发现幻觉 (REVISE)】
            │
            └── (附带具体修改意见) ──▶ 返回给 [Worker Agent] 重新生成

因为这种的工作流有一个明显好处就是能看懂“同义改写”与“语境”，并且能将错误的信息返还给Agent进行修改，作为AIMessage进行下一轮的思考。当然问题也很明显，一方面是思考链条会延长，token消耗增加，且可能出现死循环的情况，因为如果 Worker 总是改不对，Supervisor 就会一直打回，导致两个 Agent 无限来回扯皮。所以我想的话就是使用代码进行标签捏造的检查，然后通过这一关后使用Multi-agent进行语义核查避免捏造。
                    [Worker 生成草稿]
                           │
                           ▼
  【第一道防线：代码层 (比如上面的CitationGroundingGuard)】
       └─ 检查文档编号是否越界？格式对不对？（耗时 < 1ms，0 成本）
       ├─ [失败] ──▶ 直接本地代码快速拦截，不浪费任何 Token
	   └─ ──────────────────
      					   ▼
       					 [通过]
                           │
                           ▼
  【第二道防线：认知层 (Supervisor Agent)】
       └─ 深入核对复杂的语句逻辑、事实支撑（耗时 1~2s）
       ├─ [通过] ──▶ 交付用户
       └─ [有幻觉] ──▶ 反哺修正

Agent学习记录-7

Agent的学习记录--7

Stage 2——鲁棒性与防护

工具执行失败 (Tool Failure) 捕获与自愈反哺和空结果 (Empty Results) 智能降级与防幻觉引导

重复调用与死循环检测器

幻觉引用校验器