Agentic RAG 工程化实战：把知识检索变成第一类“工具”的关键技术点

发表于 分类于 ，
本文字数： 4.7k 阅读时长 ≈ 4 分钟
以企业级问答与操作协同为例，讲解如何把“检索”设计成 AI Agent 的第一类工具，覆盖多路检索、反幻觉、预算控制与可观测性，附 Python 代码骨架与调试要点。

Agentic RAG 工程化实战：把知识检索变成第一类“工具”的关键技术点

技术主题：AI Agent（人工智能代理）
内容方向：关键技术点讲解（RAG 工程化、工具调用与反幻觉）

引言

很多落地项目里，RAG 常被当作“在回答前顺手查一下文档”。这样的松散集成会带来三个问题：检索时机不可控、来源不可靠、成本时延不可预期。本文分享我在企业知识库问答 + 工具执行的实践：把“检索”升级为第一类工具（Tool-First），让 Agent 以计划-执行的方式主动调用检索，并对引用、成本与质量负责。文末提供可复用的 Python 骨架与调试清单。

一、场景与挑战

  • 场景：员工问“PR-4827 什么时候发版？按规范需要走哪些审批？顺便创建一个发布会议”。Agent 既要回答规则，又要操作日历创建会议。
  • 挑战：
    • 文档来源多（规范、变更记录、Wiki），格式杂（PDF/网页/表格）；
    • 关键词难以直接命中（语义表达差异、术语混用）；
    • 需要“带证据”的回答（可点击引用），并控制时延与费用。

二、总体设计（Tool-First 的 Agentic RAG）

  • 意图分流：先判断“是否需要检索/需要哪些知识域/是否要调用外部工具”；
  • 多路检索：BM25（关键词）、向量召回（语义）、结构化查询（表格/数据库）；
  • 重排与去重：基于交叉编码器/简单打分的相关性重排，去除相似片段；
  • 证据约束生成：回答必须引用 K 个证据块，不满足时回退重新检索；
  • 预算控制：在每轮计划里设定 token/时间预算，必要时降级（缩短上下文、降采样）；
  • 观测与评估：记录检索-生成链路与引用质量，构建离线评测集持续回归。

三、关键技术点与代码骨架（Python）

下面的代码骨架聚焦可读性与“工程化骨架”，可替换为任意向量库/LLM 实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
from dataclasses import dataclass
from typing import List, Dict, Any, Tuple
import time

# 1) 工具接口：把检索当工具
@dataclass
class DocChunk:
    id: str
    text: str
    source: str
    score: float

class Retriever:
    def __init__(self, bm25, vec_index):
        self.bm25 = bm25
        self.vec_index = vec_index

    def bm25_search(self, q: str, k: int = 10) -> List[DocChunk]:
        # 伪实现：返回包含 score 的片段
        return self.bm25.search(q, k)

    def vec_search(self, q: str, k: int = 10) -> List[DocChunk]:
        return self.vec_index.search(q, k)

    def hybrid(self, q: str, kb: str, k: int = 8) -> List[DocChunk]:
        # 多路召回并合并去重
        a = self.bm25_search(q + " " + kb, k)
        b = self.vec_search(q, k)
        merged = dedup(a + b, key=lambda c: c.id)
        return rerank(merged)[:k]

# 2) 计划-执行：是否需要检索？需要哪些域？
@dataclass
class Plan:
    need_search: bool
    kb: List[str]
    need_tool: List[str]
    budget_tokens: int
    budget_ms: int

class Planner:
    def decide(self, user_query: str) -> Plan:
        # 可替换为 LLM 分类器；这里启发式
        need_search = True if any(k in user_query for k in ["规范", "审批", "时间", "发版"]) else False
        kb = ["release", "policy"] if need_search else []
        need_tool = ["calendar.create"] if "会议" in user_query else []
        return Plan(need_search, kb, need_tool, budget_tokens=4096, budget_ms=3000)

