首页/资讯中心/详情

LangChain 1.3.11与LangGraph：从RAG系统搭建到多智能体工作流实战

LangChain 1.3.11与LangGraph：从RAG系统搭建到多智能体工作流实战
📅 发布时间：2026/7/30 12:52:34

当前B站最全最细的 LangchainV1.3 应用开发全套教程！从LangChain构建RAG应用到LangGraph构建多智能体工作流！少走99%的弯路！

最近在开发大模型应用时，很多同学反馈 LangChain 生态版本更新快、组件复杂，从 RAG 系统搭建到多智能体工作流实现，网上资料零散不成体系。本文基于 LangChain 1.3.11 最新版本，完整拆解从基础概念到项目实战的全流程，包含环境配置、核心组件详解、RAG 系统构建、LangGraph 多智能体开发等关键环节，每个步骤都提供可运行代码示例和避坑指南。

无论你是刚接触 LangChain 的新手，还是希望深入多智能体开发的进阶开发者，都能从本文获得实用价值。学完后你将掌握：LangChain 1.3.x 核心组件的正确用法、RAG 知识库的完整构建流程、LangGraph 多智能体工作流设计，以及企业级应用的最佳实践方案。

1. LangChain 与 RAG 技术核心概念解析

1.1 什么是 LangChain 及其技术演进

LangChain 是一个用于开发大语言模型（LLM）应用程序的框架，它提供了一套标准化的接口和组件，帮助开发者更高效地构建基于 LLM 的应用。从最初的 0.1.x 版本到现在的 1.3.x 版本，LangChain 经历了从简单的链式调用到复杂的代理系统的演进。

在 1.3.x 版本中，LangChain 的核心架构更加模块化，主要包含以下几个关键组件：

  • Models：支持多种 LLM 提供商（OpenAI、Anthropic、本地模型等）
  • Prompts：模板化管理提示词，支持动态变量注入
  • Chains：将多个组件组合成可复用的工作流
  • Agents：让 LLM 决定执行哪些工具的高级推理系统
  • Memory：在不同对话间保持状态记忆
  • Indexes：文档加载、文本分割、向量存储等检索增强功能

1.2 RAG 技术原理与核心价值

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将检索系统与生成模型相结合的技术架构。其核心思想是：在生成答案前，先从知识库中检索相关信息，然后将检索到的信息与用户问题一起提供给 LLM，从而生成更准确、更具事实依据的答案。

传统 LLM 的局限性在于其知识受限于训练数据，且无法访问最新或专有信息。RAG 通过以下方式解决这些问题：

  1. 知识实时性：可以随时更新向量数据库中的知识，让模型获取最新信息
  2. 事实准确性：基于检索到的文档生成答案，减少模型幻觉
  3. 成本效益：不需要重新训练大模型即可扩展知识库
  4. 可解释性：可以追溯答案来源，增强可信度

RAG 系统的基本工作流程包括：文档加载 → 文本分割 → 向量化 → 存储 → 检索 → 生成答案。每个环节都有不同的技术选型和优化策略，我们将在实战部分详细展开。

1.3 LangGraph 与多智能体系统介绍

LangGraph 是 LangChain 生态中用于构建多智能体工作流的库，它基于状态机概念，可以创建复杂的、有状态的对话系统。与传统的链式调用不同，LangGraph 允许智能体根据当前状态决定下一步行动，支持循环、条件分支等复杂逻辑。

多智能体系统的核心优势在于：

  • 分工协作：不同智能体专注于特定任务，提高整体效率
  • 错误恢复：一个智能体失败时，其他智能体可以接管或协助
  • 复杂任务分解：将复杂问题拆解为子任务，由专门智能体处理
  • 状态持久化：在整个对话过程中保持上下文和状态

LangGraph 使用节点（Nodes）和边（Edges）来定义工作流，每个节点代表一个处理步骤，边定义状态转换条件。这种设计模式特别适合需要多轮对话、工具调用和复杂推理的应用场景。

2. 环境准备与版本兼容性配置

