企业官网建设流程全解析

简介：面向AI开发者和技术学习者的Agentic RAG可运行源码包，聚焦代理式检索增强生成的核心原理与实现路径，帮助读者解决传统RAG单一知识源和一次性检索的痛点。压缩包共3个文件，以HTML文件展示解析内容，配合在线运行配置与Git忽略规则文件，整体仅8KB，轻量易用。已有147人学习。通过这份资源可快速了解Agentic RAG的单代理与多代理架构设计，对比其与传统RAG的区别，并掌握函数调用语言模型与代理框架（如DSPy、LangChain）两种实施方法，理解代理如何动态选择与整合信息源。同时，源码与演示文件支持本地或在线环境运行调试，便于直接观察检索与生成链路，掌握Agentic RAG的搭建与调试要点，为企业级智能信息检索场景的落地提供参考。

1. Agentic RAG为什么不是“RAG加了个Agent”

把同一个问答任务交给普通RAG和Agentic RAG，最先感受到的差异往往是：后者变慢了，而且慢得有道理。Agentic RAG不是简单的“RAG加Agent”，而是把一个一次性的检索过程，拆成“判断要不要检索、怎么检索、结果够不够、不够就再检索”的闭环。这篇文章围绕一份可运行的Agentic RAG源码展开，讲清最小实现、核心循环和调试方法。适合正在做知识库问答、文档数量大且问题多变的开发者和算法工程团队。

2. 先理解Agentic RAG的运作机制：从检索循环到工具调用

2.1 普通RAG到Agentic RAG的演进：三个关键差异

普通RAG的流程大家都很熟：用户提问，系统把问题做向量化，到向量库里召回Top-K片段，拼进提示词，一次性让模型生成答案。这个流程在文档少、问题直接的场景下很好用，但一旦问题变成“对比A方案和B方案在成本上的差异，并给出实施建议”，单次检索就开始露怯：它不知道要拆成两个子问题，也不知道第一次召回的结果其实漏掉了关键表格。

Agentic RAG和它的本质区别在于，检索不再是预处理步骤，而是一个可以被模型反复调用的工具。第一个关键差异是反馈回路：普通RAG生成完就结束了，Agentic RAG会拿着检索结果再去问模型“这些够不够”，不够就换关键词、换查询语句、换一个向量库再查一次。第二个差异是工具选择：Agent可以决定用语义检索、用关键词精确匹配、甚至去查一个时间范围过滤后的索引，而不是永远走同一个向量召回流程。第三个差异是多步规划：复杂问题会被拆成多个检索动作，先查背景，再查对比，再查实施案例，每一步的结果都会影响下一步的动作。

这三个差异让Agentic RAG在处理多跳问题时明显强于普通RAG，但也带来了新的代价：延迟变长、token消耗变大、失败模式变多。单纯把检索结果堆给模型已经不够了，你必须对每一步的输入输出有控制权，这就是可运行源码存在的意义。

2.2 核心循环：规划-检索-验证-再检索

整个Agentic RAG的骨架是一个循环，不是一条直线。我用文字描述一遍这个循环，后面章节的源码解析会严格对应它：

1. 用户输入query，系统把它交给Agent（通常就是一次带工具声明的模型调用）。
2. Agent决定动作：直接回答，或者发起一次检索。
3. 如果发起检索，Agent会指定查询语句、检索方式、返回条数。
4. 检索结果作为“观测”追加到上下文中。
5. 系统重新调用Agent，基于历史观测和当前结果再次决策。
6. 如果结果充分或者达到最大轮数，Agent输出最终回答，循环结束。

这个循环里最关键的判断发生在第5步：模型需要区分“当前信息足够回答”和“我还需要更多信息”。很多实现会在这一步单独调用一次轻量模型，只让它输出一个“足够/不够”的判定，避免在主干模型里混入太多检索噪音。

还有一个容易被忽略的设计：历史检索去重。如果Agent在循环里反复检索同一个文档片段，不仅浪费token，还会把模型绕进死循环。成熟的实现一般会维护一个已看过片段ID的集合，检索结果进来先过滤一遍，没有新信息就直接强制进入回答阶段。

2.3 可运行源码的价值：黑匣子与调试成本

