企业官网建设流程全解析

在实际企业级 AI 应用开发中，如何将私有知识、文档数据与大语言模型（LLM）的能力安全、高效地结合，是一个核心挑战。直接让 LLM 处理长文档或回答专业问题，往往会遇到“幻觉”（生成不准确信息）、上下文长度限制和知识更新滞后等问题。检索增强生成（RAG）技术正是为解决这些问题而生，它通过“检索”相关文档片段来“增强”LLM的生成过程，确保回答基于可信来源。而 LangChain 作为当前最流行的 LLM 应用开发框架，提供了构建 RAG 系统所需的模块化组件和抽象，让开发者能更专注于业务逻辑而非底层连接。

本文将围绕 LangChain 与 RAG 技术栈，从零开始构建一个企业级 AI 知识库的原型。我们将不局限于简单的问答，而是深入向量数据库选型、检索策略优化，并最终引入 Agent 智能体，让系统具备自主调用工具、执行多步骤复杂任务的能力。整个过程会涉及环境搭建、代码实现、核心参数调优以及生产环境下的常见问题排查，目标是交付一个可学习、可复现、可扩展的技术方案。

1. 理解 RAG 与 LangChain 的核心工作机制

在动手写代码之前，必须厘清几个核心概念以及它们是如何协同工作的。这能帮助你在后续配置出错或效果不佳时，快速定位问题所在。

1.1 RAG 系统的基本流程：检索、增强、生成

一个标准的 RAG 流程可以拆解为索引（Indexing）和查询（Querying）两个阶段。

索引阶段的目标是将原始的非结构化文档（如 PDF、Word、Markdown）转化为便于检索的格式。这个过程通常包括：

1. 加载（Loading）：使用文档加载器（Document Loader）从文件系统、数据库或网络读取文档内容。
2. 分割（Splitting）：由于 LLM 有上下文窗口限制，必须将长文档切分成语义连贯的“块”（Chunks）。分割策略（如按字符、按句子、按段落）和块大小、重叠区（Overlap）的设置至关重要，直接影响检索质量。
3. 嵌入（Embedding）：使用嵌入模型（Embedding Model）将每个文本块转换为一个高维向量（Vector）。这个向量在数学上表征了文本的语义信息，语义相近的文本，其向量在空间中的距离也更近。
4. 存储（Storing）：将文本块及其对应的向量存储到向量数据库（Vector Database）中。向量数据库专门为高效的海量向量相似性搜索而设计。

查询阶段是用户与系统交互的过程：

1. 问题嵌入：将用户提出的问题（Query）同样通过嵌入模型转换为向量。
2. 相似性检索：在向量数据库中，搜索与问题向量最相似的 K 个文本块向量。这个过程就是“检索”，其核心是找到与问题语义最相关的知识片段。
3. 上下文构建：将检索到的 K 个文本块作为“上下文”（Context），与原始问题一起，按照特定的提示词（Prompt）模板进行组装，形成最终的输入。
4. 生成回答：将组装好的提示词发送给 LLM（如 GPT、Claude 或本地部署的模型），LLM 基于给定的上下文生成最终答案。这就是“增强生成”，它强制 LLM 在提供的材料中寻找答案，大幅减少幻觉。

1.2 LangChain 的模块化设计：用“链”组装复杂应用

LangChain 不是一个黑盒应用，而是一个工具箱。它将 RAG 流程中的每个步骤抽象成了独立的模块，并通过“链”（Chain）的概念将它们串联起来。理解这些核心模块是灵活运用 LangChain 的关键：

• 文档加载器（Document Loaders）：DirectoryLoader,PyPDFLoader,UnstructuredFileLoader等，用于从各种来源加载文档。
• 文本分割器（Text Splitters）：RecursiveCharacterTextSplitter,CharacterTextSplitter等，负责将文档切块。
• 嵌入模型（Embedding Models）：OpenAIEmbeddings,HuggingFaceEmbeddings等，提供文本到向量的转换能力。这里需要区分“嵌入模型”和后续的“LLM”，它们通常是两个独立的模型。
• 向量存储（Vectorstores）：Chroma,FAISS,Pinecone（云服务）,Milvus等，是向量数据库在 LangChain 中的接口封装。
• 大语言模型（LLMs/Chat Models）：ChatOpenAI,ChatAnthropic,Ollama（本地）等，负责最终的文本生成。
• 链（Chains）：RetrievalQA,ConversationalRetrievalChain等，是预组装好的、针对特定任务（如问答）的工作流。你也可以用LCEL（LangChain Expression Language）自定义链。
• 智能体（Agents）：create_react_agent,create_openai_tools_agent等，是能理解目标、自主选择并调用工具（如搜索、计算、数据库查询）的 LLM 增强系统。Agent 让 LLM 从“答题者”变为“执行者”。

