在实际企业级 AI 应用开发中如何将私有知识、文档数据与大语言模型LLM的能力安全、高效地结合是一个核心挑战。直接让 LLM 处理长文档或回答专业问题往往会遇到“幻觉”生成不准确信息、上下文长度限制和知识更新滞后等问题。检索增强生成RAG技术正是为解决这些问题而生它通过“检索”相关文档片段来“增强”LLM的生成过程确保回答基于可信来源。而 LangChain 作为当前最流行的 LLM 应用开发框架提供了构建 RAG 系统所需的模块化组件和抽象让开发者能更专注于业务逻辑而非底层连接。本文将围绕 LangChain 与 RAG 技术栈从零开始构建一个企业级 AI 知识库的原型。我们将不局限于简单的问答而是深入向量数据库选型、检索策略优化并最终引入 Agent 智能体让系统具备自主调用工具、执行多步骤复杂任务的能力。整个过程会涉及环境搭建、代码实现、核心参数调优以及生产环境下的常见问题排查目标是交付一个可学习、可复现、可扩展的技术方案。1. 理解 RAG 与 LangChain 的核心工作机制在动手写代码之前必须厘清几个核心概念以及它们是如何协同工作的。这能帮助你在后续配置出错或效果不佳时快速定位问题所在。1.1 RAG 系统的基本流程检索、增强、生成一个标准的 RAG 流程可以拆解为索引Indexing和查询Querying两个阶段。索引阶段的目标是将原始的非结构化文档如 PDF、Word、Markdown转化为便于检索的格式。这个过程通常包括加载Loading使用文档加载器Document Loader从文件系统、数据库或网络读取文档内容。分割Splitting由于 LLM 有上下文窗口限制必须将长文档切分成语义连贯的“块”Chunks。分割策略如按字符、按句子、按段落和块大小、重叠区Overlap的设置至关重要直接影响检索质量。嵌入Embedding使用嵌入模型Embedding Model将每个文本块转换为一个高维向量Vector。这个向量在数学上表征了文本的语义信息语义相近的文本其向量在空间中的距离也更近。存储Storing将文本块及其对应的向量存储到向量数据库Vector Database中。向量数据库专门为高效的海量向量相似性搜索而设计。查询阶段是用户与系统交互的过程问题嵌入将用户提出的问题Query同样通过嵌入模型转换为向量。相似性检索在向量数据库中搜索与问题向量最相似的 K 个文本块向量。这个过程就是“检索”其核心是找到与问题语义最相关的知识片段。上下文构建将检索到的 K 个文本块作为“上下文”Context与原始问题一起按照特定的提示词Prompt模板进行组装形成最终的输入。生成回答将组装好的提示词发送给 LLM如 GPT、Claude 或本地部署的模型LLM 基于给定的上下文生成最终答案。这就是“增强生成”它强制 LLM 在提供的材料中寻找答案大幅减少幻觉。1.2 LangChain 的模块化设计用“链”组装复杂应用LangChain 不是一个黑盒应用而是一个工具箱。它将 RAG 流程中的每个步骤抽象成了独立的模块并通过“链”Chain的概念将它们串联起来。理解这些核心模块是灵活运用 LangChain 的关键文档加载器Document LoadersDirectoryLoader,PyPDFLoader,UnstructuredFileLoader等用于从各种来源加载文档。文本分割器Text SplittersRecursiveCharacterTextSplitter,CharacterTextSplitter等负责将文档切块。嵌入模型Embedding ModelsOpenAIEmbeddings,HuggingFaceEmbeddings等提供文本到向量的转换能力。这里需要区分“嵌入模型”和后续的“LLM”它们通常是两个独立的模型。向量存储VectorstoresChroma,FAISS,Pinecone云服务,Milvus等是向量数据库在 LangChain 中的接口封装。大语言模型LLMs/Chat ModelsChatOpenAI,ChatAnthropic,Ollama本地等负责最终的文本生成。链ChainsRetrievalQA,ConversationalRetrievalChain等是预组装好的、针对特定任务如问答的工作流。你也可以用LCELLangChain Expression Language自定义链。智能体Agentscreate_react_agent,create_openai_tools_agent等是能理解目标、自主选择并调用工具如搜索、计算、数据库查询的 LLM 增强系统。Agent 让 LLM 从“答题者”变为“执行者”。LangChain vs. LangGraphLangGraph 是 LangChain 的一个扩展库用于构建有状态的、多步骤的复杂工作流。如果说 LangChain 的 Chain 是线性管道那么 LangGraph 允许你构建带循环、分支和并行执行的有向图。对于简单的 RAGLangChain 足够对于需要多轮规划、自我修正的复杂 AgentLangGraph 更合适。2. 环境准备与核心依赖配置我们将构建一个基于本地文件的 RAG 系统并逐步升级到 Agent。为了兼顾学习成本和实用性技术栈选择如下编程语言Python 3.10核心框架LangChain向量数据库Chroma轻量易于本地部署和实验嵌入模型text-embedding-3-smallOpenAI API或BAAI/bge-small-zh-v1.5Hugging Face中文友好LLM初期使用 OpenAI GPT 系列 API稳定、效果佳后期可替换为本地模型如通过 Ollama 部署。其他工具用于 Agent 的搜索、计算等工具。2.1 创建项目与虚拟环境首先创建一个干净的项目目录并设置独立的 Python 环境这是避免依赖冲突的最佳实践。# 创建项目目录 mkdir enterprise-ai-knowledge-base cd enterprise-ai-knowledge-base # 创建并激活虚拟环境 (推荐使用 conda 或 venv) python -m venv venv # 在 Windows 上: venv\Scripts\activate # 在 macOS/Linux 上: source venv/bin/activate2.2 安装依赖包创建requirements.txt文件内容如下。注意我们根据功能模块对依赖进行了分组注释。# 核心框架 langchain0.1.0 langchain-community0.0.10 # 社区维护的集成组件 langchain-openai0.0.5 # OpenAI 官方集成 langchain-chroma0.1.0 # Chroma 向量数据库集成 # 向量数据库客户端 chromadb0.4.22 # 文档加载与处理 pypdf3.17.4 # 用于读取PDF unstructured[pdf,docx]0.10.30 # 强大的文档解析库 markdown3.5.1 # 嵌入模型 (可选如果使用 HuggingFace 模型) sentence-transformers2.2.2 # 网络请求与API调用 httpx0.25.1 openai1.6.1 # 开发与工具 (用于Agent示例) langchain-experimental0.0.49 # 包含一些实验性功能如高级Agent python-dotenv1.0.0 # 管理环境变量使用 pip 安装所有依赖pip install -r requirements.txt2.3 配置 API 密钥与环境变量为了安全地管理敏感信息如 OpenAI API Key务必使用环境变量。在项目根目录创建.env文件# .env 文件 OPENAI_API_KEYsk-your-openai-api-key-here # 如果使用其他模型服务可在此添加 # ANTHROPIC_API_KEY... # GROQ_API_KEY...在代码中使用python-dotenv加载配置# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量到环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. 构建基础 RAG 知识库从文档加载到智能问答现在我们开始实现一个最基础的 RAG 流水线。假设你的docs/目录下有一些 PDF 或 TXT 格式的公司文档。3.1 文档加载与智能分割文档分割是 RAG 的“暗物质”其质量直接决定检索的上限。一个糟糕的分割会把一个完整的概念切成两半导致检索时上下文缺失。# rag_core/document_processor.py from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from typing import List from langchain.schema import Document class DocumentProcessor: def __init__(self, chunk_size: int 1000, chunk_overlap: int 200): 初始化文档处理器。 :param chunk_size: 每个文本块的最大字符数。 :param chunk_overlap: 块之间的重叠字符数用于保持上下文连贯。 self.text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ] # 中文友好的分隔符 ) def load_documents_from_dir(self, directory_path: str, glob_pattern: str **/*.pdf) - List[Document]: 从目录加载所有指定类型的文档。 # 根据文件类型选择加载器 if glob_pattern.endswith(.pdf): loader DirectoryLoader(directory_path, globglob_pattern, loader_clsPyPDFLoader) elif glob_pattern.endswith(.txt): loader DirectoryLoader(directory_path, globglob_pattern, loader_clsTextLoader) else: # 使用 Unstructured 加载器处理多种格式 from langchain_community.document_loaders import UnstructuredFileLoader loader DirectoryLoader(directory_path, globglob_pattern, loader_clsUnstructuredFileLoader) print(f正在从 {directory_path} 加载文档...) documents loader.load() print(f成功加载 {len(documents)} 个文档。) return documents def split_documents(self, documents: List[Document]) - List[Document]: 将加载的文档分割成小块。 print(正在分割文档...) split_docs self.text_splitter.split_documents(documents) print(f文档被分割成 {len(split_docs)} 个文本块。) return split_docs # 使用示例 if __name__ __main__: processor DocumentProcessor(chunk_size800, chunk_overlap150) raw_docs processor.load_documents_from_dir(./docs, **/*.pdf) chunks processor.split_documents(raw_docs) # 查看第一个块的内容和元数据 if chunks: print(f示例块内容前300字符: {chunks[0].page_content[:300]}...) print(f元数据: {chunks[0].metadata})关键参数解释chunk_size通常设置在 500-1500 之间。太小会丢失上下文太大会引入噪声并增加 LLM 处理负担。需要根据你的文档类型技术文档、会议纪要、法律条文和 LLM 的上下文窗口进行权衡。chunk_overlap通常为chunk_size的 10%-20%。重叠可以防止一个句子或关键概念被硬生生切断是提升检索连贯性的简单有效手段。separators分割符列表按优先级尝试。我们加入了中文标点使其对中文文档更友好。3.2 向量化与存储连接 Chroma 数据库文本块准备好后需要将它们转化为向量并存入数据库。这里我们选择 OpenAI 的嵌入模型和 Chroma 向量数据库。# rag_core/vector_store_manager.py import chromadb from chromadb.config import Settings from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from langchain.schema import Document from typing import List, Optional import shutil import os class VectorStoreManager: def __init__(self, persist_directory: str ./chroma_db, embedding_model_name: str text-embedding-3-small): 初始化向量存储管理器。 :param persist_directory: Chroma 数据库持久化目录。 :param embedding_model_name: 使用的嵌入模型名称。 self.persist_directory persist_directory # 初始化嵌入函数 self.embeddings OpenAIEmbeddings(modelembedding_model_name) self.vector_store None def create_and_persist_from_documents(self, documents: List[Document], collection_name: str knowledge_base): 从文档列表创建向量存储并持久化。 # 如果目录已存在先删除仅用于演示生产环境应增量添加 if os.path.exists(self.persist_directory): print(f检测到已有数据库目录 {self.persist_directory}将重新创建...) shutil.rmtree(self.persist_directory) print(f正在创建向量存储集合名: {collection_name}...) self.vector_store Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory, collection_namecollection_name ) print(f向量存储已创建并保存至 {self.persist_directory}) def load_existing_vector_store(self, collection_name: str knowledge_base): 加载已持久化的向量存储。 if not os.path.exists(self.persist_directory): raise FileNotFoundError(f持久化目录 {self.persist_directory} 不存在请先创建向量存储。) print(f正在从 {self.persist_directory} 加载向量存储...) self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings, collection_namecollection_name ) count self.vector_store._collection.count() print(f向量存储加载成功集合 {collection_name} 中包含 {count} 个向量。) return self.vector_store def get_retriever(self, search_type: str similarity, search_kwargs: Optional[dict] None): 获取检索器用于后续的链中。 :param search_type: 检索类型如 similarity相似度, mmr最大边际相关性, similarity_score_threshold带阈值。 :param search_kwargs: 检索参数如 {k: 4} 或 {k: 4, score_threshold: 0.7}。 if self.vector_store is None: raise ValueError(向量存储未初始化请先加载或创建。) if search_kwargs is None: search_kwargs {k: 4} # 默认返回最相似的4个块 retriever self.vector_store.as_retriever( search_typesearch_type, search_kwargssearch_kwargs ) return retriever # 使用示例将上一节的 chunks 存入向量库 if __name__ __main__: from document_processor import DocumentProcessor processor DocumentProcessor() raw_docs processor.load_documents_from_dir(./docs, **/*.txt) chunks processor.split_documents(raw_docs) vs_manager VectorStoreManager(persist_directory./my_chroma_db) vs_manager.create_and_persist_from_documents(chunks, collection_namecompany_docs)关键点解析嵌入模型选择OpenAIEmbeddings需要有效的OPENAI_API_KEY。对于纯中文场景或内网环境可以考虑HuggingFaceEmbeddings例如model_name“BAAI/bge-small-zh-v1.5”。更换模型时整个向量库需要重建因为不同模型生成的向量空间不同。检索器Retriever这是 LangChain 中一个核心抽象它封装了从向量库中查找相关文档的逻辑。search_type和search_kwargs是调优检索效果的重要杠杆。持久化persist_directory指定了 Chroma 存储数据的本地路径。这意味着索引只需创建一次后续可以直接加载无需重复计算嵌入向量节省成本和时间。3.3 组装问答链让 RAG 跑起来有了检索器我们就可以将其与 LLM 组合成一个完整的问答链。# rag_core/qa_chain_builder.py from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from vector_store_manager import VectorStoreManager class QABot: def __init__(self, vector_store_manager: VectorStoreManager, model_name: str gpt-3.5-turbo): 初始化问答机器人。 :param vector_store_manager: 已初始化的向量存储管理器。 :param model_name: 使用的 LLM 模型名称。 self.llm ChatOpenAI(modelmodel_name, temperature0) # temperature0 使输出更确定 self.vector_store_manager vector_store_manager self.qa_chain None def build_qa_chain(self, prompt_template: str None): 构建检索问答链。 # 1. 获取检索器 retriever self.vector_store_manager.get_retriever(search_typesimilarity, search_kwargs{k: 4}) # 2. 定义提示词模板 if prompt_template is None: prompt_template 请根据以下上下文信息回答问题。如果你不知道答案就诚实地回答不知道不要编造信息。 上下文 {context} 问题{question} 基于上下文的答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 3. 构建链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, # 最常用的类型将所有检索到的上下文塞入提示词 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回源文档便于追溯和调试 ) print(问答链构建成功。) return self.qa_chain def ask(self, question: str): 提出问题并获取答案。 if self.qa_chain is None: self.build_qa_chain() print(f问题: {question}) result self.qa_chain.invoke({query: question}) answer result[result] source_docs result[source_documents] print(f答案: {answer}) print(\n--- 参考来源 ---) for i, doc in enumerate(source_docs[:2]): # 显示前两个来源 print(f[来源 {i1}] {doc.page_content[:200]}...) print(f 元数据: {doc.metadata}\n) return answer, source_docs # 主程序入口 if __name__ __main__: # 1. 加载已有的向量存储 vs_manager VectorStoreManager(persist_directory./my_chroma_db) vs_manager.load_existing_vector_store(collection_namecompany_docs) # 2. 创建问答机器人并提问 bot QABot(vs_manager, model_namegpt-3.5-turbo) bot.build_qa_chain() while True: user_question input(\n请输入您的问题输入 quit 退出: ) if user_question.lower() quit: break bot.ask(user_question)代码逻辑与调优点链类型chain_type我们使用了“stuff”它简单地将所有检索到的上下文拼接后传给 LLM。对于非常多的上下文可能会超出令牌限制。其他选项包括“map_reduce”分别总结再汇总、“refine”迭代优化答案和“map_rerank”它们适用于处理大量文档但更复杂且可能损失精度。提示词工程模板中的指令“如果你不知道答案就诚实地回答不知道”对于减少幻觉非常关键。你可以根据业务需求定制模板例如要求答案以特定格式列出或引用源文档的页码。返回源文档return_source_documentsTrue是调试和建立用户信任的必备功能。你可以展示答案的依据让用户判断可信度。运行此脚本一个基础的、基于私有文档的知识问答系统就构建完成了。你可以向docs/目录添加新的文档然后重建向量库来更新知识。4. 进阶优化检索效果与引入 Agent 智能体基础 RAG 可能面临检索不准、答案冗长或无法处理复杂问题等挑战。下面我们进行两方面的增强优化检索策略和引入智能体。4.1 优化检索策略超越简单相似度搜索简单的向量相似度搜索有时会返回相关但不精确的片段。以下是几种优化方案方案一使用 MMR最大边际相关性检索MMR 在保证相关性的同时增加结果集的多样性避免返回多个高度重复的片段。# 在 VectorStoreManager.get_retriever 中调整参数 retriever self.vector_store.as_retriever( search_typemmr, # 改为 MMR search_kwargs{k: 6, fetch_k: 20, lambda_mult: 0.7} # fetch_k 是初始候选集大小lambda_mult 控制多样性权重 )方案二混合检索Hybrid Search结合向量搜索语义相似和关键词搜索字面匹配取长补短。这需要向量数据库支持如 Weaviate, Qdrant。Chroma 的新版本也开始支持。方案三查询转换Query Transformation在检索前对用户问题进行改写或扩展例如生成多个相关问题再进行检索Multi-Query或将复杂问题分解Step-Back Prompting。这可以通过额外的 LLM 调用实现。方案四重排序Re-ranking先召回较多的候选文档如 k20再用一个更精细的交叉编码器模型Cross-Encoder对它们进行重排序选出最相关的几个如 top_k4。这能显著提升精度但会增加延迟和计算成本。# 伪代码示例使用 Cohere 或 BGE 的重排序器 # from langchain.retrievers import ContextualCompressionRetriever # from langchain.retrievers.document_compressors import CrossEncoderReranker # compressor CrossEncoderReranker(modelBAAI/bge-reranker-large, top_n4) # compression_retriever ContextualCompressionRetriever(base_retrieverretriever, base_compressorcompressor)4.2 构建第一个 Agent让 LLM 学会使用工具Agent 的核心思想是赋予 LLM 使用工具Tools的能力。LLM 根据用户目标自主决定调用哪个工具、传入什么参数并根据工具返回的结果规划下一步行动直到完成任务。让我们构建一个简单的 Agent它除了能查询知识库还能进行数学计算和搜索网络以模拟搜索为例。首先定义几个工具# agent/tools.py from langchain.tools import Tool from langchain_community.utilities import WikipediaAPIWrapper from langchain.chains import LLMMathChain from langchain_openai import ChatOpenAI from vector_store_manager import VectorStoreManager from qa_chain_builder import QABot import json def setup_knowledge_base_tool(vector_store_manager): 将我们之前构建的 RAG 系统封装成一个工具 qa_bot QABot(vector_store_manager) qa_bot.build_qa_chain() def knowledge_base_lookup(query: str) - str: 用于查询公司内部知识库。输入应是一个明确的问题。 answer, sources qa_bot.ask(query) # 将答案和来源格式化为字符串 result { answer: answer, sources: [{content: doc.page_content[:500], metadata: doc.metadata} for doc in sources[:2]] } return json.dumps(result, ensure_asciiFalse) return Tool( nameCompany_Knowledge_Base, funcknowledge_base_lookup, description当需要回答关于公司产品、政策、流程或历史文档的具体问题时使用此工具。输入必须是一个完整的问题。 ) def setup_calculator_tool(llm): 数学计算工具 math_chain LLMMathChain.from_llm(llmllm, verboseFalse) return Tool( nameCalculator, funcmath_chain.run, description用于执行数学计算。输入应是一个数学表达式例如 2 2 或 sin(45度)。 ) def setup_wikipedia_tool(): 维基百科搜索工具示例需要网络 wikipedia WikipediaAPIWrapper(top_k_results2, doc_content_chars_max500) return Tool( nameWikipedia, funcwikipedia.run, description用于获取关于人物、地点、事件、概念等通用事实信息。输入应是一个明确的搜索主题。 )然后创建 Agent 并运行# agent/company_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain import hub # 用于拉取预定义的提示词 from tools import setup_knowledge_base_tool, setup_calculator_tool, setup_wikipedia_tool from vector_store_manager import VectorStoreManager from langchain_openai import ChatOpenAI def create_company_agent(): # 1. 初始化组件 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) vs_manager VectorStoreManager(persist_directory./my_chroma_db) vs_manager.load_existing_vector_store() # 2. 准备工具列表 tools [ setup_knowledge_base_tool(vs_manager), setup_calculator_tool(llm), setup_wikipedia_tool() ] # 3. 获取 ReAct 提示词模板 prompt hub.pull(hwchase17/react) # 一个经典的 ReAct 格式提示词 # 4. 创建 Agent agent create_react_agent(llm, tools, prompt) # 5. 创建执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 打印 Agent 的思考过程便于调试 handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5 # 防止无限循环 ) return agent_executor if __name__ __main__: agent create_company_agent() # 测试复杂问题 questions [ 我们公司最新的年假政策是怎样的然后请帮我计算如果我有15天年假已经用了7.5天还剩多少天, 爱因斯坦的主要贡献是什么顺便告诉我他的出生年份。, 根据知识库我们项目X的截止日期是什么时候并计算从今天到截止日期还有多少天。 ] for q in questions: print(f\n{*60}) print(f用户问题: {q}) print(f{*60}) try: result agent.invoke({input: q}) print(f\n最终答案: {result[output]}) except Exception as e: print(fAgent 执行出错: {e})运行这个 Agent你会看到verboseTrue模式下输出的详细思考过程Thought, Action, Observation这正是 ReAct 框架的核心。Agent 会判断问题类型决定先调用知识库工具再调用计算器工具最后整合答案。4.3 处理复杂工作流LangGraph 初探当任务需要严格的步骤顺序、循环或分支时基础的 Agent 可能不够用。例如“监控一个工单状态直到它被解决然后发邮件通知”。这时可以用 LangGraph。LangGraph 将工作流定义为“图”Graph节点是函数或工具边是条件转移。下面是一个极简的审批流程示例# agent/graph_workflow.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langchain_openai import ChatOpenAI import operator # 1. 定义状态结构 class AgentState(TypedDict): question: str context: Annotated[list, operator.add] # 用于累积上下文信息 answer: str needs_approval: bool # 2. 定义节点函数 def retrieve_node(state: AgentState): 模拟检索步骤 print(f[检索节点] 正在处理问题: {state[question]}) # 这里可以接入真实的 RAG 检索 state[context].append(f检索到与{state[question]}相关的3份文档。) state[needs_approval] 财务 in state[question] # 假设涉及财务的问题需要审批 return state def generate_answer_node(state: AgentState): 模拟生成答案步骤 print(f[生成节点] 基于上下文生成答案。) state[answer] f根据 {state[context][-1]}生成的初步答案是这是一个示例回答。 return state def approval_node(state: AgentState): 模拟审批步骤 if state[needs_approval]: print([审批节点] 问题涉及财务需要经理审批...模拟等待) state[answer] (已通过经理审批) else: print([审批节点] 无需审批直接通过。) return state # 3. 构建图 def create_approval_graph(): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_answer_node) workflow.add_node(approval, approval_node) # 设置边 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, generate) # 条件边根据 needs_approval 决定是否走审批节点 workflow.add_conditional_edges( generate, lambda x: approval if x[needs_approval] else END, {approval: approval, END: END} ) workflow.add_edge(approval, END) # 编译图 return workflow.compile() # 运行图 if __name__ __main__: graph create_approval_graph() # 执行图传入初始状态 result graph.invoke({question: 请问本季度的财务预算是多少, context: [], answer: , needs_approval: False}) print(f\n最终状态: {result})这个例子展示了如何通过定义状态和节点来构建一个可控的、有状态的工作流。对于需要严格流程控制的业务场景如客服工单处理、数据审核流水线LangGraph 比传统链或简单 Agent 更强大。5. 部署、监控与生产环境最佳实践让一个原型在本地运行起来是一回事将其部署为稳定可靠的生产服务是另一回事。5.1 部署方案选型部署场景推荐方案说明快速原型/内部工具FastAPI Docker将核心逻辑封装为 FastAPI 接口用 Docker 容器化。易于部署到云服务器或 Kubernetes。需要Web界面Gradio / Streamlit快速构建交互式 UI适合演示和内部工具。可同样用 Docker 部署。企业级微服务LangServeLangChain 官方服务化框架能自动将 Chain 或 Agent 暴露为 API并生成 OpenAPI 文档。云原生/高并发Kubernetes 服务网格将向量数据库、LLM 代理、应用服务分别部署通过服务网格管理通信和流量。一个简单的 FastAPI 部署示例# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent.company_agent import create_company_agent import uvicorn app FastAPI(title企业知识库问答 API) agent None class QueryRequest(BaseModel): question: str app.on_event(startup) async def startup_event(): global agent print(正在初始化 Agent...) agent create_company_agent() # 注意初始化可能较慢考虑异步或健康检查 print(Agent 初始化完成。) app.post(/ask) async def ask_question(request: QueryRequest): if agent is None: raise HTTPException(status_code503, detail服务正在初始化请稍后重试。) try: result agent.invoke({input: request.question}) return {answer: result[output]} except Exception as e: raise HTTPException(status_code500, detailf处理问题时出错: {str(e)}) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)5.2 关键生产环境考量配置外置化所有 API Key、模型名称、数据库连接字符串、超时参数等必须通过环境变量或配置中心管理绝不能硬编码。异步处理LLM 调用和向量检索可能是 I/O 密集型操作。使用asyncio和异步 HTTP 客户端如httpx来避免阻塞提高并发能力。超时与重试为所有外部服务调用OpenAI API、向量数据库设置合理的超时和重试机制。限流与熔断防止滥用或下游服务故障导致系统雪崩。可以使用tenacity进行重试使用circuitbreaker实现熔断。日志与监控结构化日志记录每个请求的问题、答案、检索到的源文档 ID、Token 使用量、耗时和任何错误。这对于调试和成本分析至关重要。应用性能监控集成如 Prometheus Grafana监控请求延迟、错误率、Token 消耗等指标。链路追踪在微服务架构下使用 OpenTelemetry 追踪一个请求经过 RAG、LLM、Agent 等组件的完整路径。成本控制缓存频繁查询的问题-答案对。监控并分析 Token 使用情况优化提示词长度。对于内部知识考虑使用更小的、领域微调过的开源模型通过 Ollama、vLLM 等部署以降低 API 成本。5.3 常见问题排查清单当你的 RAG 或 Agent 系统出现问题时可以按照以下清单逐项排查问题现象可能原因检查点与解决方案答案与文档内容不符幻觉1. 检索到的上下文不相关。2. 提示词指令不够强。3. LLM 温度参数过高。1. 检查检索结果 (source_documents)看 top_k 是否合理考虑优化分割策略或使用重排序。2. 强化提示词如“必须严格依据上下文回答”。3. 将 LLM 的temperature设为 0 或更低值。检索不到任何相关内容1. 向量数据库为空或未正确加载。2. 查询问题与文档语义差异太大。3. 嵌入模型不匹配重建后未更新。1. 检查向量库文档数量确认加载路径和集合名称正确。2. 尝试用更口语化或更具体的方式提问或实施查询扩展。3. 确认查询时使用的嵌入模型与建库时一致。Agent 陷入循环或调用错误工具1. 工具描述不清晰。2. Agent 提示词不适合当前任务。3.max_iterations设置过小或过大。1. 优化工具的描述 (description)使其功能边界更明确。2. 尝试不同的提示词如“zero-shot-react-description”。3. 观察verbose日志调整max_iterations并设置early_stopping_method。响应速度慢1. 网络延迟调用远程 API。2. 检索的k值过大或未使用索引。3. LLM 生成速度慢。1. 考虑将模型或嵌入服务部署在离用户更近的区域。2. 减少k值确保向量数据库有合适索引。3. 对于简单问题可换用更快的模型如gpt-3.5-turbo而非gpt-4。处理长文档时答案不完整1. 文本块分割过大超出 LLM 上下文窗口。2. 使用的chain_type如stuff无法处理过长上下文。1. 减小chunk_size确保单个块加上提示词后不超过模型限制。2. 换用map_reduce或refine链类型来处理超长文档。“上下文理解”和“语境推测”未被统一检索这是语义搜索的固有问题嵌入模型认为它们是不同概念。1. 在索引前对文本进行关键词归一化或同义词扩展。2. 采用混合检索结合关键词匹配来弥补语义搜索的不足。3. 使用更擅长理解同义词的嵌入模型或针对领域数据微调嵌入模型。构建企业级 AI 知识库是一个迭代过程从最简单的 RAG 管道开始逐步引入更优的检索策略、更复杂的 Agent 逻辑并最终用工程化手段保障其稳定、可控和可维护。本文提供的代码和思路是一个坚实的起点你可以根据具体的业务数据、性能要求和合规需求进行调整与深化。下一步可以探索更高级的 Agent 框架如 CrewAI、AutoGen、对领域数据进行嵌入模型微调、或者实现复杂多模态 RAG将图片、表格等内容也纳入知识体系。