我最早接触Agentic RAG时，先试了一个现成的开源方案，发现效果不对。问题在于：你根本不知道是哪一环出了问题。可能是检索结果太差，可能是判定逻辑太严格，可能是提示词没写清楚，也可能是上下文把模型带偏了。没有源码，改参数就像拆盲盒，调了十几次阈值也看不出规律。

这就是为什么“可运行源码”比“思路讲解”值钱。拿到源码后，我能在每一个循环节点打印日志，看到模型选择工具时的原始输出、检索回来的片段ID、判定模块给的分数，甚至手动跳过一个步骤来定位瓶颈。对我来说，可运行的Agentic RAG源码意味着两件事：第一，它把黑匣子变成了白盒，你能一步步跟着数据流走；第二，它给了你一个基线，后续换模型、换向量库、改提示词，都能对比出真实的改进幅度。下面几章，我按一个常规可运行代码包的结构来拆，带你把它跑起来再读进去。

3. 跑通最小可运行版本：项目结构、依赖安装与首个Demo

3.1 拿到源码先看目录结构与入口文件

一份可运行的Agentic RAG源码，无论怎么组织，核心模块就那么几个：Agent循环、检索器封装、工具调用、提示词管理。我习惯把代码包按职责分成四块，先看清楚再动手跑：

agentic_rag/ ├── main.py # 命令行入口，接收query并打印完整trace ├── agent/ │ ├── loop.py # Agent主循环，控制最大轮数和收敛逻辑 │ ├── planner.py # 结果充分性判定，输出“继续检索/直接回答” │ └── tools.py # 工具定义，声明retrieve工具的输入输出格式 ├── rag/ │ ├── retriever.py # 混合检索器，融合向量召回与关键词召回 │ └── embedding.py # 文本向量化封装，对接嵌入模型服务 ├── config.yaml # 模型名、向量库地址、Top-K、轮数上限等参数 └── requirements.txt

入口是main.py，它只做三件事：读取配置、构造Agent、把query交给循环。真正的逻辑在agent/loop.py里，这也是后面调试时打日志最多的文件。

rag/retriever.py被设计成可替换接口，这样你可以先用内置的简单检索器跑通，再换成本地向量库或者外部搜索服务。工具定义单独放在tools.py，是因为Agent模型需要一份严格的JSON Schema来描述retrieve工具，声明不清晰会导致模型频繁输出非法调用。

3.2 环境准备与依赖安装：Python版本、LLM接口与向量库配置

老规矩，先建虚拟环境再装依赖。这份代码要求Python至少3.10，因为用了match语法和类型联合操作符，低版本直接跑不起来。依赖不多，核心就一个关键点：LLM客户端和向量库SDK按你实际用的服务来填，别被锁死。

# 创建虚拟环境并激活 python3.10 -m venv .venv source .venv/bin/activate # 安装基础依赖 pip install pyyaml # 你接的LLM服务对应的SDK，例如某个大模型API的官方客户端 pip install your-llm-sdk # 你用的向量库对应的SDK，例如某个开源向量数据库的客户端 pip install your-vector-db-sdk

装完后打开config.yaml，这里有几个参数会直接影响Demo能不能跑通。llm.model填你实际使用的模型名，llm.api_base填服务地址，retriever.top_k控制每次检索返回多少片段，agent.max_steps控制最大循环轮数。第一次跑建议把max_steps设成3，太小会让Agent没机会展示多轮检索，太大会拖慢调试速度。

需要注意的一处是：很多向量库默认返回的相似度分数不是标准化的，不同索引之间的分数范围差异很大。这份源码会在retriever.py内部对分数做归一化，你不需要在配置里手动调整阈值，但改换向量库时一定要确认检索结果字段名是否和代码里一致。

3.3 跑一次完整对话：命令、日志与返回结果解读

环境准备好后，用一行命令触发首轮对话。建议先拿一个需要多跳检索的问题来测，比如“对比某数据库和某缓存方案在写入性能上的差异”，这样才能看到Agent多次调用检索工具的完整过程。

python main.py --query "对比某数据库和某缓存方案在写入性能上的差异"

跑完后控制台会输出一段trace。我截取关键的几行来解读：