LangChain vs. LangGraph：LangGraph 是 LangChain 的一个扩展库，用于构建有状态的、多步骤的复杂工作流。如果说 LangChain 的 Chain 是线性管道，那么 LangGraph 允许你构建带循环、分支和并行执行的有向图。对于简单的 RAG，LangChain 足够；对于需要多轮规划、自我修正的复杂 Agent，LangGraph 更合适。

2. 环境准备与核心依赖配置

我们将构建一个基于本地文件的 RAG 系统，并逐步升级到 Agent。为了兼顾学习成本和实用性，技术栈选择如下：

• 编程语言：Python 3.10+
• 核心框架：LangChain
• 向量数据库：Chroma（轻量，易于本地部署和实验）
• 嵌入模型：text-embedding-3-small（OpenAI API）或BAAI/bge-small-zh-v1.5（Hugging 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/activate

2.2 安装依赖包

创建requirements.txt文件，内容如下。注意，我们根据功能模块对依赖进行了分组注释。

# 核心框架 langchain==0.1.0 langchain-community==0.0.10 # 社区维护的集成组件 langchain-openai==0.0.5 # OpenAI 官方集成 langchain-chroma==0.1.0 # Chroma 向量数据库集成 # 向量数据库客户端 chromadb==0.4.22 # 文档加载与处理 pypdf==3.17.4 # 用于读取PDF unstructured[pdf,docx]==0.10.30 # 强大的文档解析库 markdown==3.5.1 # 嵌入模型 (可选，如果使用 HuggingFace 模型) sentence-transformers==2.2.2 # 网络请求与API调用 httpx==0.25.1 openai==1.6.1 # 开发与工具 (用于Agent示例) langchain-experimental==0.0.49 # 包含一些实验性功能，如高级Agent python-dotenv==1.0.0 # 管理环境变量

使用 pip 安装所有依赖：

pip install -r requirements.txt

2.3 配置 API 密钥与环境变量

为了安全地管理敏感信息（如 OpenAI API Key），务必使用环境变量。在项目根目录创建.env文件：

# .env 文件 OPENAI_API_KEY=sk-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_size=chunk_size, chunk_overlap=chunk_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, glob=glob_pattern, loader_cls=PyPDFLoader) elif glob_pattern.endswith(".txt"): loader = DirectoryLoader(directory_path, glob=glob_pattern, loader_cls=TextLoader) else: # 使用 Unstructured 加载器处理多种格式 from langchain_community.document_loaders import UnstructuredFileLoader loader = DirectoryLoader(directory_path, glob=glob_pattern, loader_cls=UnstructuredFileLoader) 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_size=800, chunk_overlap=150) 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(model=embedding_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( documents=documents, embedding=self.embeddings, persist_directory=self.persist_directory, collection_name=collection_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_directory=self.persist_directory, embedding_function=self.embeddings, collection_name=collection_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_type=search_type, search_kwargs=search_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_name="company_docs")

关键点解析：

1. 嵌入模型选择：OpenAIEmbeddings需要有效的OPENAI_API_KEY。对于纯中文场景或内网环境，可以考虑HuggingFaceEmbeddings，例如model_name=“BAAI/bge-small-zh-v1.5”。更换模型时，整个向量库需要重建，因为不同模型生成的向量空间不同。
2. 检索器（Retriever）：这是 LangChain 中一个核心抽象，它封装了从向量库中查找相关文档的逻辑。search_type和search_kwargs是调优检索效果的重要杠杆。
3. 持久化：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(model=model_name, temperature=0) # temperature=0 使输出更确定 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_type="similarity", search_kwargs={"k": 4}) # 2. 定义提示词模板 if prompt_template is None: prompt_template = """请根据以下上下文信息回答问题。如果你不知道答案，就诚实地回答不知道，不要编造信息。 上下文： {context} 问题：{question} 基于上下文的答案：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 3. 构建链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", # 最常用的类型，将所有检索到的上下文塞入提示词 retriever=retriever, chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True # 返回源文档，便于追溯和调试 ) 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"[来源 {i+1}] {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_name="company_docs") # 2. 创建问答机器人并提问 bot = QABot(vs_manager, model_name="gpt-3.5-turbo") bot.build_qa_chain() while True: user_question = input("\n请输入您的问题（输入 'quit' 退出）: ") if user_question.lower() == 'quit': break bot.ask(user_question)

代码逻辑与调优点：

1. 链类型（chain_type）：我们使用了“stuff”，它简单地将所有检索到的上下文拼接后传给 LLM。对于非常多的上下文，可能会超出令牌限制。其他选项包括“map_reduce”（分别总结再汇总）、“refine”（迭代优化答案）和“map_rerank”，它们适用于处理大量文档，但更复杂且可能损失精度。
2. 提示词工程：模板中的指令“如果你不知道答案，就诚实地回答不知道”对于减少幻觉非常关键。你可以根据业务需求定制模板，例如要求答案以特定格式列出，或引用源文档的页码。
3. 返回源文档：return_source_documents=True是调试和建立用户信任的必备功能。你可以展示答案的依据，让用户判断可信度。

