前面四章已经让 Agent 具备了工具调用、多步规划、错误恢复和记忆能力。但它还有一个重要限制模型只能依靠训练数据和当前上下文回答问题。如果用户问的是公司内部文档、产品手册、私有知识库或者最近刚发生的事件模型就可能答不上来甚至会“一本正经地胡说”。这一章要解决的就是这个问题给 Agent 接上外部知识库让它在回答前先检索相关文档再基于检索结果生成答案。这种能力通常被称为RAG检索增强生成。一、什么是检索增强检索增强的核心思想很简单模型不一定要“记住”所有知识但它应该知道“去哪里查”。传统问答有两种极端方式方式优点缺点完全依赖模型实现简单容易过时、无法访问私有知识完全依赖搜索信息准确结果可能分散、难以直接回答RAG 把两者结合起来用户提问系统从知识库中检索相关文档把检索结果作为上下文交给模型模型基于上下文生成答案。用一句话概括检索负责“找资料”模型负责“读懂并回答”。二、知识库系统的基本流程一个典型的检索增强系统包含五个环节1用户提问 2 ↓ 3查询改写 / 向量化 4 ↓ 5从知识库检索相关文档 6 ↓ 7把文档拼入提示词 8 ↓ 9模型生成答案更具体一点文档入库把 PDF、Markdown、网页、数据库记录等文本切分成小块。向量化每个文本块通过嵌入模型转换成向量。存储向量存入向量数据库例如 FAISS、Chroma、Milvus。检索用户提问时先把问题也转成向量再查找最相似的文本块。生成把最相关的文档片段放入提示词让模型基于这些片段回答。三、为什么需要文本分块大模型和向量数据库都不适合一次性处理超长文档。如果直接把整本手册塞进去会出现几个问题超出模型上下文窗口检索时难以定位真正相关的段落每次调用都要处理大量无关内容成本高、速度慢。所以通常会把文档切成较小的片段例如每段 300-800 字。这样检索时只返回最相关的几段模型也能更高效地理解。分块时要注意不要从句子中间硬切保留一定重叠避免关键信息被切断每个块最好附带来源信息方便追溯。四、为什么需要向量检索传统关键词检索只能匹配字面相同的词。比如用户问“怎么重置密码”文档里写的是“如何找回登录密码”关键词匹配可能找不到。向量检索把文本转成语义向量即使字面不完全一致只要意思接近也能被检索出来。简单理解关键词检索看“词是否一样”向量检索看“意思是否接近”。不过向量检索也不是万能的。它适合找语义相近的片段但对精确编号、具体条款、严格术语匹配可能不如关键词检索。因此生产系统里常把两者结合使用。五、本章项目结构本章在第四章基础上增加知识库能力项目结构如下1project/ 2├── .env 3├── agent_with_rag.py 4├── verify_rag.py 5├── knowledge_base/ 6│ ├── docker_kubernetes_guide.md 7│ └── product_faq.md 8└── agent_memory/ 9 ├── chat_history.json 10 └── user_profile.json其中knowledge_base/存放示例知识库文档agent_with_rag.py带检索增强能力的 Agentverify_rag.py验证检索与回答效果。为了贴近实际场景这里准备了两类示例文档一份关于容器编排技术的基础指南另一份是产品常见问题解答。六、完整可运行代码agent_with_rag.py 第五章示例检索增强与知识库单文件整合版 运行方式 1. 在项目根目录创建 .env 文件写入 OPENAI_API_KEY 2. 安装依赖 pip install langchain langchain-openai faiss-cpu python-dotenv 3. 准备知识库文档 - knowledge_base/docker_kubernetes_guide.md - knowledge_base/product_faq.md 4. 运行 python agent_with_rag.py import os import json import glob from datetime import datetime from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_core.tools import tool from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS # ---------------------------- # 1. 加载环境变量 # ---------------------------- load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY, ) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o) MODEL_BASE_URL os.getenv(MODEL_BASE_URL, ).strip() EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) if not OPENAI_API_KEY: raise RuntimeError( 未找到 OPENAI_API_KEY请在项目根目录创建 .env 文件并写入 API Key。 ) # ---------------------------- # 2. 初始化模型 # ---------------------------- llm_kwargs { model: MODEL_NAME, temperature: 0.2, api_key: OPENAI_API_KEY, } if MODEL_BASE_URL: llm_kwargs[base_url] MODEL_BASE_URL llm ChatOpenAI(**llm_kwargs) # 初始化嵌入模型用于把文本转成向量 embeddings OpenAIEmbeddings( modelEMBEDDING_MODEL, api_keyOPENAI_API_KEY, **( {base_url: MODEL_BASE_URL} if MODEL_BASE_URL else {} ), ) # ---------------------------- # 3. 定义工具 # ---------------------------- tool def get_weather(city: str) - str: 查询指定城市的天气信息。 当用户询问某个城市的天气时调用此工具。 参数 - city: 城市名称例如“郑州”“北京”“上海” if not city or not city.strip(): return 缺少城市名称无法查询天气。 city city.strip() weather_data { 郑州: 晴22°C微风, 北京: 多云18°C北风2级, 上海: 小雨20°C东南风3级, 广州: 阴26°C湿度较高, 深圳: 晴28°C适合出行, } if city in weather_data: return f{city}天气{weather_data[city]} return f未找到“{city}”的天气数据请确认城市名称。 tool def get_current_time(dummy: str ) - str: 获取当前日期和时间。 当用户询问时间、日期、现在几点时调用此工具。 参数 - dummy: 可选占位参数不需要传入实际内容 now datetime.now() return now.strftime(%Y年%m月%d日 %H:%M:%S) tool def convert_currency(amount: float, from_currency: str, to_currency: str) - str: 进行简单的货币换算。 当用户询问汇率或进行金额换算时调用此工具。 参数 - amount: 需要换算的金额 - from_currency: 源货币例如 USD、CNY - to_currency: 目标货币例如 CNY、USD if amount 0: return 换算金额必须大于 0。 from_currency from_currency.upper() to_currency to_currency.upper() rates { USD_CNY: 7.2, CNY_USD: 1 / 7.2, EUR_CNY: 7.8, CNY_EUR: 1 / 7.8, } key f{from_currency}_{to_currency} if key not in rates: return ( f暂不支持 {from_currency} 到 {to_currency} 的换算 请尝试 USD/CNY 或 EUR/CNY。 ) result amount * rates[key] result round(result, 2) return f{amount} {from_currency} ≈ {result} {to_currency} tools [get_weather, get_current_time, convert_currency] # ---------------------------- # 4. 记忆管理器 # ---------------------------- MEMORY_DIR agent_memory CHAT_HISTORY_FILE os.path.join(MEMORY_DIR, chat_history.json) USER_PROFILE_FILE os.path.join(MEMORY_DIR, user_profile.json) def ensure_memory_dir(): 确保记忆存储目录存在。 os.makedirs(MEMORY_DIR, exist_okTrue) def load_chat_history(session_id: str) - list: 从文件加载指定会话的聊天历史。 ensure_memory_dir() if not os.path.exists(CHAT_HISTORY_FILE): return [] try: with open(CHAT_HISTORY_FILE, r, encodingutf-8) as f: data json.load(f) except Exception: return [] if not isinstance(data, dict): return [] return data.get(session_id, []) def save_chat_history(session_id: str, history: list): 将指定会话的聊天历史写入文件。 ensure_memory_dir() all_histories {} if os.path.exists(CHAT_HISTORY_FILE): try: with open(CHAT_HISTORY_FILE, r, encodingutf-8) as f: all_histories json.load(f) if not isinstance(all_histories, dict): all_histories {} except Exception: all_histories {} all_histories[session_id] history with open(CHAT_HISTORY_FILE, w, encodingutf-8) as f: json.dump(all_histories, f, ensure_asciiFalse, indent2) def load_user_profile() - dict: 加载用户长期偏好。 ensure_memory_dir() if not os.path.exists(USER_PROFILE_FILE): return {} try: with open(USER_PROFILE_FILE, r, encodingutf-8) as f: profile json.load(f) return profile if isinstance(profile, dict) else {} except Exception: return {} def save_user_profile(profile: dict): 保存用户长期偏好。 ensure_memory_dir() with open(USER_PROFILE_FILE, w, encodingutf-8) as f: json.dump(profile, f, ensure_asciiFalse, indent2) def update_user_profile_from_conversation(profile: dict, last_human: str, last_ai: str) - dict: 从最近一轮对话中抽取用户偏好更新长期记忆。 profile profile.copy() city_keywords [郑州, 北京, 上海, 广州, 深圳, 杭州, 成都] for city in city_keywords: if city in last_human and city not in profile: profile[city] city name_prefixes [我叫, 名字是, 叫我] for prefix in name_prefixes: if prefix in last_human: idx last_human.find(prefix) len(prefix) name last_human[idx:].strip().split()[0].split(,)[0].strip() if name and name not in profile: profile[name] name return profile def trim_history(messages: list, max_rounds: int 10) - list: 对聊天历史做滑动窗口裁剪。 if len(messages) max_rounds * 2: return messages return messages[-max_rounds * 2:] def build_history_messages(history: list) - list: 将 JSON 格式的历史转成 LangChain 消息对象。 messages [] for item in history: role item.get(role) content item.get(content, ) if role human: messages.append(HumanMessage(contentcontent)) elif role ai: messages.append(AIMessage(contentcontent)) return messages def serialize_history(messages: list) - list: 将 LangChain 消息对象转成可保存的 JSON 列表。 result [] for msg in messages: if isinstance(msg, HumanMessage): result.append({role: human, content: msg.content}) elif isinstance(msg, AIMessage): result.append({role: ai, content: msg.content}) return result # ---------------------------- # 5. 知识库管理器 # ---------------------------- KNOWLEDGE_BASE_DIR knowledge_base VECTOR_STORE_DIR vector_store VECTOR_STORE_PATH os.path.join(VECTOR_STORE_DIR, faiss_index) def ensure_kb_dir(): 确保知识库目录存在。 os.makedirs(KNOWLEDGE_BASE_DIR, exist_okTrue) def ensure_vector_store_dir(): 确保向量存储目录存在。 os.makedirs(VECTOR_STORE_DIR, exist_okTrue) def load_kb_documents() - list: 从 knowledge_base 目录加载所有 Markdown 文档。 返回格式 [ {content: ..., source: docker_kubernetes_guide.md}, {content: ..., source: product_faq.md}, ] ensure_kb_dir() md_files glob.glob(os.path.join(KNOWLEDGE_BASE_DIR, *.md)) documents [] for file_path in md_files: try: with open(file_path, r, encodingutf-8) as f: content f.read().strip() if content: source os.path.basename(file_path) documents.append({content: content, source: source}) except Exception as e: print(f读取文件失败{file_path}错误{e}) return documents def split_documents(documents: list, chunk_size: int 500, chunk_overlap: int 50) - list: 将文档切分为较小的文本块。 参数 - documents: 文档列表每项包含 content 和 source - chunk_size: 每个文本块的最大字符数 - chunk_overlap: 相邻文本块之间的重叠字符数 返回格式 [ {content: ..., source: docker_kubernetes_guide.md}, ... ] splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ], ) chunks [] for doc in documents: content doc[content] source doc[source] # 使用 LangChain 文本分割器切分 split_texts splitter.split_text(content) for text in split_texts: if text.strip(): chunks.append({ content: text.strip(), source: source, }) return chunks def build_vector_store(documents: list) - FAISS: 从文档列表构建 FAISS 向量索引。 如果本地已有缓存索引则直接加载否则重新构建。 ensure_vector_store_dir() if documents: # 准备 FAISS 需要的文本和元数据 texts [doc[content] for doc in documents] metadatas [{source: doc[source]} for doc in documents] # 从文本和元数据创建向量索引 vector_store FAISS.from_texts( textstexts, embeddingembeddings, metadatasmetadatas, ) # 保存索引到本地避免每次启动都重新构建 vector_store.save_local(VECTOR_STORE_PATH) return vector_store # 如果没有文档尝试加载已有索引 if os.path.exists(VECTOR_STORE_PATH): return FAISS.load_local( VECTOR_STORE_PATH, embeddings, allow_dangerous_deserializationTrue, ) # 实在没有数据时返回空索引 return FAISS.from_texts( texts[暂无知识库文档], embeddingembeddings, metadatas[{source: empty}], ) def get_or_build_vector_store() - FAISS: 获取或构建向量存储。 每次调用都会重新加载知识库文档并重建索引。 生产环境中可以改为增量更新或定时刷新。 documents load_kb_documents() chunks split_documents(documents) if not chunks: # 如果知识库为空仍返回一个可用向量存储 return build_vector_store([]) return build_vector_store(chunks) def retrieve_context(query: str, top_k: int 3) - str: 根据用户问题检索相关文档片段。 参数 - query: 用户问题 - top_k: 返回最相关的 top_k 个文档片段 返回格式 一段拼接好的参考文本包含内容和来源。 vector_store get_or_build_vector_store() # similarity_search 会返回最相似的文档片段 docs vector_store.similarity_search(query, ktop_k) if not docs: return 未在知识库中找到相关信息。 context_parts [] for i, doc in enumerate(docs, start1): source doc.metadata.get(source, 未知来源) content doc.page_content.strip() context_parts.append( f[片段{i}] 来源{source}\n{content} ) return \n\n.join(context_parts) # ---------------------------- # 6. 检索增强工具 # ---------------------------- tool def search_knowledge_base(query: str) - str: 在本地知识库中检索与问题相关的文档片段。 当用户询问产品说明、技术文档、内部规范、教程内容时调用此工具。 参数 - query: 用户的问题或关键词 if not query or not query.strip(): return 缺少查询内容无法检索知识库。 query query.strip() context retrieve_context(query, top_k3) return context # 把检索工具加入工具列表 tools.append(search_knowledge_base) # ---------------------------- # 7. 创建带检索增强的 Agent # ---------------------------- prompt ChatPromptTemplate.from_messages( [ ( system, 你是一个智能助手。你可以调用工具查询天气、时间、汇率也可以检索本地知识库。 回答时请遵循以下规则 1. 如果知识库中有相关信息优先基于知识库内容回答并注明来源。 2. 如果知识库信息不足可以结合常识回答但要明确说明哪些来自知识库哪些是补充说明。 3. 不要编造知识库中没有的具体条款、版本号或内部规定。 4. 如果用户之前提到过自己的城市、姓名或偏好请在回答时自然引用。 5. 如果检索结果为空请如实告知用户并尝试引导用户提供更多信息。 ), MessagesPlaceholder(variable_namechat_history), (human, {input}), (placeholder, {agent_scratchpad}), ] ) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, max_iterations8, handle_parsing_errorsTrue, ) # ---------------------------- # 8. 带检索增强的调用入口 # ---------------------------- def ask_agent(session_id: str, user_input: str) - str: 执行一次带记忆和检索增强的 Agent 调用。 流程 1. 读取历史对话 2. 加载用户长期偏好 3. 裁剪历史 4. 调用 Agent 5. 保存本轮对话 6. 更新用户长期偏好。 raw_history load_chat_history(session_id) history_messages build_history_messages(raw_history) history_messages trim_history(history_messages, max_rounds8) profile load_user_profile() enriched_input user_input if profile: profile_hint 用户已知信息 if profile.get(name): profile_hint f 姓名{profile[name]} if profile.get(city): profile_hint f 所在城市{profile[city]} if profile.get(preferences): profile_hint f 偏好{profile[preferences]} enriched_input profile_hint \n用户当前问题 user_input result agent_executor.invoke( { input: enriched_input, chat_history: history_messages, } ) answer result.get(output, 抱歉助手未能生成回答。) # 保存本轮对话到短期记忆 new_history raw_history [ {role: human, content: user_input}, {role: ai, content: answer}, ] save_chat_history(session_id, new_history) # 更新长期记忆从最近一轮对话中抽取偏好 profile update_user_profile_from_conversation(profile, user_input, answer) save_user_profile(profile) return answer # ---------------------------- # 9. 命令行交互入口 # ---------------------------- def main(): 启动命令行交互演示检索增强 Agent。 print( * 60) print(第五章示例检索增强与知识库 Agent) print( * 60) print(说明) print(1. 请先在 knowledge_base/ 目录下放置 .md 文档。) print(2. 输入问题后Agent 会自动检索知识库并回答。) print(3. 输入 q / quit / exit 退出。) print( * 60) session_id rag_demo_session while True: try: user_input input(\n你).strip() except (EOFError, KeyboardInterrupt): print(\n再见) break if not user_input: continue if user_input.lower() in (q, quit, exit, 退出): print(再见) break answer ask_agent(session_id, user_input) print(fAI{answer}) if __name__ __main__: main()七、示例知识库文档1.knowledge_base/docker_kubernetes_guide.md1# Docker 与 Kubernetes 容器编排基础指南 2 3## 一、Docker 基础 4 5Docker 是一种容器化技术允许开发者将应用及其依赖打包成一个轻量、可移植的容器。 6 7常用命令 8 9- docker build -t myapp:latest .根据 Dockerfile 构建镜像。 10- docker run -d -p 8080:80 myapp:latest后台运行容器并将宿主机的 8080 端口映射到容器的 80 端口。 11- docker ps查看正在运行的容器。 12- docker logs container_id查看容器日志。 13- docker exec -it container_id /bin/bash进入正在运行的容器。 14 15## 二、Docker Compose 16 17Docker Compose 用于定义和运行多容器应用。通过 docker-compose.yml 文件声明服务、网络和卷。 18 19示例 20 21yaml 22version: 3.8 23services: 24 web: 25 image: nginx:latest 26 ports: 27 - 80:80 28 db: 29 image: mysql:8.0 30 environment: 31 MYSQL_ROOT_PASSWORD: example八、验证脚本verify_rag.py1 2检索增强能力验证脚本 3 4运行方式 51. 确保 knowledge_base/ 目录下已有示例文档。 62. 运行python verify_rag.py 7 8 9import os 10import json 11from agent_with_rag import ask_agent, save_chat_history, save_user_profile 12 13# ---------------------------- 14# 1. 测试用例 15# ---------------------------- 16 17RAG_TEST_CASES [ 18 { 19 name: 知识库检索Docker 常用命令, 20 session_id: rag_verify_001, 21 steps: [ 22 { 23 input: Docker 有哪些常用命令, 24 expect_keywords: [docker, 命令], 25 }, 26 { 27 input: 如何查看正在运行的容器, 28 expect_keywords: [docker ps], 29 }, 30 ], 31 }, 32 { 33 name: 知识库检索Kubernetes 核心概念, 34 session_id: rag_verify_002, 35 steps: [ 36 { 37 input: Kubernetes 中 Pod 是什么, 38 expect_keywords: [Pod], 39 }, 40 { 41 input: Deployment 的作用是什么, 42 expect_keywords: [Deployment], 43 }, 44 ], 45 }, 46 { 47 name: 知识库检索产品 FAQ, 48 session_id: rag_verify_003, 49 steps: [ 50 { 51 input: 如何重置密码, 52 expect_keywords: [忘记密码, 重置], 53 }, 54 { 55 input: 支持哪些支付方式, 56 expect_keywords: [微信, 支付宝, 银联, Visa, MasterCard], 57 }, 58 ], 59 }, 60 { 61 name: 检索空结果处理, 62 session_id: rag_verify_004, 63 steps: [ 64 { 65 input: 你们公司明天几点放假, 66 expect_not_keywords: [], 67 }, 68 ], 69 }, 70] 71 72# ---------------------------- 73# 2. 验证逻辑 74# ---------------------------- 75 76def run_rag_verification(): 77 逐用例执行检索增强验证。 78 79 report [] 80 passed 0 81 failed 0 82 83 for case in RAG_TEST_CASES: 84 session_id case[session_id] 85 86 # 清理会话历史避免受之前运行影响 87 save_chat_history(session_id, []) 88 89 # 清理用户画像避免长期记忆干扰 90 save_user_profile({}) 91 92 print(f\n{*60}) 93 print(f测试{case[name]}) 94 print(f会话ID{session_id}) 95 96 step_results [] 97 98 for step in case[steps]: 99 user_input step[input] 100 expect_keywords step.get(expect_keywords, []) 101 expect_not_keywords step.get(expect_not_keywords, []) 102 103 print(f\n 用户{user_input}) 104 105 answer ask_agent(session_id, user_input) 106 107 print(f AI{answer}) 108 109 # 检查必须出现的关键词 110 keyword_ok True 111 for kw in expect_keywords: 112 if kw not in answer: 113 keyword_ok False 114 print(f 期望回答中包含“{kw}”但未找到。) 115 116 # 检查不应出现的关键词 117 not_keyword_ok True 118 for kw in expect_not_keywords: 119 if kw in answer: 120 not_keyword_ok False 121 print(f 期望回答中不包含“{kw}”但找到了。) 122 123 step_passed keyword_ok and not_keyword_ok 124 status PASS if step_passed else FAIL 125 126 step_results.append({ 127 input: user_input, 128 output: answer, 129 status: status, 130 expect_keywords: expect_keywords, 131 expect_not_keywords: expect_not_keywords, 132 }) 133 134 if step_passed: 135 passed 1 136 else: 137 failed 1 138 139 report.append({ 140 name: case[name], 141 session_id: session_id, 142 steps: step_results, 143 }) 144 145 # 打印汇总 146 print(f\n{*60}) 147 print(f检索增强验证汇总共 {passed failed} 个步骤{passed} 通过{failed} 失败) 148 print(f{*60}) 149 150 # 保存报告 151 output_dir agent_traces 152 os.makedirs(output_dir, exist_okTrue) 153 154 report_file os.path.join(output_dir, rag_verify_report.json) 155 156 with open(report_file, w, encodingutf-8) as f: 157 json.dump(report, f, ensure_asciiFalse, indent2) 158 159 print(f\n验证报告已保存到{report_file}) 160 161 162if __name__ __main__: 163 run_rag_verification()九、运行说明安装依赖1pip install langchain langchain-openai faiss-cpu python-dotenv配置.env文件1OPENAI_API_KEYsk-xxxxxxxxxxxxxxxx 2MODEL_NAMEgpt-4o 3MODEL_BASE_URL 4EMBEDDING_MODELtext-embedding-3-small准备知识库文档确保knowledge_base/目录下有1knowledge_base/ 2├── docker_kubernetes_guide.md 3└── product_faq.md运行 Agent1python agent_with_rag.py运行验证脚本1python verify_rag.py十、本章小结概念作用本章实现文本分块把长文档切成可检索的小片段RecursiveCharacterTextSplitter向量化把文本转成语义向量OpenAIEmbeddings向量检索根据语义相似度查找文档FAISS.similarity_searchRAG检索 生成结合search_knowledge_base工具来源追溯让回答可验证返回片段时附带来源文件名到这里Agent 已经具备工具调用能力多步规划能力错误恢复能力短期记忆与长期记忆能力检索增强与知识库能力。