[step 1] agent action: retrieve [step 1] retrieve query: "某数据库写入性能 benchmark" [step 1] top_k: 3 [step 1] retrieved 3 docs, ids: [d-12, d-47, d-88] [step 2] agent action: retrieve [step 2] retrieve query: "某缓存方案写入性能 benchmark" [step 2] top_k: 3 [step 2] retrieved 3 docs, ids: [d-03, d-58, d-99] [step 3] agent action: answer [step 3] final answer generated, total tokens: 2147

注意第一步和第二步检索了不同的query，这就是Agent拆解问题的体现：它没有一次把所有内容查完，而是分别查两个对象。第二步查的是“某缓存方案”，说明模型已经理解用户要对比，并自动把目标拆成了两个子查询。

如果第一次跑结果不理想，先看日志里两步检索的query是否偏离了用户原意。很多翻车不是检索器的问题，而是模型生成的子查询本身就没写好，后续调试也需要优先检查这一步。

4. 核心源码解析：Agent循环、检索器与工具注册表

4.1 Agent循环骨架：从用户输入到最终回答

loop.py是整个系统的核心，它把模型决策、检索执行、历史去重串起来。下面是一个精简但可运行的主循环代码，我加了一些关键路径的注释：

# agent/loop.py import json MAX_STEPS = 6 # 最大循环轮数，防止Agent失速 def run_agent(query: str, retriever, llm, max_steps: int = MAX_STEPS) -> str: # messages 保存对话历史，新增的检索结果以 tool 角色消息追加进来 messages = [ {"role": "system", "content": "你是一个检索助手，必须通过工具获取信息后再回答。"}, {"role": "user", "content": query}, ] seen_ids = set() # 已检索过的文档ID，避免重复消费相同片段 for step in range(max_steps): # 请求模型输出一个动作：retrieve 或 answer response = llm.chat(messages, tools=[RETRIEVE_TOOL]) action = json.loads(response.tool_calls[0].arguments) if action["type"] == "answer": return action["output"] if action["type"] == "retrieve": sub_query = action.get("query", query) top_k = action.get("top_k", 3) docs = retriever.search(sub_query, top_k=top_k) # 过滤掉已经看过的文档 new_docs = [d for d in docs if d["id"] not in seen_ids] if not new_docs: # 没有新信息，强制回答，避免死循环 messages.append({"role": "tool", "content": "检索完成，但没有新信息，请直接回答。"}) continue seen_ids.update(d["id"] for d in new_docs) messages.append({"role": "tool", "content": json.dumps(new_docs, ensure_ascii=False)}) # 达到最大轮数时，让模型基于现有上下文做最后总结 final = llm.chat(messages + [{"role": "user", "content": "请基于已有信息总结回答。"}]) return final.content

这段代码里有几个值得关注的参数。max_steps是整体的安全阀，我建议线上场景设置在8以内，原因后文避坑章节会专门展开。top_k由模型自己决定，但源码里加了边界保护，最小是1，最大是5，防止模型一次请求太多片段导致上下文爆炸。

seen_ids去重是防止反复检索的关键，它直接避免了最常见的“循环不收敛”问题。当你看到一层层检索日志但进展不大时，先检查去重集合是否真的生效了。

4.2 检索器封装：向量召回与关键词检索的切换逻辑

检索器模块的核心价值在于给Agent提供统一的查询接口，而内部可以自由切换策略。常见做法是做一个混合检索，综合向量召回的语义相关性和关键词召回的精确命中，最后做一次加权融合。

# rag/retriever.py class HybridRetriever: def __init__(self, vector_store, keyword_store, alpha=0.5): self.vector_store = vector_store self.keyword_store = keyword_store self.alpha = alpha # 向量检索占比，0~1之间 def search(self, query: str, top_k: int = 3): # 向量库召回数量放大一倍，给融合留出裁剪空间 vec_results = self.vector_store.query(query, top_k=top_k * 2) kw_results = self.keyword_store.query(query, top_k=top_k * 2) # 先做分数归一化，再做加权求和 scores = {} for doc_id, score in vec_results: norm_score = self._normalize(score, vec_results) scores[doc_id] = scores.get(doc_id, 0) + self.alpha * norm_score for doc_id, score in kw_results: norm_score = self._normalize(score, kw_results) scores[doc_id] = scores.get(doc_id, 0) + (1 - self.alpha) * norm_score ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [{"id": doc_id, "score": score} for doc_id, score in ranked[:top_k]]