运行此脚本，一个基础的、基于私有文档的知识问答系统就构建完成了。你可以向docs/目录添加新的文档，然后重建向量库来更新知识。

4. 进阶：优化检索效果与引入 Agent 智能体

基础 RAG 可能面临检索不准、答案冗长或无法处理复杂问题等挑战。下面我们进行两方面的增强：优化检索策略和引入智能体。

4.1 优化检索策略：超越简单相似度搜索

简单的向量相似度搜索有时会返回相关但不精确的片段。以下是几种优化方案：

方案一：使用 MMR（最大边际相关性）检索MMR 在保证相关性的同时，增加结果集的多样性，避免返回多个高度重复的片段。

# 在 VectorStoreManager.get_retriever 中调整参数 retriever = self.vector_store.as_retriever( search_type="mmr", # 改为 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）先召回较多的候选文档（如 k=20），再用一个更精细的交叉编码器模型（Cross-Encoder）对它们进行重排序，选出最相关的几个（如 top_k=4）。这能显著提升精度，但会增加延迟和计算成本。

# 伪代码示例：使用 Cohere 或 BGE 的重排序器 # from langchain.retrievers import ContextualCompressionRetriever # from langchain.retrievers.document_compressors import CrossEncoderReranker # compressor = CrossEncoderReranker(model="BAAI/bge-reranker-large", top_n=4) # compression_retriever = ContextualCompressionRetriever(base_retriever=retriever, base_compressor=compressor)

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_ascii=False) return Tool( name="Company_Knowledge_Base", func=knowledge_base_lookup, description="当需要回答关于公司产品、政策、流程或历史文档的具体问题时使用此工具。输入必须是一个完整的问题。" ) def setup_calculator_tool(llm): """数学计算工具""" math_chain = LLMMathChain.from_llm(llm=llm, verbose=False) return Tool( name="Calculator", func=math_chain.run, description="用于执行数学计算。输入应是一个数学表达式，例如 '2 + 2' 或 'sin(45度)'。" ) def setup_wikipedia_tool(): """维基百科搜索工具（示例，需要网络）""" wikipedia = WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=500) return Tool( name="Wikipedia", func=wikipedia.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(model="gpt-3.5-turbo", temperature=0) 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( agent=agent, tools=tools, verbose=True, # 打印 Agent 的思考过程，便于调试 handle_parsing_errors=True, # 优雅处理解析错误 max_iterations=5 # 防止无限循环 ) 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(f"Agent 执行出错: {e}")

运行这个 Agent，你会看到verbose=True模式下输出的详细思考过程（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 部署示例：

# 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_code=503, detail="服务正在初始化，请稍后重试。") try: result = agent.invoke({"input": request.question}) return {"answer": result["output"]} except Exception as e: raise HTTPException(status_code=500, detail=f"处理问题时出错: {str(e)}") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

5.2 关键生产环境考量

1. 配置外置化：所有 API Key、模型名称、数据库连接字符串、超时参数等必须通过环境变量或配置中心管理，绝不能硬编码。
2. 异步处理：LLM 调用和向量检索可能是 I/O 密集型操作。使用asyncio和异步 HTTP 客户端（如httpx）来避免阻塞，提高并发能力。
3. 超时与重试：为所有外部服务调用（OpenAI API、向量数据库）设置合理的超时和重试机制。
4. 限流与熔断：防止滥用或下游服务故障导致系统雪崩。可以使用tenacity进行重试，使用circuitbreaker实现熔断。
5. 日志与监控：
   • 结构化日志：记录每个请求的问题、答案、检索到的源文档 ID、Token 使用量、耗时和任何错误。这对于调试和成本分析至关重要。
   • 应用性能监控：集成如 Prometheus + Grafana，监控请求延迟、错误率、Token 消耗等指标。
   • 链路追踪：在微服务架构下，使用 OpenTelemetry 追踪一个请求经过 RAG、LLM、Agent 等组件的完整路径。
6. 成本控制：
   • 缓存频繁查询的问题-答案对。
   • 监控并分析 Token 使用情况，优化提示词长度。
   • 对于内部知识，考虑使用更小的、领域微调过的开源模型（通过 Ollama、vLLM 等部署），以降低 API 成本。

5.3 常见问题排查清单

当你的 RAG 或 Agent 系统出现问题时，可以按照以下清单逐项排查：

构建企业级 AI 知识库是一个迭代过程，从最简单的 RAG 管道开始，逐步引入更优的检索策略、更复杂的 Agent 逻辑，并最终用工程化手段保障其稳定、可控和可维护。本文提供的代码和思路是一个坚实的起点，你可以根据具体的业务数据、性能要求和合规需求进行调整与深化。下一步，可以探索更高级的 Agent 框架（如 CrewAI、AutoGen）、对领域数据进行嵌入模型微调、或者实现复杂多模态 RAG，将图片、表格等内容也纳入知识体系。