2.1 基础环境要求

在开始 LangChain 1.3.x 开发前，需要确保开发环境满足以下要求：

  • Python 版本：3.8 或更高版本（推荐 3.9+）
  • 操作系统：Windows、Linux 或 macOS（本文示例基于 Ubuntu 22.04）
  • 内存要求：至少 8GB RAM，处理大型文档时推荐 16GB+
  • 网络环境：能够访问 Hugging Face、OpenAI 等模型服务

建议使用虚拟环境管理 Python 依赖，避免版本冲突：

# 创建虚拟环境 python -m venv langchain_env source langchain_env/bin/activate # Linux/macOS # 或 langchain_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip

2.2 LangChain 1.3.11 核心依赖安装

LangChain 1.3.x 版本采用了模块化架构，需要根据具体需求安装相应的子包。以下是核心依赖的安装命令：

# 安装 LangChain 核心库 pip install langchain==1.3.11 # 安装社区贡献的组件 pip install langchain-community==0.3.6 # 安装文本处理相关工具 pip install langchain-text-splitters==0.3.5 # 安装实验性功能（包含 LangGraph） pip install langchain-experimental==0.1.12 # 安装 LangGraph 用于多智能体工作流 pip install langgraph==0.2.42 # 安装 OpenAI 接口（如果使用 OpenAI 模型） pip install openai==1.62.0 # 安装向量数据库客户端（以 Chroma 为例） pip install chromadb==0.5.0 # 安装文档处理工具 pip install pypdf==4.2.0 python-docx==1.1.2

重要版本兼容性说明：LangChain 1.3.11 与 langchain-community 0.3.6 版本兼容性最佳。如果遇到导入错误，请检查各包版本是否匹配。

2.3 模型服务配置

根据使用的 LLM 服务商，需要配置相应的 API 密钥：

# 配置环境变量（推荐方式） import os os.environ["OPENAI_API_KEY"] = "your-openai-api-key" os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key" # 或者使用本地模型（如 Ollama） os.environ["OLLAMA_BASE_URL"] = "http://localhost:11434"

对于国内开发者，如果需要使用代理访问国际模型服务，建议在系统层面配置网络代理，而不是在代码中直接设置，以避免安全风险。

3. LangChain 1.3.x 核心组件详解

3.1 模型调用与统一接口

LangChain 提供了统一的接口来调用不同的 LLM 服务，这使得切换模型提供商变得非常简单。以下是如何使用不同模型的示例：

from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain_community.llms import Ollama # 使用 OpenAI GPT-4 openai_llm = ChatOpenAI( model="gpt-4", temperature=0.7, max_tokens=1000 ) # 使用 Anthropic Claude anthropic_llm = ChatAnthropic( model="claude-3-sonnet-20240229", temperature=0.7, max_tokens=1000 ) # 使用本地 Ollama 模型 local_llm = Ollama(model="llama2") # 统一的调用方式 response = openai_llm.invoke("请介绍一下 LangChain") print(response.content)

LangChain 的模型接口设计遵循了依赖倒置原则，无论底层使用什么模型，上层的调用方式基本一致，这大大提高了代码的可维护性和可扩展性。

3.2 提示词模板与动态变量

提示词模板是 LangChain 的核心功能之一，它帮助开发者管理复杂的提示词，支持动态变量注入：

from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate # 创建复杂的提示词模板 system_template = """你是一个专业的{role}助手。请根据以下上下文回答问题。 上下文： {context} 问题：{question} 要求： - 回答要专业、准确 - 如果上下文不足以回答问题，请明确说明 - 使用{language}语言回答 """ prompt_template = ChatPromptTemplate.from_messages([ ("system", system_template), HumanMessagePromptTemplate.from_template("{human_input}") ]) # 使用模板生成最终提示词 formatted_prompt = prompt_template.format_messages( role="技术", context="LangChain 是一个 LLM 应用开发框架", question="LangChain 的主要功能是什么？", language="中文", human_input="请详细解释一下" ) response = openai_llm.invoke(formatted_prompt) print(response.content)