alpha是一个值得花时间调的参数。在代码库里它的默认值是0.5，但对技术文档类知识库，关键词精确匹配往往比语义向量更可靠，我会把它调到0.3左右；对开放式问题、口语化问题，向量检索更重要，可以调到0.7。判断方法很简单：看日志里最终召回的片段是否包含关键实体词，如果总是漏掉专有名词，就提高关键词权重。

归一化逻辑不能省，因为向量库的余弦相似度和关键词匹配的BM25分数根本不在一个量级。不同向量库返回的分数范围也差异很大，代码里需要在_normalize中按本次查询结果的最大值做线性缩放，保证两种分数权重可比。

4.3 结果评估与“是否继续检索”的判定逻辑

Agent是否继续检索，这部分设计决定了系统的智商上限。一份可运行源码里通常有两种判定方式并存：基于分数的规则判定，和基于轻量模型的自评估。两者各有适用场景，代码里会做一个组合。

# agent/planner.py def is_sufficient(docs, query, llm, score_threshold=0.65): # 策略一：如果召回的片段平均分足够高，认为信息量够 if not docs: return False avg_score = sum(d["score"] for d in docs) / len(docs) if avg_score >= score_threshold: return True # 策略二：分数不够高时，让轻量模型看一遍片段内容再判断 verdict = llm.ask( f"根据以下资料能否回答用户问题，只回答足够或不够。\n" f"问题：{query}\n资料：{json.dumps(docs, ensure_ascii=False)}", max_tokens=8, ) return "足够" in verdict

score_threshold不要拍脑袋设成固定值。不同向量库召回的分数分布差异很大，有些索引的相似度普遍在0.8以上，有些只有0.5，所以源码里常见做法是先跑一批历史query，统计召回分数的中位数，再把阈值设在中位数附近。

策略二依赖轻量模型判断则要注意token消耗。每次循环都让模型读一遍检索结果，成本会翻好几倍。我一般会把策略二的触发条件收紧：只有平均分在0.4到阈值之间时，才会调用模型做二次判定，高于阈值直接过，低于0.4直接认为信息不足、重新检索。这样既保住了判定质量，又控制了成本。

5. Agentic RAG的4个避坑检查点：循环失速、上下文污染与成本失控

5.1 现象：Agent在重复检索同一批文档，迟迟不给出最终回答

这个现象我遇到得太多了。日志里能看到Agent反复生成retrieve动作，而且查询语句越来越像，返回的文档ID也都是见过的。整个循环一直在空转，直到撞上最大轮数才被迫输出一个质量很差的答案。

原因基本有两层：一是去重逻辑没生效，每次检索都直接把结果追加到上下文里，模型看到一堆重复片段以为自己一直在获取新信息；二是判定逻辑太严，score_threshold设得过高，模型永远觉得信息不够。

解决方法是双管齐下。先把去重集合实现出来，确保new_docs为空时强制进入回答阶段；再把score_threshold调低一档，我通常先跑10条历史query，统计正常回答时的平均分，再向下取一个相对宽松的值。

5.2 现象：多轮检索后上下文被无关片段塞满，答案质量下降

这个问题比特慢更隐蔽。表面上Agent只检索了两三轮，但每轮返回3个片段，每个片段几百字，三四轮下来上下文就堆了几千字，其中大量是无用的表格、代码片段和重复语义。模型被无关内容分散注意力，最终答案反而比普通RAG还差。

原因在于检索结果没有做二次过滤就直接进入上下文。向量库召回的Top-K里经常混着语义相近但信息冗余的片段，多轮叠加后冗余成倍增长。

解决思路是做“临时上下文”和“最终上下文”的分离。每轮检索后的片段先放入一个临时观察区，只有被判定为“关键信息”的内容才追加进主对话历史。轻量模型或者一个简单的关键词覆盖率规则都能实现这个筛选。我在项目里常用规则方法：计算片段里包含query核心实体的数量，少于一个实体的片段直接丢弃。

5.3 现象：工具调用格式错误导致Agent循环直接崩溃

