当前位置：首页 > news >正文

手把手搭建RAG+Agent智能问答Demo（LangChain+Chroma+BGE），附面试深挖清单

news 2026/6/30 6:41:37

一份能跑通、能写进简历的RAG+Agent学习项目，顺带把面试官爱问的考点全拆解了。

前言：为什么要做这个Demo
环境与依赖
项目架构——一张图看懂数据流
模块一：文档加载与分块
模块二：Embedding与向量库构建
模块三：检索与Rerank重排序
模块四：对话记忆与幻觉优化
模块五：Agent工具调用
模块六：知识库增量更新
主线联调与踩坑实录
轻量优化思路（仅针对Demo）
面试真题汇总 & 技术亮点总结

前言：为什么要做这个Demo

写这个Demo的时候，我刚好在重新梳理RAG和Agent的落地边界。本地知识库RAG问答 + Agent自主决策调用外部工具。整套代码不到600行，但涵盖了文档切片、向量检索、Rerank重排、对话记忆、提示词防幻觉、Agent ReAct范式、知识库增量添加这些核心点。用这个Demo当简历项目，至少能抗住80%的大模型应用开发基础面。

项目业务场景可以描述为：企业内部规章制度智能问答助手，当制度库中找不到相关条款时，自动转接通用搜索工具补充答案。

环境与依赖

先列一下我本地的配置，防止因为版本对不上跑不起来：

Python 3.10
LangChain 0.1.16
langchain-openai 0.1.3 (也可以换成ChatGLM等，后面细说)
chromadb 0.4.24
sentence-transformers 2.7.0
unstructured 0.13.0 (解析PDF/Word用)
python-dotenv 1.0.0
硬件：纯CPU也能跑，但Embedding加载模型略慢，建议至少16G内存

一键安装：

pip install langchain==0.1.16 langchain-openai chromadb sentence-transformers unstructured python-dotenv

我在项目根目录放了个.env文件，管理API Key和本地模型名：

OPENAI_API_KEY=sk-xxxx # Agent调用gpt-3.5用，也可以换deepseek BGE_MODEL_PATH=BAAI/bge-small-zh-v1.5 # Embedding模型，离线可用 RERANK_MODEL=BAAI/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_size=512, chunk_overlap=50) -> List[Document]: """ 递归文本切分，保留一定重叠防止关键信息被截断 """ splitter = RecursiveCharacterTextSplitter( separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""], # 中文友好 chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, ) 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 tokens），overlap一般取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_name=model_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_directory=persist_dir, embedding_function=embedding_model, collection_name="enterprise_qa" ) vector_store.add_documents(docs) else: vector_store = Chroma.from_documents( documents=docs, embedding=embedding_model, persist_directory=persist_dir, collection_name="enterprise_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服务（FastAPI+TorchServe），向量库调用远程Embedding接口。另外强调归一化很重要，因为余弦相似度归一化后等于内积，计算更快。
易错坑点：很多人忘记normalize_embeddings=True，导致Chroma默认用欧氏距离，检索效果明显变差。

面试题3：Chroma和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_k=10, top_n=3): self.vector_store = vector_store self.reranker = CrossEncoder(rerank_model_path, max_length=512) # 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, k=self.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), key=lambda x: x[1], reverse=True) 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_name="history"), # 历史对话 ("human", "【参考资料】\n{context}\n\n【用户问题】\n{input}"), ]) return prompt # memory在chain外维护 memory = ConversationBufferMemory(return_messages=True, memory_key="history")

面试题5：你这里用ConversationBufferMemory，对话长了不会爆Token吗？

深挖方向：立刻追问“还有其他Memory类型用过吗？怎么做长对话摘要？”
踩分点：BufferMemory直白简单适合Demo。生产上会改用ConversationSummaryMemory，定期对历史做摘要压缩，或使用滑动窗口只保留最近N条。我还会提一句ConversationTokenBufferMemory可以直接限制Token数，面试官一听就知道你真实用过。
易错坑点：Prompt里占位符名称不匹配导致历史信息传不进去，比如memory_key="history"但prompt里写成了chat_history，LangChain会静默失败，查半天。

面试题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( name="CompanyKnowledgeBase", func=search_kb, description="查询企业内部知识库，输入问句，返回相关规章制度片段。" ) def web_search_tool(): """ 简化的搜索工具，实际可用SerpAPI或自定义 """ def web_search(query: str) -> str: # 这里mock一个搜索接口，真实环境换成Google/Bing API # 只做演示，所以返回个固定提示 return f"这是通过网络搜索 '{query}' 获取的结果（演示占位）。" return Tool( name="WebSearch", func=web_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( agent=agent, tools=tools, memory=memory, handle_parsing_errors=True, max_iterations=5, # 防止无限循环 verbose=True # 演示时方便看Thought过程 ) return executor

面试题7：Agent为什么选ReAct范式，和Function Calling有啥区别？

深挖方向：会问“ReAct的观察-思考-行动循环具体怎么工作？你怎么处理Agent解析失败？”
踩分点：ReAct让模型显式输出推理过程（Thought），可解释性强，调试方便。Function Calling是模型直接输出结构化函数调用，对微调效果依赖强，用通用模型容易格式错误。我设置handle_parsing_errors=True，解析失败时将原始输出回灌给模型重试，保证鲁棒性。
易错坑点：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_path="BAAI/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_directory=persist_dir, embedding_function=self.embedding, collection_name="enterprise_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_size=512, chunk_overlap=50) 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(model="gpt-3.5-turbo", temperature=0.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传给prompt（Agent内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(f"Assistant: {result['output']}")