提示词模板支持条件逻辑、部分填充等高级功能，适合构建复杂的多轮对话系统。

3.3 文档加载与文本处理

LangChain 提供了丰富的文档加载器，支持从各种来源加载文档：

from langchain_community.document_loaders import PyPDFLoader, TextLoader, WebBaseLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 加载 PDF 文档 pdf_loader = PyPDFLoader("example.pdf") pdf_documents = pdf_loader.load() # 加载网页内容 web_loader = WebBaseLoader(["https://example.com"]) web_documents = web_loader.load() # 文本分割配置 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "] ) # 分割文档 split_documents = text_splitter.split_documents(pdf_documents) print(f"原始文档数：{len(pdf_documents)}") print(f"分割后文档块数：{len(split_documents)}")

文本分割是 RAG 系统的关键步骤，合适的分块策略能显著影响检索效果。建议根据文档类型和内容特点调整分块大小和重叠长度。

3.4 向量化与相似度检索

向量化是将文本转换为数值向量的过程，LangChain 支持多种嵌入模型和向量数据库：

from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 初始化嵌入模型 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 创建向量数据库 vectorstore = Chroma.from_documents( documents=split_documents, embedding=embeddings, persist_directory="./chroma_db" ) # 相似度检索 query = "LangChain 的主要功能是什么？" similar_docs = vectorstore.similarity_search(query, k=3) print("检索到的相关文档：") for i, doc in enumerate(similar_docs): print(f"{i+1}. {doc.page_content[:200]}...")

在实际项目中，需要根据数据规模和性能要求选择合适的向量数据库。Chroma 适合中小规模项目，Milvus、Pinecone 等适合大规模生产环境。

4. 构建完整的 RAG 知识库系统

4.1 RAG 系统架构设计

一个完整的 RAG 系统包含以下核心模块：

  1. 文档处理流水线：文档加载 → 文本清洗 → 分块 → 向量化
  2. 检索模块：向量相似度检索 + 可选的关键词检索（混合检索）
  3. 重排序模块：对检索结果进行相关性重排序
  4. 生成模块：基于检索内容生成答案
  5. 评估模块：监控系统效果和质量

以下是完整的 RAG 系统实现：

from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor class AdvancedRAGSystem: def __init__(self, vectorstore, llm): self.vectorstore = vectorstore self.llm = llm self.setup_retrieval_chain() def setup_retrieval_chain(self): # 基础检索器 base_retriever = self.vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 5} ) # 添加结果压缩（提高相关性） compressor = LLMChainExtractor.from_llm(self.llm) self.compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=base_retriever ) # 自定义提示词模板 custom_prompt = PromptTemplate( template="""基于以下上下文信息，请回答问题。如果上下文不足以回答问题，请说明你不知道。 上下文： {context} 问题：{question} 请提供详细、准确的回答：""", input_variables=["context", "question"] ) # 创建检索问答链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.compression_retriever, chain_type_kwargs={"prompt": custom_prompt}, return_source_documents=True ) def query(self, question): result = self.qa_chain.invoke({"query": question}) return result # 使用示例 rag_system = AdvancedRAGSystem(vectorstore, openai_llm) result = rag_system.query("LangChain 如何帮助构建 RAG 系统？") print("答案：", result["result"]) print("来源文档：", [doc.metadata.get('source', '未知') for doc in result["source_documents"]])

4.2 高级检索策略优化

基础相似度检索在某些场景下效果有限，我们可以实现更高级的检索策略：