第一次把代码接到一个新模型时，经常出现模型输出的工具调用不是合法JSON。日志里会直接抛JSONDecodeError，循环当场中断。这不是个例，而是所有工具调用类应用都会遇到的现实问题。

原因是不同模型的工具调用输出稳定性差异很大，有的模型在复杂指令下会额外输出解释性文字，有的会把JSON内容拆成多段，这些都是代码解析器没能预料的。

解决方法是给工具调用解析加一层容错。常见做法是：先把模型输出里大括号包裹的部分提取出来，再尝试解析；如果解析失败，把错误信息拼到提示词里让模型重新生成一次。我在代码里习惯封装一个parse_tool_call函数，专门做这个清理和重试逻辑，实测可以把工具调用的成功率从75%拉到95%以上。

5.4 现象：检索耗时被放大，整体响应超过10秒

很多团队上了Agentic RAG后被吐槽“太慢了”，单个问题要等十几秒。这个现象的责任不在模型，而在流程设计。每轮循环都要调用一次LLM做决策，再加上一次检索，三轮下来就是六次网络调用，普通RAG一次就结束。

原因是串行调用太多，而且每次循环都重新传整段历史上下文。模型输入侧的token数随轮数线性增长，推理时间也随之增长。

解决思路有三条，可以叠加使用：第一，把多个检索动作合并成并行调用，比如Agent要对比两个对象时，同时发起两路检索；第二，给每轮循环做一个上下文裁剪，只保留与当前子查询相关的片段，不无限追加；第三，在判定逻辑里加入“无检索必要”的快速通道，用户问题如果直接命中知识库里的FAQ，就直接回答，不进入循环。调试时建议在日志里统计每次循环的耗时占比，你会发现瓶颈基本稳定在模型决策调用上，这时压缩轮数通常比优化检索更快。

6. 把Agentic RAG用出自己的版本：参数调优与验证方法

6.1 从“能跑”到“好用”：四个必调参数

代码包能跑通只是第一步，要让它在你的知识库上真正好用，下面四个参数必须按自己的数据调一遍。我第一次接某个项目时就只改了模型名没动参数，结果一测问题百出。

这四个参数是相互影响的，不要单独调一个。top_k调大会让平均分更容易达到阈值，你会错误地认为判定逻辑工作得很好；实际是检索结果变多了，模型判断的信息量变大了。我建议每调整一个参数后，固定跑同一组测试query，记录召回的文档ID和判定结果，再决定下一步改哪个。

6.2 回归测试集与验证：用真实Query衡量Agent行为

调参没有依据就是玄学。我在本地维护了一个小型回归测试集，大概30条query，每条标注了期望行为：应该检索几次、应该使用哪个子查询、最终答案必须包含哪些关键实体。每次改完代码或参数，跑一遍这个测试集，用脚本对比行为差异。

# eval/run_regression.py import yaml from agent.loop import run_agent cases = [ {"query": "某数据库写入性能如何", "need_retrieve": True, "expect_entity": ["某数据库"]}, {"query": "直接回复一句话，说明你没有理解用户意图", "need_retrieve": False}, ] config = yaml.safe_load(open("config.yaml")) agent = build_agent(config) miss_count = 0 for case in cases: trace, answer = run_agent(case["query"], agent) if case["need_retrieve"] and len(trace.retrieve_steps) == 0: miss_count += 1 if not any(e in answer for e in case.get("expect_entity", [])): miss_count += 1 print(f"miss count: {miss_count}/{len(cases)}")

这个验证脚本的价值不是追指标，而是防止参数调整带崩原有能力。我吃过一次亏：为了降低延迟，把max_steps从6改到3，结果某个多跳问题的检索链路被打断，答案质量明显下降。如果没有回归测试，这个问题要等到上线用户反馈才会暴露。我的习惯是每次参数调整必须让回归集miss数不增加，才允许进入下一步。

Agentic RAG的源码解析到最后，最值钱的还是你自己手里的那套测试集和调参记录。遇到问题不要急着换更强的模型，先把循环日志翻一遍，找到是决策、检索、还是判定环节在拖后腿。希望这篇文章的拆解思路能帮到你。

本文还有配套的精品资源，点击获取