一份能跑通、能写进简历的RAGAgent学习项目顺带把面试官爱问的考点全拆解了。前言为什么要做这个Demo环境与依赖项目架构——一张图看懂数据流模块一文档加载与分块模块二Embedding与向量库构建模块三检索与Rerank重排序模块四对话记忆与幻觉优化模块五Agent工具调用模块六知识库增量更新主线联调与踩坑实录轻量优化思路仅针对Demo面试真题汇总 技术亮点总结前言为什么要做这个Demo写这个Demo的时候我刚好在重新梳理RAG和Agent的落地边界。本地知识库RAG问答 Agent自主决策调用外部工具。整套代码不到600行但涵盖了文档切片、向量检索、Rerank重排、对话记忆、提示词防幻觉、Agent ReAct范式、知识库增量添加这些核心点。用这个Demo当简历项目至少能抗住80%的大模型应用开发基础面。项目业务场景可以描述为企业内部规章制度智能问答助手当制度库中找不到相关条款时自动转接通用搜索工具补充答案。环境与依赖先列一下我本地的配置防止因为版本对不上跑不起来Python 3.10LangChain 0.1.16langchain-openai 0.1.3 (也可以换成ChatGLM等后面细说)chromadb 0.4.24sentence-transformers 2.7.0unstructured 0.13.0 (解析PDF/Word用)python-dotenv 1.0.0硬件纯CPU也能跑但Embedding加载模型略慢建议至少16G内存一键安装pip install langchain0.1.16 langchain-openai chromadb sentence-transformers unstructured python-dotenv我在项目根目录放了个.env文件管理API Key和本地模型名OPENAI_API_KEYsk-xxxx # Agent调用gpt-3.5用也可以换deepseek BGE_MODEL_PATHBAAI/bge-small-zh-v1.5 # Embedding模型离线可用 RERANK_MODELBAAI/bge-reranker-base面试考点这里已经埋了一个常被问到的问题——“为什么Embedding和生成模型不共用同一个”下面代码部分会展开。项目架构——一张图看懂数据流整体流程我画了个文字版先直观感受下文档(多格式) → 加载解析(unstructured) → 递归文本分块(RecursiveCharacterTextSplitter) → BGE Embedding向量化 → 存入Chroma向量库 ↓ 用户提问 → 向量检索(Top-K) → Rerank重排序(取Top-N) → 拼接历史记忆 → 拼装System Prompt(防幻觉约束) → LLM判断是否需要调用工具(Agent) → 如果知识库命中: 直接生成答案 → 如果信息不足: 调用搜索工具, 再汇总生成架构层面最容易被追问的三个点为什么在向量检索后还要接Rerank对话记忆放在Prompt拼接的哪个位置Agent决策的触发条件是什么会不会一直调用工具增加耗时后面每写一个模块我都会把对应的面试回答方式附上。模块一文档加载与分块先看加载和切分的代码。我这边用unstructured库统一处理.txt,.pdf,.docx避免针对不同格式写分支逻辑。# loader.py import os from typing import List from langchain.document_loaders import UnstructuredFileLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def load_documents(file_paths: List[str]) - List[Document]: 批量加载文件返回LangChain Document列表 all_docs [] for path in file_paths: if not os.path.exists(path): print(f文件 {path} 不存在跳过) continue # UnstructuredFileLoader 能自动识别格式 loader UnstructuredFileLoader(path) docs loader.load() all_docs.extend(docs) return all_docs def split_documents(docs: List[Document], chunk_size512, chunk_overlap50) - List[Document]: 递归文本切分保留一定重叠防止关键信息被截断 splitter RecursiveCharacterTextSplitter( separators[\n\n, \n, 。, , , , , , ], # 中文友好 chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, ) return splitter.split_documents(docs) # 使用示例 # files [docs/员工手册.pdf, docs/报销流程.docx] # raw_docs load_documents(files) # chunks split_documents(raw_docs)面试题1为什么RecursiveCharacterTextSplitter要用递归方式而不是固定字数硬切深挖方向面试官可能会追问“你是怎么设置chunk_size和overlap的依据是什么”踩分点递归切分优先保证段落/句子完整性避免把一句话硬生生砍成两截。chunk_size取决于Embedding模型的上下文窗口BGE是512 tokensoverlap一般取10%15%比较稳防止边界信息丢失。易错坑点把chunk_size设成和模型最大长度完全一致——但向量检索时短查询和长文档的相似度会严重失真实际512~768比较中庸。模块二Embedding与向量库构建Embedding我选的是BGE-small-zh因为离线环境不用调API而且中文语义匹配够用。向量库用Chroma单机零配置。# embedding_vectordb.py from langchain.embeddings import HuggingFaceBgeEmbeddings from langchain.vectorstores import Chroma from langchain.schema import Document from typing import List import os def get_embedding_model(model_path: str) - HuggingFaceBgeEmbeddings: 加载本地BGE Embedding模型 encode_kwargs{normalize_embeddings: True} 保证向量归一化 return HuggingFaceBgeEmbeddings( model_namemodel_path, model_kwargs{device: cpu}, # 可改cuda encode_kwargs{normalize_embeddings: True} ) def create_vector_store(docs: List[Document], embedding_model, persist_dir./chroma_db): 新建或增量到Chroma向量库 if os.path.exists(persist_dir) and os.listdir(persist_dir): # 如果已有库加载后追加 vector_store Chroma( persist_directorypersist_dir, embedding_functionembedding_model, collection_nameenterprise_qa ) vector_store.add_documents(docs) else: vector_store Chroma.from_documents( documentsdocs, embeddingembedding_model, persist_directorypersist_dir, collection_nameenterprise_qa ) vector_store.persist() return vector_store # 示例 # bge get_embedding_model(BAAI/bge-small-zh-v1.5) # vs create_vector_store(chunks, bge)面试题2你这里用的是HuggingFace的Embedding如果部署到生产模型加载太慢怎么办深挖方向面试官可能直接问“怎么把BGE部署成独立服务”或者“如何保证向量计算的一致性”踩分点可以封装成HTTP服务FastAPITorchServe向量库调用远程Embedding接口。另外强调归一化很重要因为余弦相似度归一化后等于内积计算更快。易错坑点很多人忘记normalize_embeddingsTrue导致Chroma默认用欧氏距离检索效果明显变差。面试题3Chroma和FAISS怎么选Milvus呢深挖方向追问“Chroma存百万级别向量撑得住吗”踩分点Demo场景优先选零运维的Chroma轻量级直接持久化。FAISS更适用于纯内存检索且需要高级索引IVF、HNSW时但不自带元数据过滤。Milvus是分布式的适合线上大规模。我这里明确说了是Demo用Chroma完全够顺带展示我知道生产选型逻辑。模块三检索与Rerank重排序检索器返回Top-10然后我用BGE-Reranker重排取前3条塞给LLM。这个操作在中文场景下提升很大因为向量相似度有时会被高频词带偏。# retriever_rerank.py from langchain.vectorstores import Chroma from sentence_transformers import CrossEncoder from typing import List, Tuple class RerankRetriever: def __init__(self, vector_store: Chroma, rerank_model_path: str, top_k10, top_n3): self.vector_store vector_store self.reranker CrossEncoder(rerank_model_path, max_length512) # CrossEncoder直接打分 self.top_k top_k self.top_n top_n def retrieve(self, query: str) - List[str]: 先粗排后精排返回top_n个最相关文档内容 # 粗排向量相似度取top_k raw_docs self.vector_store.similarity_search(query, kself.top_k) if not raw_docs: return [] # 构造 (query, doc) 对 pairs [[query, doc.page_content] for doc in raw_docs] scores self.reranker.predict(pairs) # 返回list of float # 按分数重新排序截取top_n scored_docs sorted(zip(raw_docs, scores), keylambda x: x[1], reverseTrue) top_docs [doc.page_content for doc, score in scored_docs[:self.top_n]] return top_docs # 使用 # retriever RerankRetriever(vector_store, BAAI/bge-reranker-base) # context retriever.retrieve(年假怎么申请)面试题4为什么有了向量检索还要Rerank直接多返回几条不就好了深挖方向“Rerank模型的原理是什么CrossEncoder和Bi-Encoder区别”踩分点向量检索是独立编码query和doc再算余弦属于“双塔”模式速度快但忽略细粒度交互。Rerank用CrossEncoder将query和doc拼接后送入Transformer充分交互重排序质量明显更高。代价是计算量巨大所以只对少量候选精排。易错坑点如果直接让LLM看10条文档上下文窗口容易爆而且无关内容引入噪声Rerank相当于在信息入口处做了筛选。这个Demo没做Rerank的话多轮对话幻觉率明显上升。模块四对话记忆与幻觉优化记忆用ConversationBufferMemory简单存最近K轮。防幻觉主要靠System Prompt强约束“只能根据提供的资料回答不知道就明确说不知道禁止编造”。# memory_prompt.py from langchain.memory import ConversationBufferMemory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder def build_chat_prompt(system_extra) - ChatPromptTemplate: 拼接系统提示词约束模型行为 base_system ( 你是一个严格遵守企业知识库的智能助手。\n 1. 只根据【参考资料】回答若无法从中得出答案请直接说根据现有资料无法回答该问题。\n 2. 不要编造任何信息不要使用外部知识。\n 3. 回答简洁引用资料中相关原文片段。\n ) if system_extra: base_system system_extra prompt ChatPromptTemplate.from_messages([ (system, base_system), MessagesPlaceholder(variable_namehistory), # 历史对话 (human, 【参考资料】\n{context}\n\n【用户问题】\n{input}), ]) return prompt # memory在chain外维护 memory ConversationBufferMemory(return_messagesTrue, memory_keyhistory)面试题5你这里用ConversationBufferMemory对话长了不会爆Token吗深挖方向立刻追问“还有其他Memory类型用过吗怎么做长对话摘要”踩分点BufferMemory直白简单适合Demo。生产上会改用ConversationSummaryMemory定期对历史做摘要压缩或使用滑动窗口只保留最近N条。我还会提一句ConversationTokenBufferMemory可以直接限制Token数面试官一听就知道你真实用过。易错坑点Prompt里占位符名称不匹配导致历史信息传不进去比如memory_keyhistory但prompt里写成了chat_historyLangChain会静默失败查半天。面试题6防幻觉提示词真的有用吗有没有更好的技术手段深挖方向会问“RAG评测里经常用Faithfulness指标你怎么保证生成内容忠实于检索结果”踩分点提示词是性价比最高的方法但治标不治本。更彻底的做法包括要求模型逐句标注引用来源、检索时强制保留原文元数据后置校验、或者用另一个NLI模型对生成内容做事实一致性打分。Demo阶段提示词足够应对学习场景。模块五Agent工具调用这部分是整套Demo的灵魂。我用了LangChain的ReAct Agent工具列表配了一个“公司规章制度查询”的函数工具和一个“联网搜索”的备用工具。当知识库查不到时Agent自动决定调用搜索工具。# agent_tools.py from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from retriever_rerank import RerankRetriever import requests def knowledge_base_tool(retriever: RerankRetriever): 将我们之前写的检索器包装成Tool def search_kb(query: str) - str: docs retriever.retrieve(query) if not docs: return 知识库中未找到相关信息。 return \n\n.join(docs) return Tool( nameCompanyKnowledgeBase, funcsearch_kb, description查询企业内部知识库输入问句返回相关规章制度片段。 ) def web_search_tool(): 简化的搜索工具实际可用SerpAPI或自定义 def web_search(query: str) - str: # 这里mock一个搜索接口真实环境换成Google/Bing API # 只做演示所以返回个固定提示 return f这是通过网络搜索 {query} 获取的结果演示占位。 return Tool( nameWebSearch, funcweb_search, description当知识库无法回答问题时使用此工具进行联网搜索。输入搜索关键词。 ) def build_agent(llm, retriever: RerankRetriever, memory): 构造ReAct Agent绑工具和记忆 tools [knowledge_base_tool(retriever), web_search_tool()] # ReAct模板 react_prompt PromptTemplate.from_template( Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Previous conversation history: {chat_history} Question: {input} Thought: {agent_scratchpad} ) agent create_react_agent(llm, tools, react_prompt) executor AgentExecutor( agentagent, toolstools, memorymemory, handle_parsing_errorsTrue, max_iterations5, # 防止无限循环 verboseTrue # 演示时方便看Thought过程 ) return executor面试题7Agent为什么选ReAct范式和Function Calling有啥区别深挖方向会问“ReAct的观察-思考-行动循环具体怎么工作你怎么处理Agent解析失败”踩分点ReAct让模型显式输出推理过程Thought可解释性强调试方便。Function Calling是模型直接输出结构化函数调用对微调效果依赖强用通用模型容易格式错误。我设置handle_parsing_errorsTrue解析失败时将原始输出回灌给模型重试保证鲁棒性。易错坑点max_iterations设太大容易对话死循环设太小任务未完成就提前退出我这里设5次是Demo实测的平衡值。面试题8你这里的WebSearch工具明显是Mock的真的接入搜索API要注意什么深挖方向“如何把搜索结果和知识库结果融合怎么避免敏感信息外泄”踩分点搜索结果URL需要过滤、摘要截断调用API前对query做脱敏结果返回后用同一个Rerank模型与知识库片段混合排序保证最终上下文质量一致。Demo简化了但逻辑框架一样。模块六知识库增量更新实际业务文档会持续新增我单独写了个增量添加接口避免每次重建向量库。# knowledge_manager.py from langchain.vectorstores import Chroma from embedding_vectordb import get_embedding_model import os class KnowledgeManager: def __init__(self, persist_dir./chroma_db, model_pathBAAI/bge-small-zh-v1.5): self.persist_dir persist_dir self.embedding get_embedding_model(model_path) if os.path.exists(persist_dir): self.vector_store Chroma( persist_directorypersist_dir, embedding_functionself.embedding, collection_nameenterprise_qa ) else: self.vector_store None def add_documents(self, docs): if self.vector_store is None: raise ValueError(向量库未初始化请先创建。) self.vector_store.add_documents(docs) self.vector_store.persist() print(f成功添加 {len(docs)} 个文档片段) # 使用 # km KnowledgeManager() # new_chunks split_documents(load_documents([新政策.pdf])) # km.add_documents(new_chunks)面试题9增量添加时会不会导致Embedding不一致之前存的向量和后来加的新模型怎么对齐深挖方向问“如果Embedding模型升级了存量向量怎么办”踩分点保证一直用同一个模型名字和参数包括normalize_embeddings新生成的向量就和旧向量在同一空间。模型迭代时必须全量重建索引不能增量。这里用persist_directory和固定模型路径锁死Demo场景不用担心。易错坑点如果偷偷换了模型或者改了normalize参数Chroma不会报错但相似度会完全乱套查出来毫无关联的内容。主线联调与踩坑实录把所有模块串起来的main.py长这样# main.py from dotenv import load_dotenv load_dotenv() from loader import load_documents, split_documents from embedding_vectordb import get_embedding_model, create_vector_store from retriever_rerank import RerankRetriever from memory_prompt import build_chat_prompt, memory from agent_tools import build_agent from langchain_openai import ChatOpenAI import os # 1. 初始化 files [docs/员工手册.pdf, docs/报销制度.docx] raw load_documents(files) chunks split_documents(raw, chunk_size512, chunk_overlap50) bge get_embedding_model(os.getenv(BGE_MODEL_PATH)) vs create_vector_store(chunks, bge) # 首次运行创建 retriever RerankRetriever(vs, os.getenv(RERANK_MODEL)) # 2. 准备LLM这里用OpenAI gpt-3.5可换本地 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.1) # 3. 构建Agent agent_executor build_agent(llm, retriever, memory) # 4. 交互循环 print(知识库助手已启动输入exit退出) while True: user_input input(User: ) if user_input.lower() exit: break # 先检索一下作为context传给promptAgent内tool会再次检索这里是为了一致性演示实际可简化 context_docs retriever.retrieve(user_input) context_str \n\n.join(context_docs) prompt build_chat_prompt() # 格式化消息传入executor result agent_executor.invoke({ input: user_input, context: context_str, chat_history: memory.buffer_as_str # 历史字符串化 }) print(fAssistant: {result[output]})踩坑1Chroma持久化后每次重启都需要persist_directory一致否则数据丢失第一次跑完忘记指定persist_directory第二天重启发现知识库空了。Chroma默认全内存模式必须显式持久化并在加载时传入相同路径。排查时在Chroma()初始化里加了个print查看collection的文档数量。面试映射这个问题可以直接对应故障排查题“向量库查询为空可能的原因有哪些” 参考答案路径不对、Embedding模型不匹配、未调用persist()、Collection被误删。踩坑2Agent解析Final Answer格式失败一直报ParsingError加了handle_parsing_errorsTrue后观察到模型有时输出“Final Answer: ...”但冒号中间有个中文符号。LangChain正则没匹配到。解决方式是在Prompt里强化格式要求“Final Answer后紧跟英文冒号”。对不强模型来说经常发生。面试映射“Agent解析失败你怎么处理的” 可以聊Prompt工程约束 代码兜底重试再展开讲结构化输出的重要性。轻量优化思路仅针对Demo异步检索现在检索和LLM调用是串行的可以改为检索预先异步进行Agent决策时直接取结果降低首字延迟。缓存常见问题对高频query做一层Redis缓存Demo可以用字典减少重复Embedding和Rerank计算。分批Embedding文档过多时一次性调BGE会OOM加个batch_size循环处理就好。Agent提示词更精细把工具描述写详细参数类型、使用时机能大幅减少误调用。这些都是几行代码就能加的优化不涉及分布式改造只是让Demo跑得更顺。面试真题汇总 技术亮点总结我把全文涉及的面试题整理出来方便准备简历的读者对照自测为什么用RecursiveCharacterTextSplitterchunk_size和overlap怎么定的BGE Embedding为什么要归一化如何部署成独立服务Chroma、FAISS、Milvus选型差异与适用场景Rerank的原理是什么CrossEncoder和Bi-Encoder区别对话记忆爆Token怎么办总结记忆和滑动窗口怎么实现防幻觉提示词设计思路还有哪些更彻底的技术方案Agent选ReAct还是Function Calling解析失败如何处理增量添加文档时怎么保证向量空间一致整个系统从输入到输出的延迟瓶颈在哪里如果知识库和工具返回冲突信息Agent如何抉择本文项目的核心技术亮点完整闭环文档加载→分块→向量化→检索Rerank→记忆→Agent决策→多工具联动工业界落地思路Demo虽小模块拆分清晰可轻松替换任意组件向量库、模型、工具最后欢迎大家在评论区贴出你运行时的报错截图一起讨论本文全程实战手写希望能给正在学习大模型落地的同学一点接地气的参考。