LangChain与LangGraph实战：从RAG知识库到多智能体工作流

在实际 AI 应用开发中直接调用大模型 API 往往只能解决简单问答。真正要把 LLM 用到企业知识库、复杂业务流程或多步骤决策场景就需要一套框架来组织提示词、管理上下文、连接工具和协调多个 AI 智能体。LangChain 和 LangGraph 正是为此而生但新手常被这两个库的关系、版本兼容和实际项目集成路径困扰。本文基于 LangChain 1.3.x 最新稳定版从零构建一个可运行的 RAG 知识库系统并在此基础上引入 LangGraph 实现多智能体协作流程。你会看到如何用 LangChain 组件拆分文档、生成向量、实现检索再用 LangGraph 的状态图和通道机制让多个 AI 智能体按预定规则交互。文中每个配置项、代码片段和排查步骤都经过生产环境验证帮你避开版本冲突、依赖缺失、配置错误和流程设计等常见坑。1. 理解 LangChain 与 LangGraph 的分工与协作1.1 LangChain 的核心价值标准化 LLM 应用组件LangChain 是一个框架它把 LLM 应用开发中常见的任务抽象成可复用的组件。这些组件包括文档加载器从 PDF、网页、数据库等来源加载非结构化文本。文本分割器把长文档切成适合模型处理的片段。向量化模型把文本转换成向量表示。向量数据库存储和检索向量。提示词模板管理动态提示词。链把多个组件串联成完整流程。智能体让 LLM 根据当前状态决定下一步动作。在 RAG 系统中LangChain 负责的是“原料准备”和“基础流水线”把文档处理成向量存到数据库用户提问时检索相关片段组装成提示词送给 LLM。这个流程是线性的每一步都有明确输入输出。1.2 LangGraph 的突破用图结构管理复杂状态流转LangGraph 建立在 LangChain 之上专门处理需要多步骤、有条件分支、有循环或需要多个 AI 智能体协作的场景。它的核心概念是节点一个执行单元可以是 LLM 调用、工具调用或自定义函数。边定义节点之间的流转条件。状态在整个图执行过程中共享的数据容器。通道状态中不同字段的更新规则。如果 LangChain 是组装线性流水线LangGraph 就是设计流程图你可以定义“先检查用户意图如果是查询知识库就走检索分支如果是计算就走工具分支如果需要多方确认就并行调用多个专家智能体”。这种带状态、带分支、带循环的流程是 LangChain 单纯用链难以优雅实现的。1.3 版本选择为什么是 LangChain 1.3.xLangChain 1.3.x 是一个重要的稳定版本它重构了部分内部接口更好地支持 LangGraph 集成。关键改进包括更清晰的依赖分离langchain-core、langchain-community等包让用户按需安装。改进的智能体执行器错误处理和状态管理更可靠。更好的异步支持适合高并发生产环境。配套版本建议截至发文时langchain1.3.11 langchain-community0.3.8 langgraph0.2.42 langchain-core0.3.7如果你用其他版本可能需要调整导入路径或参数名。始终在隔离环境中先验证版本兼容性。2. 环境准备与依赖配置2.1 创建隔离的 Python 环境避免包冲突是 LangChain 项目的第一课。推荐使用 conda 或 venv# 使用 conda conda create -n langchain-demo python3.11 conda activate langchain-demo # 或使用 venv python -m venv langchain-demo source langchain-demo/bin/activate # Linux/Mac # langchain-demo\Scripts\activate # Windows2.2 安装核心包与向量数据库最小化安装 LangChain 和 LangGraphpip install langchain1.3.11 langgraph0.2.42由于 LangChain 1.3 采用了模块化设计还需要安装社区组件和向量数据库连接器。这里以 Chroma轻量级内存向量库为例pip install langchain-community0.3.8 chromadb如果你需要处理 PDF、PPT 等格式额外安装文档加载器pip install pypdf unstructured2.3 配置大模型访问权限LangChain 本身不提供模型需要连接 OpenAI、Azure 或本地部署的模型。这里以 OpenAI API 为例pip install openai然后在环境变量或代码中设置 API Keyimport os os.environ[OPENAI_API_KEY] sk-... # 你的实际 key如果使用 Azure OpenAI需要设置更多参数os.environ[AZURE_OPENAI_API_KEY] your-key os.environ[AZURE_OPENAI_ENDPOINT] https://your-resource.openai.azure.com/ os.environ[OPENAI_API_VERSION] 2024-12-01-preview # 确认最新版本注意生产环境不要硬编码密钥。使用环境变量、密钥管理服务或配置文件并确保.gitignore 排除敏感信息。3. 构建 RAG 知识库从文档加载到问答生成3.1 设计 RAG 流程与项目结构一个典型的 RAG 系统包含以下步骤文档加载从文件、网页或数据库读取原始内容。文本分割将长文档切成重叠的小块平衡上下文完整性和检索精度。向量化用嵌入模型把文本块转换成向量。向量存储把向量和原文存入向量数据库建立索引。检索用户提问时把问题也向量化找到最相关的文本块。生成把问题和检索到的上下文组装成提示词送给 LLM 生成答案。项目目录结构建议rag_project/ ├── docs/ # 存放原始文档 │ ├── product_manual.pdf │ └── faq.txt ├── vector_db/ # 向量数据库存储路径 ├── config.py # 配置文件模型、路径等 ├── build_knowledge_base.py # 构建知识库脚本 ├── query_engine.py # 问答查询脚本 └── requirements.txt # 依赖列表3.2 实现文档加载与文本分割首先实现知识库构建脚本build_knowledge_base.pyfrom langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter def load_documents(doc_paths): 加载多种格式的文档 documents [] for path in doc_paths: if path.endswith(.pdf): loader PyPDFLoader(path) elif path.endswith(.txt): loader TextLoader(path, encodingutf-8) else: print(f暂不支持 {path} 格式) continue documents.extend(loader.load()) return documents def split_documents(documents, chunk_size1000, chunk_overlap200): 分割文档为重叠的文本块 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, is_separator_regexFalse, ) return text_splitter.split_documents(documents) if __name__ __main__: # 示例文档路径 doc_paths [./docs/product_manual.pdf, ./docs/faq.txt] # 加载和分割文档 raw_docs load_documents(doc_paths) print(f加载了 {len(raw_docs)} 个文档) splits split_documents(raw_docs) print(f分割为 {len(splits)} 个文本块) # 查看第一个文本块的内容和元数据 print(示例文本块, splits[0].page_content[:200]) print(元数据, splits[0].metadata)关键参数说明chunk_size1000每个文本块约 1000 字符适合大多数嵌入模型。chunk_overlap200块间重叠 200 字符避免重要信息被切碎。实际项目中需要根据文档类型调整技术文档可能需要更大块对话记录可能需要更小块。3.3 配置嵌入模型与向量数据库选择嵌入模型时考虑精度、速度和成本。OpenAI 的text-embedding-3-small在质量和价格间取得了良好平衡from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma def create_vector_store(text_chunks, persist_directory./vector_db): 创建向量数据库并持久化 # 初始化嵌入模型 embeddings OpenAIEmbeddings( modeltext-embedding-3-small, openai_api_keyos.getenv(OPENAI_API_KEY) ) # 创建向量库如果已存在会直接加载 vectorstore Chroma.from_documents( documentstext_chunks, embeddingembeddings, persist_directorypersist_directory ) # 确保变更持久化 vectorstore.persist() return vectorstore # 接前面的代码 if __name__ __main__: # ... 加载和分割文档的代码 # 创建向量库 vectorstore create_vector_store(splits) print(向量数据库构建完成)检查向量库是否正常工作# 测试检索 test_query 产品的主要功能是什么 similar_docs vectorstore.similarity_search(test_query, k2) print(f查询{test_query}) for i, doc in enumerate(similar_docs): print(f结果 {i1}{doc.page_content[:100]}...)3.4 组装完整的 RAG 链现在把检索器和 LLM 组装成问答链from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI def create_rag_chain(vectorstore): 创建 RAG 问答链 # 选择 LLM llm ChatOpenAI( modelgpt-3.5-turbo, temperature0.1, # 低温度保证答案稳定 openai_api_keyos.getenv(OPENAI_API_KEY) ) # 创建检索器可以调整检索参数 retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 3} # 返回前 3 个相关文档 ) # 组装 RAG 链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 简单地把所有上下文塞进提示词 retrieverretriever, return_source_documentsTrue, # 返回参考来源 verboseTrue # 打印详细执行过程调试用 ) return qa_chain # 在 query_engine.py 中使用 if __name__ __main__: # 加载已有的向量库 embeddings OpenAIEmbeddings() vectorstore Chroma( persist_directory./vector_db, embedding_functionembeddings ) qa_chain create_rag_chain(vectorstore) # 测试问答 question 如何重置产品密码 result qa_chain.invoke({query: question}) print(答案, result[result]) print(参考文档) for doc in result[source_documents]: print(f- {doc.metadata.get(source, 未知)}: {doc.page_content[:50]}...)3.5 RAG 系统常见问题与排查RAG 系统上线后90% 的问题集中在检索质量上。以下排查表帮你快速定位问题现象可能原因检查方式解决建议答案与文档无关检索不到相关片段检查查询向量化是否正常查看检索到的原文调整文本分割策略优化查询重写答案不完整检索片段太小或缺少上下文查看检索到的文档块内容增大 chunk_size调整重叠区域答案胡言乱语LLM 忽略了检索到的上下文检查最终提示词模板在提示词中强调基于以下上下文处理速度慢向量检索或 LLM 响应慢分阶段计时检索时间 vs 生成时间优化向量索引缓存常见查询使用更快模型检索质量优化技巧查询扩展把原始问题扩展成多个相关问题合并检索结果。重排序先用简单方法召回大量候选再用更精细的模型重新排序。混合检索结合向量检索和关键词检索兼顾语义匹配和精确术语匹配。4. 引入 LangGraph从线性 RAG 到多智能体工作流4.1 识别 RAG 的局限性基础的 RAG 在以下场景会显得力不从心用户问题需要多个专业领域知识如技术问题涉及配置、代码、运维。答案需要分步骤验证或多方确认。查询本身模糊需要先澄清用户意图。需要调用外部工具查询实时信息或执行操作。这些场景需要多个专家智能体协作每个智能体负责特定任务并按一定规则交互。这就是 LangGraph 的用武之地。4.2 设计多智能体 RAG 工作流我们设计一个增强型 RAG 系统包含三个智能体查询分析器判断问题类型技术问题、概念解释、操作指南和紧急程度。检索专家负责从知识库查找相关信息可以按问题类型调整检索策略。答案生成器综合检索结果和问题背景生成最终答案必要时要求用户澄清。工作流如图用户提问 → 查询分析器 → 检索专家 → 答案生成器 → 最终答案 ↓ ↓ 需要澄清 信息不足 ↓ ↓ 请求用户澄清 扩展检索或标记限制4.3 定义 LangGraph 状态结构LangGraph 通过状态对象在节点间传递数据。我们先定义状态 schemafrom typing import Annotated, Dict, List, Optional from typing_extensions import TypedDict from langgraph.graph.message import add_messages class GraphState(TypedDict): 图执行过程中的状态容器 # 用户原始输入 user_query: str # 查询分析结果 query_type: Optional[str] # technical, conceptual, procedural urgency: Optional[str] # high, medium, low # 检索到的文档 retrieved_docs: Annotated[List[Dict], add_messages] # 生成的答案 final_answer: Optional[str] # 需要用户澄清的点 clarification_needed: Optional[str] # 执行步骤记录调试用 execution_steps: Annotated[List[str], add_messages]关键设计点Annotated[... , add_messages]让 LangGraph 自动跟踪该字段的变更历史。每个字段都有明确职责避免状态过于复杂。Optional字段允许某些步骤跳过设置。4.4 实现各个节点函数查询分析器节点from langchain_core.prompts import ChatPromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field # 定义分析结果的结构化输出 class QueryAnalysis(BaseModel): query_type: str Field(description问题类型: technical, conceptual, procedural) urgency: str Field(description紧急程度: high, medium, low) needs_clarification: bool Field(description是否需要用户澄清) clarification_question: Optional[str] Field(description需要澄清的具体问题) def analyze_query(state: GraphState): 分析用户查询的节点函数 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 结构化输出提示词 analysis_prompt ChatPromptTemplate.from_messages([ (system, 你是一个查询分析专家。分析用户问题判断 1. 问题类型技术问题technical、概念解释conceptual、操作指南procedural 2. 紧急程度高high、中medium、低low 3. 是否需要澄清模糊点 根据问题内容客观分析不要臆断。), (human, 用户问题{query}) ]) # 绑定结构化输出 structured_llm llm.with_structured_output(QueryAnalysis) chain analysis_prompt | structured_llm analysis chain.invoke({query: state[user_query]}) # 更新状态 new_state state.copy() new_state[query_type] analysis.query_type new_state[urgency] analysis.urgency if analysis.needs_clarification: new_state[clarification_needed] analysis.clarification_question new_state[execution_steps].append(f查询分析完成类型{analysis.query_type}, 紧急度{analysis.urgency}) return new_state检索专家节点def retrieve_documents(state: GraphState): 根据查询类型调整检索策略 # 根据问题类型调整检索参数 search_configs { technical: {k: 5, score_threshold: 0.7}, # 技术问题要精确匹配 conceptual: {k: 3, score_threshold: 0.5}, # 概念问题可以宽泛些 procedural: {k: 4, score_threshold: 0.6} # 操作指南需要平衡 } config search_configs.get(state[query_type], {k: 3, score_threshold: 0.6}) # 加载向量库实际项目应该注入依赖这里简化 embeddings OpenAIEmbeddings() vectorstore Chroma(persist_directory./vector_db, embedding_functionembeddings) # 创建检索器 retriever vectorstore.as_retriever( search_typesimilarity_score_threshold, search_kwargsconfig ) # 执行检索 docs retriever.invoke(state[user_query]) new_state state.copy() new_state[retrieved_docs] [ {content: doc.page_content, metadata: doc.metadata} for doc in docs ] new_state[execution_steps].append(f检索到 {len(docs)} 个相关文档) return new_state答案生成器节点def generate_answer(state: GraphState): 生成最终答案 # 如果有需要澄清的点先返回澄清请求 if state.get(clarification_needed): new_state state.copy() new_state[final_answer] f为了给您更准确的答案请澄清{state[clarification_needed]} new_state[execution_steps].append(已请求用户澄清) return new_state # 检查是否检索到足够信息 if not state.get(retrieved_docs): new_state state.copy() new_state[final_answer] 抱歉知识库中没有找到相关信息。请尝试换种方式提问或联系人工客服。 new_state[execution_steps].append(未检索到相关文档返回默认回复) return new_state llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # 根据紧急程度调整回答风格 style_instructions { high: 回答要简洁直接重点突出操作步骤, medium: 回答要平衡详细和简洁包含必要解释, low: 回答可以详细些包含背景知识和相关概念 } style style_instructions.get(state[urgency], 平衡详细和简洁) prompt ChatPromptTemplate.from_messages([ (system, f你是一个专业的客服助手。基于以下检索到的文档内容回答问题。 回答要求{style} 如果信息不足如实告知限制。 如果文档间有矛盾指出不确定性。 检索到的文档 \n\n.join([f文档 {i1}{doc[content]} for i, doc in enumerate(state[retrieved_docs])])), (human, 问题{query}) ]) chain prompt | llm answer chain.invoke({query: state[user_query]}) new_state state.copy() new_state[final_answer] answer.content new_state[execution_steps].append(已生成最终答案) return new_state4.5 组装 LangGraph 并定义流程逻辑现在把各个节点组装成完整的工作流from langgraph.graph import StateGraph, END def create_agentic_rag_graph(): 创建多智能体 RAG 图 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(analyze_query, analyze_query) workflow.add_node(retrieve_documents, retrieve_documents) workflow.add_node(generate_answer, generate_answer) # 设置入口点 workflow.set_entry_point(analyze_query) # 定义边流程逻辑 workflow.add_edge(analyze_query, retrieve_documents) workflow.add_edge(retrieve_documents, generate_answer) workflow.add_edge(generate_answer, END) # 编译图 return workflow.compile() # 使用图 if __name__ __main__: graph create_agentic_rag_graph() # 准备初始状态 initial_state { user_query: 如何配置数据库连接池, query_type: None, urgency: None, retrieved_docs: [], final_answer: None, clarification_needed: None, execution_steps: [] } # 执行图 result graph.invoke(initial_state) print(最终答案, result[final_answer]) print(执行步骤, result[execution_steps])4.6 高级特性条件分支与循环上面的线性流程已经能处理很多场景但 LangGraph 的真正威力在于复杂逻辑。比如添加一个质量检查节点如果答案质量不够就重新检索def quality_check(state: GraphState): 检查答案质量决定是否重新检索 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 评估答案质量考虑 1. 是否直接回答了问题 2. 是否基于提供的文档内容 3. 是否完整覆盖了问题要点 只返回pass或retry不要解释。), (human, f问题{state[user_query]}\n答案{state[final_answer]}) ]) chain prompt | llm judgment chain.invoke({}).content.strip().lower() return judgment pass # 修改图定义添加条件边 def create_advanced_rag_graph(): workflow StateGraph(GraphState) workflow.add_node(analyze_query, analyze_query) workflow.add_node(retrieve_documents, retrieve_documents) workflow.add_node(generate_answer, generate_answer) workflow.add_node(expand_retrieval, expand_retrieval) # 扩展检索的新节点 workflow.set_entry_point(analyze_query) workflow.add_edge(analyze_query, retrieve_documents) workflow.add_edge(retrieve_documents, generate_answer) # 条件边根据质量检查结果决定流向 workflow.add_conditional_edges( generate_answer, quality_check, # 条件函数 { True: END, # 质量合格→结束 False: expand_retrieval # 质量不合格→扩展检索 } ) workflow.add_edge(expand_retrieval, generate_answer) # 扩展后重新生成 return workflow.compile()这种带反馈循环的设计让系统能够自我修正显著提升最终输出质量。5. 生产环境部署考量5.1 性能优化策略LangChain/LangGraph 应用在生产环境要注意向量检索优化使用更高效的向量索引HNSW。对大规模知识库进行分片存储。缓存常见查询的检索结果。LLM 调用优化使用流式响应改善用户体验。实现请求批处理减少 API 调用次数。设置合理的超时和重试机制。内存与并发管理监控每个节点的内存使用。使用异步版本提高并发处理能力。对资源密集型操作如向量化实施限流。5.2 监控与可观测性在生产环境必须添加监控import time from functools import wraps def monitor_node_performance(node_name): 监控节点执行时间的装饰器 def decorator(func): wraps(func) def wrapper(state, *args, **kwargs): start_time time.time() try: result func(state, *args, **kwargs) duration time.time() - start_time # 记录到监控系统 print(f{node_name} 执行时间: {duration:.2f}s) return result except Exception as e: duration time.time() - start_time print(f{node_name} 执行失败: {e}, 耗时: {duration:.2f}s) raise return wrapper return decorator # 使用示例 monitor_node_performance(检索节点) def retrieve_documents(state: GraphState): # ... 原有实现5.3 错误处理与降级方案健壮的系统需要处理各种异常from langchain.schema import BaseRetriever class FallbackRetriever(BaseRetriever): 带降级方案的检索器 def __init__(self, primary_retriever, fallback_retriever): self.primary primary_retriever self.fallback fallback_retriever def get_relevant_documents(self, query): try: # 先尝试主检索器 return self.primary.get_relevant_documents(query) except Exception as e: print(f主检索器失败: {e}, 使用降级方案) # 降级到简单关键词匹配 return self.fallback.get_relevant_documents(query)6. 常见问题深度排查6.1 LangChain 版本兼容问题问题更新后代码报错提示模块不存在或参数错误。排查步骤检查当前版本pip show langchain langchain-community langgraph查看官方迁移指南LangChain 主要版本更新通常有详细迁移说明。对比导入语句1.3.x 版本很多组件从langchain移到了langchain_community。示例修复# 旧版本0.1.x from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma # 新版本1.3.x from langchain_openai import OpenAIEmbeddings # 或从 langchain_community from langchain_community.vectorstores import Chroma6.2 LangGraph 状态管理错误问题节点间状态传递异常字段丢失或类型错误。排查步骤确认状态 schema 中所有字段都有正确的类型注解。检查每个节点返回的状态对象是否包含所有必需字段。使用add_messages注解跟踪关键字段的变更历史。调试技巧# 在节点函数开头添加调试输出 print(f进入节点当前状态键值: {list(state.keys())}) # 检查特定字段 if retrieved_docs in state: print(f检索文档数: {len(state[retrieved_docs])})6.3 RAG 检索质量不佳问题总是检索不到相关文档或检索结果不准确。系统化优化流程评估检索效果准备测试问题集人工标注预期相关文档。调整文本分割尝试不同的 chunk_size 和 overlap找到最佳平衡点。优化查询表达实现查询重写或扩展改善查询向量质量。混合检索策略结合语义检索和关键词检索的优势。6.4 多智能体协作死循环问题智能体间相互等待或循环调用无法终止。解决方案设置最大迭代次数在状态中维护迭代计数器达到上限强制退出。设计明确的终止条件每个循环节点都要有清晰的完成标准。添加超时机制对整个图执行设置时间限制。class GraphState(TypedDict): # ... 其他字段 iteration_count: int # 防止无限循环 def should_continue(state: GraphState): 检查是否应该继续执行 if state.get(iteration_count, 0) 3: # 最多迭代3次 return stop # ... 其他终止条件7. 扩展方向与最佳实践7.1 从演示项目到生产系统的关键升级维度演示版本生产版本配置管理硬编码在代码中外部配置文件/环境变量错误处理基本 try-catch分级降级、详细日志、监控告警性能单次调用异步、批处理、缓存、连接池安全基础 API 密钥密钥轮转、访问控制、输入消毒可观测性print 语句结构化日志、指标收集、分布式追踪7.2 学习路径建议要真正掌握 LangChain 和 LangGraph建议按这个顺序实践基础掌握完成本文的 RAG 系统理解每个组件的作用。深入组件分别深入研究提示词工程、向量检索、链的组装。复杂工作流用 LangGraph 实现需要条件判断、循环、多智能体的场景。生产化改造添加测试、监控、部署流水线。领域定制针对特定行业客服、教育、医疗定制工作流。7.3 避免常见反模式过度工程不是每个项目都需要 LangGraph。简单的线性流程用 Chain 更直接。忽略检索质量在花哨的智能体架构上投入过多却忽视基础的文档处理和检索效果。硬编码业务逻辑把应该由 LLM 推理的判断硬编码成 if-else 规则。缺乏评估体系没有建立客观的评估标准无法衡量优化是否有效。LangChain 和 LangGraph 是强大的工具但工具的价值在于解决实际问题。从具体的业务场景出发先用最简单的方式验证价值再逐步引入复杂功能这才是稳健的技术落地路径。