from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.retrievers.merger_retriever import MergerRetriever from langchain_community.vectorstores import FAISS class HybridRetrievalSystem: def __init__(self, documents, embeddings): self.documents = documents self.embeddings = embeddings self.setup_hybrid_retriever() def setup_hybrid_retriever(self): # 1. 向量检索器 vectorstore = FAISS.from_documents(self.documents, self.embeddings) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 2. 关键词检索器（BM25） from langchain.retrievers import BM25Retriever bm25_retriever = BM25Retriever.from_documents(self.documents) bm25_retriever.k = 3 # 3. 混合检索器 self.ensemble_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.5, 0.5] ) def search(self, query): return self.ensemble_retriever.get_relevant_documents(query) # 使用混合检索 hybrid_system = HybridRetrievalSystem(split_documents, embeddings) results = hybrid_system.search("LangChain 向量数据库") print(f"混合检索结果数量：{len(results)}")

混合检索结合了语义相似度和关键词匹配的优势，能显著提高检索的召回率和准确率。

4.3 RAG 系统评估与监控

构建 RAG 系统后，需要建立评估机制来监控系统效果：

class RAGEvaluator: def __init__(self, rag_system): self.rag_system = rag_system def evaluate_answer_relevance(self, question, ground_truth): """评估答案相关性""" result = self.rag_system.query(question) generated_answer = result["result"] # 使用 LLM 评估相关性（实际项目中可使用更精确的评估方法） evaluation_prompt = f""" 请评估生成的答案与标准答案的相关性。 问题：{question} 标准答案：{ground_truth} 生成答案：{generated_answer} 请给出相关性评分（1-5分）和简要理由： """ evaluation = openai_llm.invoke(evaluation_prompt) return evaluation.content def evaluate_retrieval_quality(self, question, expected_docs_count=3): """评估检索质量""" result = self.rag_system.query(question) retrieved_docs = result["source_documents"] metrics = { "retrieved_count": len(retrieved_docs), "expected_count": expected_docs_count, "coverage_ratio": len(retrieved_docs) / expected_docs_count if expected_docs_count > 0 else 0 } return metrics # 评估示例 evaluator = RAGEvaluator(rag_system) relevance_score = evaluator.evaluate_answer_relevance( "什么是 LangChain？", "LangChain 是一个用于开发大语言模型应用的框架" ) retrieval_metrics = evaluator.evaluate_retrieval_quality("什么是 LangChain？") print("相关性评估：", relevance_score) print("检索质量指标：", retrieval_metrics)

5. LangGraph 多智能体工作流实战

5.1 LangGraph 基础概念与架构

LangGraph 使用图结构来定义多智能体工作流，核心概念包括：

  • State：工作流的共享状态，在不同节点间传递信息
  • Nodes：处理节点，执行特定任务
  • Edges：边，定义状态转换条件
  • Conditions：条件判断，决定下一步执行哪个节点

以下是一个简单的 LangGraph 工作流示例：

from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator # 定义状态结构 class AgentState(TypedDict): question: str context: str answer: str needs_clarification: bool clarification_question: str # 创建图结构 graph_builder = StateGraph(AgentState) # 定义节点函数 def retrieve_context(state: AgentState): """检索相关上下文""" question = state["question"] # 模拟检索过程 retrieved_context = f"关于'{question}'的相关信息：LangGraph 是用于构建多智能体工作流的库。" return {"context": retrieved_context} def generate_answer(state: AgentState): """生成答案""" question = state["question"] context = state["context"] # 使用 LLM 生成答案 prompt = f"基于以下上下文回答问题：\n上下文：{context}\n问题：{question}\n答案：" answer = openai_llm.invoke(prompt).content return {"answer": answer} def check_clarification(state: AgentState): """检查是否需要澄清问题""" question = state["question"] answer = state["answer"] # 简单逻辑：如果答案包含"不确定"或"需要更多信息"，则要求澄清 needs_clarification = any(word in answer for word in ["不确定", "需要更多信息", "不清楚"]) clarification_question = "请提供更多细节信息" if needs_clarification else "" return { "needs_clarification": needs_clarification, "clarification_question": clarification_question } # 添加节点 graph_builder.add_node("retrieve", retrieve_context) graph_builder.add_node("generate", generate_answer) graph_builder.add_node("clarify", check_clarification) # 设置边和条件 graph_builder.set_entry_point("retrieve") graph_builder.add_edge("retrieve", "generate") graph_builder.add_conditional_edges( "generate", lambda state: "clarify" if state["needs_clarification"] else END, {"clarify": "clarify", END: END} ) graph_builder.add_edge("clarify", "generate") # 编译图 graph = graph_builder.compile()

