在实际的大模型应用开发中很多团队会遇到一个典型困境单个工具调用或简单的问答流程很容易跑通但一旦涉及多步骤决策、状态记忆、外部工具集成和知识库检索的复杂场景代码就会迅速变得难以维护和扩展。这正是 LangChain 生态试图解决的核心问题。本文将以 2026 年的技术视角带你从零搭建一个整合了 Agent智能体、RAG检索增强生成和 LangGraph工作流编排的企业级项目让你避开初期常见的架构陷阱和配置弯路。这个项目实战会聚焦于一个可运行的案例构建一个能自动处理用户技术问题的工作流。它不仅能从内部知识库RAG检索相关信息还能根据问题复杂度决定是否调用外部工具如代码分析器并通过 LangGraph 管理整个对话状态。学完后你将掌握如何将这些组件平滑地组装成一个可靠的生产应用而不仅仅是跑通 demo。1. 理解 LangChain、LangGraph、Agent 与 RAG 的各自角色在开始写代码之前必须先理清这几个核心概念分别解决什么问题以及它们如何协作。很多初学者失败的原因是把它们混为一谈或者在不合适的场景强行套用。1.1 LangChain是大模型应用的“脚手架”和“工具箱”LangChain 本身不是一个独立的应用运行时而是一个开发框架。它提供了大量标准化组件比如与各种大模型OpenAI、通义千问、DeepSeek 等对话的通用接口。处理提示词Prompt的模板和管理工具。常见的链Chain式调用模式例如LLMChain、SequentialChain。与外部工具Calculator、SQLExecutor、API 工具集成的基础能力。你可以把 LangChain 理解为 Spring 框架之于 Java 应用。它定义了组件如何组织、配置如何加载、公共逻辑如何复用但具体业务逻辑需要你自己填充。1.2 Agent是能自主调用工具的“决策大脑”Agent 的核心能力是“根据当前目标和上下文决定下一步该做什么以及调用哪个工具”。一个最简单的 Agent 工作流程是接收用户输入如“计算 3 的平方根并告诉我结果”。分析输入判断需要调用计算器工具。执行计算器工具获得结果。将结果组织成自然语言回复给用户。LangChain 内置了多种 Agent 类型如 ReAct、OpenAI Functions它们本质上都是基于大模型的推理能力来驱动工具调用。在企业级应用中Agent 常用于需要动态决策的场景比如客户问题自动分类、多步数据查询、复杂任务分解等。1.3 RAG是为大模型注入最新、专有知识的“记忆外挂”大模型本身的知识有截止日期且无法学习企业内部的私有文档。RAG 通过以下步骤解决这个问题知识预处理将企业内部文档Markdown、PDF、Word拆分成小块Chunk。向量化存储使用文本嵌入模型将每个 Chunk 转换为向量存入向量数据库。检索增强当用户提问时先将问题转换为向量从向量库中找出最相关的几个 Chunk。生成答案将问题和检索到的 Chunk 一起送给大模型让它基于这些上下文生成答案。RAG 的核心价值是让大模型能回答“超出训练数据范围”或“涉及内部知识”的问题。在技术客服、内部知识查询等场景中必不可少。1.4 LangGraph是管理复杂、有状态工作流的“流程引擎”当你的应用需要多次调用 Agent、RAG 或外部服务并且这些调用之间存在状态依赖时LangGraph 就派上用场了。例如一个客服对话可能先查知识库如果没解决就转人工并记录当前对话状态。一个数据分析任务可能需要先查询 A 系统再用结果查询 B 系统最后生成报告。LangGraph 允许你用图Graph的方式定义工作流节点和流转条件并自动维护整个流程的状态。它弥补了 LangChain 中普通 Chain 无法轻松处理多步、有状态、带循环的流程缺陷。1.5 四者关系为什么需要组合使用在实际项目中很少有场景只用到其中一个。典型的工作流如下用户输入问题。LangGraph驱动的工作流开始执行初始化状态。第一个节点调用RAG模块从知识库检索相关文档。第二个节点由Agent判断如果检索结果足够回答直接生成回复如果不够Agent 决定调用哪个工具获取更多信息。LangGraph根据 Agent 的决策流转到对应工具节点执行后更新状态。最终回复由LangChain的提示词模板和 LLM 包装后返回给用户。这个流程中每个组件各司其职LangChain 提供基础能力RAG 提供知识Agent 提供决策LangGraph 负责编排。接下来我们通过一个实战项目把它们串起来。2. 环境准备与依赖配置避开版本冲突陷阱企业级项目最怕环境不一致和依赖冲突。下面是一个经过验证的依赖组合适用于 Python 3.9-3.11 环境。建议使用conda或venv创建隔离环境。2.1 项目结构与核心依赖创建一个新项目目录结构如下企业级Agent-RAG-LangGraph/ ├── requirements.txt ├── config/ │ └── settings.py ├── knowledge_base/ │ ├── raw_documents/ │ └── processed/ ├── agents/ │ └── tech_support_agent.py ├── graphs/ │ └── support_workflow.py ├── chains/ │ └── rag_chain.py └── main.pyrequirements.txt内容如下。注意这里锁定了关键库的版本这是避免后期莫名其妙错误的关键langchain0.2.0 langchain-community0.0.29 langchain-core0.2.0 langgraph0.1.0 langchain-text-splitters0.2.0 langchain-openai0.1.0 chromadb0.4.24 openai1.30.0 tiktoken0.6.0 pydantic2.7.0 python-dotenv1.0.0 fastapi0.110.0 uvicorn0.29.0安装依赖pip install -r requirements.txt2.2 模型配置与密钥管理在项目根目录创建.env文件用于存储敏感配置OPENAI_API_KEY你的OpenAI密钥 OPENAI_BASE_URL可选如果使用第三方兼容接口 MODEL_NAMEgpt-4-turbo-preview EMBEDDING_MODELtext-embedding-3-small在config/settings.py中读取配置import os from dotenv import load_dotenv load_dotenv() class Settings: openai_api_key os.getenv(OPENAI_API_KEY) openai_base_url os.getenv(OPENAI_BASE_URL) model_name os.getenv(MODEL_NAME, gpt-4-turbo-preview) embedding_model os.getenv(EMBEDDING_MODEL, text-embedding-3-small) settings Settings()注意生产环境中不应将密钥硬编码在文件中而应使用环境变量或专业的密钥管理服务。此处为演示方便使用.env。2.3 常见环境问题排查即使按照上述步骤操作仍可能遇到环境问题。下面是几个高频问题的排查清单问题现象可能原因解决方案导入langchain时报错No module named langchain虚拟环境未激活或依赖未安装确认激活环境后重新执行pip install -r requirements.txt调用 OpenAI 接口时报认证错误API_KEY 未正确加载检查.env文件路径、变量名是否正确确认密钥有效安装chromadb时编译错误系统缺少 C 编译环境在 Ubuntu 上安装build-essential在 macOS 上安装 Xcode Command Line Tools运行时报pydantic版本冲突其他包依赖了不兼容的pydantic版本使用 pip list环境准备阶段最忌讳急于求成。务必确保每一步都验证通过后再进入下一阶段。3. 构建 RAG 知识库从原始文档到可检索向量RAG 系统的质量很大程度上取决于知识库的构建质量。糟糕的文档拆分和向量化会导致检索结果不相关进而让大模型“胡说八道”。3.1 文档预处理与拆分策略在knowledge_base/raw_documents/中放入你的技术文档如.md、.txt文件。然后创建预处理脚本knowledge_base/process_documents.pyimport os from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader def process_documents(): raw_docs_path ./knowledge_base/raw_documents processed_path ./knowledge_base/processed # 支持的文件类型 supported_extensions [.md, .txt] documents [] for file in os.listdir(raw_docs_path): if any(file.endswith(ext) for ext in supported_extensions): loader TextLoader(os.path.join(raw_docs_path, file), encodingutf-8) documents.extend(loader.load()) # 使用递归字符拆分器保持段落语义完整性 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个 chunk 约 1000 字符 chunk_overlap200, # chunk 间重叠 200 字符避免割裂上下文 length_functionlen, separators[\n\n, \n, 。, , , , , ] ) chunks text_splitter.split_documents(documents) # 为每个 chunk 添加元数据便于追溯来源 for i, chunk in enumerate(chunks): chunk.metadata[chunk_id] i if source not in chunk.metadata: chunk.metadata[source] unknown return chunks if __name__ __main__: chunks process_documents() print(f共处理 {len(chunks)} 个文档块)关键参数说明chunk_size1000平衡检索精度和上下文长度。太小会丢失语义太大会引入噪声。chunk_overlap200避免一个句子被拆到两个 chunk 中影响检索连贯性。separators按中文标点优先拆分比单纯按字符数拆分更合理。3.2 向量化存储与 ChromaDB 配置创建knowledge_base/vector_store.pyfrom langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from config.settings import settings def create_vector_store(chunks): embeddings OpenAIEmbeddings( modelsettings.embedding_model, openai_api_keysettings.openai_api_key, openai_api_basesettings.openai_base_url ) vector_store Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./knowledge_base/chroma_db ) return vector_store def get_vector_store(): embeddings OpenAIEmbeddings( modelsettings.embedding_model, openai_api_keysettings.openai_api_key, openai_api_basesettings.openai_base_url ) vector_store Chroma( persist_directory./knowledge_base/chroma_db, embedding_functionembeddings ) return vector_store运行向量化from knowledge_base.process_documents import process_documents from knowledge_base.vector_store import create_vector_store chunks process_documents() vector_store create_vector_store(chunks) print(知识库向量化完成)3.3 RAG 检索链的实现在chains/rag_chain.py中实现检索增强生成链from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from config.settings import settings from knowledge_base.vector_store import get_vector_store def create_rag_chain(): llm ChatOpenAI( modelsettings.model_name, openai_api_keysettings.openai_api_key, openai_api_basesettings.openai_base_url, temperature0.1 # 降低随机性保证技术答案的准确性 ) prompt_template PromptTemplate( template你是一个技术专家助手。请根据以下上下文信息回答问题。如果上下文不足以回答问题请如实告知不要编造信息。 上下文 {context} 问题{question} 答案, input_variables[context, question] ) vector_store get_vector_store() rag_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervector_store.as_retriever( search_typesimilarity, search_kwargs{k: 3} # 检索最相关的 3 个文档块 ), chain_type_kwargs{prompt: prompt_template}, return_source_documentsTrue ) return rag_chain if __name__ __main__: chain create_rag_chain() result chain.invoke({query: 如何配置 LangChain 的 Agent}) print(答案, result[result]) print(来源文档, [doc.metadata.get(source, unknown) for doc in result[source_documents]])这个 RAG 链已经具备了生产可用的基础能力可配置的检索数量、明确的提示词模板、来源追溯。接下来我们要让它与 Agent 协作。4. 创建技术支持 Agent实现多工具决策能力单纯的 RAG 只能回答知识库内的问题对于需要计算、查询、判断的复杂问题无能为力。这时就需要 Agent 出场。4.1 定义工具集先创建几个技术支持场景常用的工具。在agents/tools/目录下创建agents/tools/code_analyzer.pyfrom langchain.tools import tool tool def analyze_code_snippet(code: str) - str: 分析提供的代码片段识别潜在问题或优化建议。 Args: code: 需要分析的代码字符串 Returns: 分析结果和建议 # 这里可以集成真实的代码分析工具如 pylint、flake8 的调用 # 为演示简化实现 if print( in code and logging not in code: return 检测到直接使用 print 语句在生产环境中建议使用 logging 模块进行日志记录。 elif password in code.lower() and env not in code.lower(): return 检测到代码中硬编码密码建议使用环境变量管理敏感信息。 else: return 代码结构基本合理未发现明显安全问题。agents/tools/system_checker.pyfrom langchain.tools import tool import psutil tool def check_system_resources() - str: 检查当前系统资源使用情况CPU、内存、磁盘。 Returns: 系统资源状态报告 cpu_percent psutil.cpu_percent(interval1) memory psutil.virtual_memory() disk psutil.disk_usage(/) return f 系统资源状态 - CPU 使用率{cpu_percent}% - 内存使用率{memory.percent}% (可用{memory.available // (1024**3)}GB) - 磁盘使用率{disk.percent}% (可用{disk.free // (1024**3)}GB) 4.2 构建技术支持 Agent在agents/tech_support_agent.py中创建主 Agentfrom langchain.agents import AgentExecutor, create_openai_functions_agent from langchain import hub from langchain_openai import ChatOpenAI from config.settings import settings from agents.tools.code_analyzer import analyze_code_snippet from agents.tools.system_checker import check_system_resources def create_tech_support_agent(): llm ChatOpenAI( modelsettings.model_name, openai_api_keysettings.openai_api_key, openai_api_basesettings.openai_base_url, temperature0 ) tools [analyze_code_snippet, check_system_resources] prompt hub.pull(hwchase17/openai-functions-agent) agent create_openai_functions_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue ) return agent_executor if __name__ __main__: agent create_tech_support_agent() result agent.invoke({ input: 请分析这段代码print(hello world) }) print(result[output])这个 Agent 已经具备了工具调用能力但它还无法与 RAG 系统协作也无法处理多轮对话。这就是我们需要 LangGraph 的原因。5. 使用 LangGraph 编排完整工作流现在到了最关键的环节用 LangGraph 把 RAG、Agent 和状态管理串联起来。我们要构建的技术支持工作流逻辑如下接收用户问题先用 RAG 检索相关知识Agent 判断是否需额外工具调用根据工具调用结果决定下一步生成最终回复并保存对话历史5.1 定义状态结构在graphs/support_workflow.py中首先定义状态类from typing import Dict, Any, List, Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class GraphState(TypedDict): messages: Annotated[List[Any], add_messages] question: str rag_result: str tool_calls: List[Dict[str, Any]] final_answer: str这个状态对象会贯穿整个工作流每个节点都可以读取和修改其中的字段。5.2 实现工作流节点接着实现各个节点函数from chains.rag_chain import create_rag_chain from agents.tech_support_agent import create_tech_support_agent rag_chain create_rag_chain() tech_agent create_tech_support_agent() def retrieve_knowledge(state: GraphState): RAG 检索节点 question state[question] result rag_chain.invoke({query: question}) return { rag_result: result[result], messages: state[messages] [{role: system, content: f知识库检索结果{result[result]}}] } def agent_decision(state: GraphState): Agent 决策节点 question state[question] rag_context state[rag_result] # 组合问题和 RAG 结果供 Agent 决策 augmented_question f 用户问题{question} 相关知识库信息{rag_context} 请判断是否需要调用工具进一步分析还是直接基于知识库信息回答。 result tech_agent.invoke({input: augmented_question}) return { messages: state[messages] [{role: assistant, content: result[output]}], final_answer: result[output] } def should_continue(state: GraphState): 条件判断节点决定工作流是否结束 # 这里可以根据 Agent 的输出判断是否需要进一步处理 # 简化实现总是结束流程 return end def finalize_response(state: GraphState): 最终响应节点 return { final_answer: state[final_answer], messages: state[messages] [{role: system, content: 对话流程结束}] }5.3 构建并运行图工作流完成节点定义后构建完整的工作流图from langgraph.graph import StateGraph, END def create_support_workflow(): workflow StateGraph(GraphState) # 添加节点 workflow.add_node(retrieve, retrieve_knowledge) workflow.add_node(agent, agent_decision) workflow.add_node(finalize, finalize_response) # 设置入口点 workflow.set_entry_point(retrieve) # 添加边定义流程 workflow.add_edge(retrieve, agent) workflow.add_conditional_edges( agent, should_continue, { end: finalize, } ) workflow.add_edge(finalize, END) return workflow.compile() # 创建图实例 support_workflow create_support_workflow()5.4 测试完整工作流创建测试脚本验证整个系统def test_workflow(): config {recursion_limit: 50} initial_state { messages: [{role: user, content: 我的代码中有 print(password) 这样的语句有什么安全问题吗}], question: 我的代码中有 print(password) 这样的语句有什么安全问题吗, rag_result: , tool_calls: [], final_answer: } result support_workflow.invoke(initial_state, configconfig) print(最终答案, result[final_answer]) print(\n完整对话历史) for msg in result[messages]: print(f{msg[role]}: {msg[content]}) if __name__ __main__: test_workflow()这个工作流展示了 LangGraph 的核心价值明确的状态流转、清晰的节点职责、灵活的条件分支。在实际生产中你可以在此基础上添加更多节点比如人工审核节点、重试机制节点、会话持久化节点等。6. 生产环境部署与优化建议让系统在开发环境运行只是第一步要真正用于生产还需要考虑很多因素。6.1 性能优化配置向量检索优化# 在创建检索器时调整参数 retriever vector_store.as_retriever( search_typemmr, # 使用最大边际相关度平衡相关性和多样性 search_kwargs{k: 5, fetch_k: 20} # 先取20个再精选5个 )大模型调用优化llm ChatOpenAI( modelsettings.model_name, timeout30, # 设置超时避免长时间阻塞 max_retries2, # 失败重试 model_kwargs{seed: 42} # 设置随机种子保证可复现性 )6.2 错误处理与容灾在工作流中增加错误处理节点def error_handler(state: GraphState, error: Exception): 错误处理节点 return { final_answer: f抱歉系统处理时遇到错误{str(error)}。请稍后重试或联系管理员。, messages: state[messages] [{role: system, content: f错误发生{str(error)}}] } # 在构建图时添加错误处理 workflow.add_node(error_handler, error_handler)6.3 监控与日志添加详细的运行日志便于问题排查import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(app.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) def retrieve_knowledge(state: GraphState): logger.info(f开始检索知识库问题{state[question]}) # ... 原有逻辑 logger.info(f检索完成找到相关文档数{len(result[source_documents])}) return result6.4 安全考虑输入验证对所有用户输入进行 sanitize防止提示词注入攻击。输出过滤对模型输出进行内容安全检查避免生成不当内容。权限控制不同用户只能访问授权范围内的知识库内容。API 限流防止恶意用户过度消耗 API 额度。7. 常见问题与排查指南在实际部署和运行过程中你会遇到各种问题。下面是按现象分类的排查指南。7.1 RAG 检索相关问题问题现象可能原因解决方案检索结果不相关文档拆分策略不合理调整 chunk_size 和分隔符尝试按段落或章节拆分检索速度慢向量数据库未优化为 ChromaDB 创建索引或考虑使用更高效的向量数据库总是返回相同文档向量化模型不适合领域文本尝试使用领域适配的嵌入模型或微调现有模型7.2 Agent 决策异常问题现象可能原因解决方案Agent 不调用工具提示词未明确工具使用场景在 Agent 提示词中提供更具体的工具调用示例工具调用参数错误工具描述不够清晰完善工具的 docstring明确参数格式和类型无限循环调用工具缺少调用次数限制在 AgentExecutor 中设置max_iterations参数7.3 LangGraph 工作流问题问题现象可能原因解决方案状态丢失节点间状态字段不匹配检查每个节点返回的字段与状态定义是否一致流程卡住条件判断逻辑有误在should_continue函数中添加详细日志内存泄漏对话历史无限增长实现历史消息截断或摘要机制7.4 性能问题问题现象可能原因解决方案响应时间过长串行调用过多考虑将可并行节点并行化如同时检索多个知识库API 调用超限未实施速率限制添加请求队列和限流机制内存占用过高大文件未及时释放检查向量数据库连接和文件句柄是否正确关闭8. 扩展方向与最佳实践掌握了基础架构后你可以从以下几个方向进一步优化系统。8.1 知识库质量提升增量更新实现知识库的增量更新机制避免全量重建。质量评估建立检索结果质量评估体系自动识别低质量文档块。多模态支持扩展支持图片、表格等非文本内容的检索。8.2 Agent 能力增强工具发现实现动态工具注册和发现机制。多 Agent 协作构建多个专业 Agent 协同工作的系统。学习优化基于用户反馈持续优化 Agent 的决策质量。8.3 工作流复杂化人工介入在关键节点引入人工审核流程。版本管理实现工作流版本的灰度发布和回滚。A/B 测试对不同工作流版本进行效果对比测试。这个项目架构已经具备了企业级应用的基础特征模块化、可扩展、有状态管理。真正的价值不在于一次性构建完美系统而在于建立能够持续迭代的技术基础。建议先从一个小而具体的业务场景开始验证再逐步扩展复杂度。