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

当前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 技术原理与核心价值RAGRetrieval-Augmented Generation检索增强生成是一种将检索系统与生成模型相结合的技术架构。其核心思想是在生成答案前先从知识库中检索相关信息然后将检索到的信息与用户问题一起提供给 LLM从而生成更准确、更具事实依据的答案。传统 LLM 的局限性在于其知识受限于训练数据且无法访问最新或专有信息。RAG 通过以下方式解决这些问题知识实时性可以随时更新向量数据库中的知识让模型获取最新信息事实准确性基于检索到的文档生成答案减少模型幻觉成本效益不需要重新训练大模型即可扩展知识库可解释性可以追溯答案来源增强可信度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 pip2.2 LangChain 1.3.11 核心依赖安装LangChain 1.3.x 版本采用了模块化架构需要根据具体需求安装相应的子包。以下是核心依赖的安装命令# 安装 LangChain 核心库 pip install langchain1.3.11 # 安装社区贡献的组件 pip install langchain-community0.3.6 # 安装文本处理相关工具 pip install langchain-text-splitters0.3.5 # 安装实验性功能包含 LangGraph pip install langchain-experimental0.1.12 # 安装 LangGraph 用于多智能体工作流 pip install langgraph0.2.42 # 安装 OpenAI 接口如果使用 OpenAI 模型 pip install openai1.62.0 # 安装向量数据库客户端以 Chroma 为例 pip install chromadb0.5.0 # 安装文档处理工具 pip install pypdf4.2.0 python-docx1.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( modelgpt-4, temperature0.7, max_tokens1000 ) # 使用 Anthropic Claude anthropic_llm ChatAnthropic( modelclaude-3-sonnet-20240229, temperature0.7, max_tokens1000 ) # 使用本地 Ollama 模型 local_llm Ollama(modelllama2) # 统一的调用方式 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技术, contextLangChain 是一个 LLM 应用开发框架, questionLangChain 的主要功能是什么, 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_size1000, chunk_overlap200, length_functionlen, 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(modeltext-embedding-3-small) # 创建向量数据库 vectorstore Chroma.from_documents( documentssplit_documents, embeddingembeddings, persist_directory./chroma_db ) # 相似度检索 query LangChain 的主要功能是什么 similar_docs vectorstore.similarity_search(query, k3) print(检索到的相关文档) for i, doc in enumerate(similar_docs): print(f{i1}. {doc.page_content[:200]}...)在实际项目中需要根据数据规模和性能要求选择合适的向量数据库。Chroma 适合中小规模项目Milvus、Pinecone 等适合大规模生产环境。4. 构建完整的 RAG 知识库系统4.1 RAG 系统架构设计一个完整的 RAG 系统包含以下核心模块文档处理流水线文档加载 → 文本清洗 → 分块 → 向量化检索模块向量相似度检索 可选的关键词检索混合检索重排序模块对检索结果进行相关性重排序生成模块基于检索内容生成答案评估模块监控系统效果和质量以下是完整的 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_typesimilarity, search_kwargs{k: 5} ) # 添加结果压缩提高相关性 compressor LLMChainExtractor.from_llm(self.llm) self.compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) # 自定义提示词模板 custom_prompt PromptTemplate( template基于以下上下文信息请回答问题。如果上下文不足以回答问题请说明你不知道。 上下文 {context} 问题{question} 请提供详细、准确的回答, input_variables[context, question] ) # 创建检索问答链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverself.compression_retriever, chain_type_kwargs{prompt: custom_prompt}, return_source_documentsTrue ) 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_count3): 评估检索质量 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(maxsize1000) 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, kk) 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 [机密, 内部, 财务,