5.2 复杂多智能体系统实现

在实际项目中，我们需要更复杂的多智能体协作系统。以下是一个包含专业分工的多智能体示例：

from langgraph.graph import StateGraph, END from typing import TypedDict, List import json class MultiAgentState(TypedDict): user_query: str research_results: List[str] technical_analysis: str business_implications: str final_answer: str current_step: str class ResearchAgent: def __init__(self, llm): self.llm = llm def research(self, query: str) -> List[str]: """研究代理：搜集相关信息""" prompt = f"""请为以下问题搜集相关信息和数据： 问题：{query} 请提供3-5个关键信息点，每个信息点用一句话描述：""" response = self.llm.invoke(prompt) # 解析响应，提取信息点 points = [point.strip() for point in response.content.split('\n') if point.strip()] return points[:5] # 返回前5个信息点 class TechnicalAnalystAgent: def __init__(self, llm): self.llm = llm def analyze(self, query: str, research_data: List[str]) -> str: """技术分析代理：进行技术评估""" research_context = "\n".join([f"- {point}" for point in research_data]) prompt = f"""基于以下研究信息，进行技术分析： 用户问题：{query} 研究信息： {research_context} 请从技术可行性、实现复杂度、技术风险等方面进行分析：""" response = self.llm.invoke(prompt) return response.content class BusinessAnalystAgent: def __init__(self, llm): self.llm = llm def evaluate(self, query: str, tech_analysis: str) -> str: """商业分析代理：评估商业影响""" prompt = f"""基于技术分析，评估商业影响： 用户问题：{query} 技术分析：{tech_analysis} 请从成本效益、市场机会、竞争优势等方面进行评估：""" response = self.llm.invoke(prompt) return response.content class CoordinatorAgent: def __init__(self, llm): self.llm = llm def synthesize(self, query: str, research: List[str], tech_analysis: str, business_eval: str) -> str: """协调代理：整合各方分析，生成最终答案""" prompt = f"""整合以下所有分析，生成最终回答： 原始问题：{query} 研究结果： {chr(10).join([f"- {point}" for point in research])} 技术分析： {tech_analysis} 商业评估： {business_eval} 请生成一个全面、专业的最终回答：""" response = self.llm.invoke(prompt) return response.content def create_multi_agent_system(llm): """创建多智能体系统""" # 初始化各专业代理 research_agent = ResearchAgent(llm) tech_agent = TechnicalAnalystAgent(llm) business_agent = BusinessAnalystAgent(llm) coordinator_agent = CoordinatorAgent(llm) # 创建图构建器 graph_builder = StateGraph(MultiAgentState) # 定义节点函数 def research_node(state: MultiAgentState): research_results = research_agent.research(state["user_query"]) return { "research_results": research_results, "current_step": "research_completed" } def technical_analysis_node(state: MultiAgentState): analysis = tech_agent.analyze(state["user_query"], state["research_results"]) return { "technical_analysis": analysis, "current_step": "technical_analysis_completed" } def business_analysis_node(state: MultiAgentState): evaluation = business_agent.evaluate(state["user_query"], state["technical_analysis"]) return { "business_implications": evaluation, "current_step": "business_analysis_completed" } def coordination_node(state: MultiAgentState): final_answer = coordinator_agent.synthesize( state["user_query"], state["research_results"], state["technical_analysis"], state["business_implications"] ) return { "final_answer": final_answer, "current_step": "completed" } # 添加节点 graph_builder.add_node("research", research_node) graph_builder.add_node("technical_analysis", technical_analysis_node) graph_builder.add_node("business_analysis", business_analysis_node) graph_builder.add_node("coordination", coordination_node) # 设置工作流 graph_builder.set_entry_point("research") graph_builder.add_edge("research", "technical_analysis") graph_builder.add_edge("technical_analysis", "business_analysis") graph_builder.add_edge("business_analysis", "coordination") graph_builder.add_edge("coordination", END) return graph_builder.compile() # 使用多智能体系统 multi_agent_graph = create_multi_agent_system(openai_llm) # 执行查询 initial_state = {"user_query": "如何评估在企业中实施 RAG 系统的可行性？"} result = multi_agent_graph.invoke(initial_state) print("最终答案：", result["final_answer"]) print("执行步骤：", result["current_step"])