# 3) 证据约束生成：没有足量证据就回退
class Generator:
    def __init__(self, llm):
        self.llm = llm

    def answer(self, query: str, evidences: List[DocChunk], k_refs: int = 2) -> Tuple[str, List[DocChunk]]:
        ctx = "\n\n".join([f"[{i+1}] ({e.source}) {e.text}" for i, e in enumerate(evidences[:6])])
        prompt = f"请基于下述证据回答，并在结尾以[数字]方式给出引用：\n{ctx}\n\n问题：{query}\n回答："
        text = self.llm.generate(prompt, max_tokens=800)
        used = extract_refs(text)  # 解析 [1][2] 之类的引用
        used_evi = [evidences[i-1] for i in used if 1 <= i <= len(evidences)]
        # 若引用数量不足，提示回退
        if len(used_evi) < k_refs:
            raise RuntimeError("insufficient evidence")
        return text, used_evi

# 4) 预算控制与降级策略
class Budget:
    def __init__(self, tokens: int, ms: int):
        self.tokens = tokens
        self.ms = ms
        self.start = time.time()
    def left_ms(self) -> int:
        return int(self.ms - (time.time() - self.start) * 1000)

# 5) 端到端编排
class AgenticRAG:
    def __init__(self, retriever: Retriever, planner: Planner, gen: Generator):
        self.retriever = retriever
        self.planner = planner
        self.gen = gen

    def run(self, query: str) -> Dict[str, Any]:
        plan = self.planner.decide(query)
        budget = Budget(plan.budget_tokens, plan.budget_ms)
        evidences: List[DocChunk] = []
        if plan.need_search:
            for domain in plan.kb:
                if budget.left_ms() < 400: break
                evidences += self.retriever.hybrid(query, domain, k=6)
            evidences = dedup(evidences, key=lambda c: c.id)
        # 生成 + 引用校验 + 失败回退（缩小范围再检索）
        try:
            text, used = self.gen.answer(query, evidences, k_refs=2)
        except Exception:
            # 降级：仅用 top-4 证据再试
            text, used = self.gen.answer(query, evidences[:4], k_refs=1)
        return {"answer": text, "citations": [u.source for u in used]}

# --- 工具函数占位 ---
def dedup(items, key):
    seen = set(); out = []
    for x in items:
        k = key(x)
        if k in seen: continue
        seen.add(k); out.append(x)
    return out

def rerank(chunks: List[DocChunk]) -> List[DocChunk]:
    # 可替换成 cross-encoder；这里按 score 排序
    return sorted(chunks, key=lambda c: c.score, reverse=True)

def extract_refs(text: str) -> List[int]:
    import re
    return [int(x) for x in re.findall(r"\[(\d+)\]", text)]

要点：

  • 检索是“第一类工具”，先被 Planner 计划，再执行；能否回答由“证据充足度”约束；
  • 混合检索 → 重排 → 引用生成 → 引用校验 → 不足则回退；
  • 预算控制要落到“每轮剩余时间”上，靠降级策略确保可控时延。

四、调试与评估方法

  1. 观测维度
  • 检索：召回@K、重排后 NDCG、重复率；
  • 生成：引用数量、引用命中率（是否真在证据中）、答案一致性；
  • 成本：每轮 token、检索次数、外部调用耗时；
  • 体验：P95 时延、失败回退率。
  1. 回归评测集
  • 构建 100-300 条业务问题的标注集，每条包含“应命中文档 ID 与片段范围”；
  • 持续评估“是否命中 + 引用是否正确 + 文本一致性分”。
  1. 常见坑
  • 只用向量检索：术语差异与 OOD 问题会导致“全军覆没”，需 BM25 兜底；
  • 大段粘贴：Chunk 太大导致无效 Token；建议 400-800 字符切分并保留来源；
  • 无引用约束：容易“幻觉复述”而不引用证据，需硬性校验并回退。

五、案例：问“某 PR 的发版规则并创建会议”

  • Planner 输出：need_search=True, kb=[release, policy], need_tool=[calendar.create]
  • AgenticRAG 先检索并回答，提取“审批人角色、发布时间窗”；
  • 若引用不足，缩小到 release 域重试；
  • 再调用 calendar.create（此处省略），将摘要附在日历描述中。

总结

把 RAG 变成 Agent 的第一类工具，本质是“让检索对结果负责”：由 Planner 明确时机与域，混合检索聚合证据，生成阶段强制引用并进行校验，超时与成本由预算控制。这样做带来可预期的时延、可验证的来源与可复用的工程骨架。你可以把上述骨架替换成任意向量库与 LLM 实现，并在评测集上持续回归，把团队的知识问答能力稳定在生产水位。