5.3 智能体间的通信与状态管理

在多智能体系统中，智能体间的通信和状态管理至关重要。LangGraph 提供了灵活的状态管理机制：

from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any import uuid class CollaborativeState(TypedDict): conversation_id: str user_input: str agent_messages: Dict[str, List[Dict]] current_agent: str consensus_reached: bool final_decision: str def create_collaborative_agent_system(llm): """创建协作式多智能体系统""" graph_builder = StateGraph(CollaborativeState) # 定义不同领域的专业代理 agents = { "technical": "技术专家，负责技术可行性分析", "business": "商业分析师，负责商业价值评估", "security": "安全专家，负责安全风险评估", "ux": "用户体验设计师，负责用户体验评估" } def route_to_agents(state: CollaborativeState): """路由节点：决定哪些代理需要参与""" user_input = state["user_input"] # 基于输入内容决定需要哪些专业代理参与 required_agents = [] input_lower = user_input.lower() if any(word in input_lower for word in ["技术", "实现", "开发", "代码"]): required_agents.append("technical") if any(word in input_lower for word in ["商业", "成本", "收益", "市场"]): required_agents.append("business") if any(word in input_lower for word in ["安全", "风险", "隐私", "保护"]): required_agents.append("security") if any(word in input_lower for word in ["用户", "体验", "界面", "设计"]): required_agents.append("ux") # 默认至少有一个代理参与 if not required_agents: required_agents = ["technical"] return {"required_agents": required_agents} def agent_processing_node(state: CollaborativeState): """代理处理节点：各个代理并行处理""" user_input = state["user_input"] required_agents = state["required_agents"] agent_responses = {} for agent_name in required_agents: agent_role = agents[agent_name] prompt = f"""你是一个{agent_role}。请从你的专业角度分析以下问题： 问题：{user_input} 请提供专业的分析意见，包括：优势、风险、建议等：""" response = llm.invoke(prompt) agent_responses[agent_name] = { "role": agent_role, "analysis": response.content, "timestamp": str(uuid.uuid4()) } return {"agent_responses": agent_responses} def consensus_building_node(state: CollaborativeState): """共识构建节点：综合各代理意见""" agent_responses = state["agent_responses"] # 生成综合报告 analysis_summary = "各专业代理分析结果：\n\n" for agent_name, response in agent_responses.items(): analysis_summary += f"{response['role']}：\n{response['analysis']}\n\n" consensus_prompt = f"""基于以下各领域专家的分析，请生成综合建议： {analysis_summary} 原始问题：{state["user_input"]} 请整合各方意见，给出平衡各方面因素的最终建议：""" consensus_response = llm.invoke(consensus_prompt) return { "final_decision": consensus_response.content, "consensus_reached": True } # 添加节点（简化版本，实际需要更复杂的路由逻辑） graph_builder.add_node("route", route_to_agents) graph_builder.add_node("process", agent_processing_node) graph_builder.add_node("consensus", consensus_building_node) # 设置工作流 graph_builder.set_entry_point("route") graph_builder.add_edge("route", "process") graph_builder.add_edge("process", "consensus") graph_builder.add_edge("consensus", END) return graph_builder.compile() # 使用协作式多智能体系统 collaborative_system = create_collaborative_agent_system(openai_llm) initial_state = { "conversation_id": str(uuid.uuid4()), "user_input": "我们应该如何设计一个企业级的 RAG 系统？需要考虑哪些关键因素？" } result = collaborative_system.invoke(initial_state) print("综合建议：", result["final_decision"])

6. 企业级 RAG 系统最佳实践

6.1 性能优化策略

在生产环境中，RAG 系统的性能优化至关重要。以下是一些关键优化策略：

import time from functools import lru_cache from concurrent.futures import ThreadPoolExecutor import asyncio class OptimizedRAGSystem: def __init__(self, vectorstore, llm): self.vectorstore = vectorstore self.llm = llm self.setup_optimizations() def setup_optimizations(self): """设置各种优化措施""" # 缓存常用的检索结果 self.retrieval_cache = {} # 设置批量处理参数 self.batch_size = 5 # 性能监控 self.metrics = { "retrieval_time": [], "generation_time": [], "cache_hit_rate": 0 } @lru_cache(maxsize=1000) def cached_retrieval(self, query: str, k: int = 3): """带缓存的检索""" cache_key = f"{query}_{k}" if cache_key in self.retrieval_cache: self.metrics["cache_hit_rate"] += 1 return self.retrieval_cache[cache_key] start_time = time.time() results = self.vectorstore.similarity_search(query, k=k) retrieval_time = time.time() - start_time self.metrics["retrieval_time"].append(retrieval_time) self.retrieval_cache[cache_key] = results return results async def process_queries_batch(self, queries: List[str]): """批量处理查询""" results = [] # 并行处理检索 with ThreadPoolExecutor() as executor: retrieval_tasks = [] for query in queries: task = executor.submit(self.cached_retrieval, query) retrieval_tasks.append(task) # 等待所有检索完成 retrieval_results = [task.result() for task in retrieval_tasks] # 批量生成答案 generation_tasks = [] for i, query in enumerate(queries): context = "\n".join([doc.page_content for doc in retrieval_results[i]]) prompt = f"基于以下上下文回答问题：\n上下文：{context}\n问题：{query}\n答案：" generation_tasks.append(self.llm.ainvoke(prompt)) # 异步等待所有生成任务完成 generation_results = await asyncio.gather(*generation_tasks) for i, result in enumerate(generation_results): results.append({ "query": queries[i], "answer": result.content, "sources": [doc.metadata for doc in retrieval_results[i]] }) return results def get_performance_metrics(self): """获取性能指标""" avg_retrieval = sum(self.metrics["retrieval_time"]) / len(self.metrics["retrieval_time"]) if self.metrics["retrieval_time"] else 0 avg_generation = sum(self.metrics["generation_time"]) / len(self.metrics["generation_time"]) if self.metrics["generation_time"] else 0 return { "average_retrieval_time": avg_retrieval, "average_generation_time": avg_generation, "total_queries_processed": len(self.metrics["retrieval_time"]), "cache_hit_rate": self.metrics["cache_hit_rate"] / len(self.metrics["retrieval_time"]) if self.metrics["retrieval_time"] else 0 } # 使用优化后的系统 optimized_rag = OptimizedRAGSystem(vectorstore, openai_llm) # 批量处理示例 queries = [ "什么是 LangChain？", "RAG 系统如何工作？", "LangGraph 有什么优势？" ] # 异步处理 async def run_batch_processing(): results = await optimized_rag.process_queries_batch(queries) for result in results: print(f"问题：{result['query']}") print(f"答案：{result['answer'][:100]}...") print("---") # 运行批量处理 import asyncio asyncio.run(run_batch_processing()) # 查看性能指标 metrics = optimized_rag.get_performance_metrics() print("性能指标：", metrics)

6.2 安全与权限控制

企业级系统必须考虑安全性和权限控制：

class SecureRAGSystem: def __init__(self, rag_system, user_manager): self.rag_system = rag_system self.user_manager = user_manager self.audit_log = [] def check_permission(self, user_id: str, query: str) -> bool: """检查用户权限""" user_role = self.user_manager.get_user_role(user_id) query_lower = query.lower() # 定义敏感话题 sensitive_topics = ["机密", "内部", "财务",