这次我们来看一个工业级 Agent 架构的实战项目。如果你正在寻找一个能真正落地的多智能体解决方案而不是停留在概念层面那么 LangGraph 结合 LangChain 的架构值得深入研究。它不是一个简单的玩具而是提供了从状态管理、多智能体协作到 RAG 与多模型接入的完整工程化框架。项目的核心在于LangGraph它是 LangChain 生态中用于构建有状态、多智能体应用的核心库。与传统的链式调用不同LangGraph 引入了StateGraph来显式管理应用状态并通过图Graph的概念来编排智能体Agent或工具Tool的执行流程。这使得构建复杂的、带循环和条件分支的多步骤应用成为可能比如客服机器人、数据分析流水线、自动化研究助手等。本文将带你从零开始深度拆解如何利用 LangGraph 构建一个工业级 Agent 系统。我们会重点关注StateGraph 的状态管控机制、多智能体项目的工程化落地以及如何无缝集成RAG 知识库和多模型如 OpenAI、DeepSeek、智谱等。整个过程强调可复现、可调试和可扩展让你看完就能动手搭建自己的智能体应用。1. 核心能力速览在深入代码之前我们先快速了解 LangGraph LangChain 架构的核心能力与门槛帮助你判断是否适合你的项目。能力项说明项目类型多智能体编排与工作流框架非单一模型核心库LangGraph (用于编排) LangChain (用于组件集成)编程语言Python (主流)另有 LangChain4j (Java/Scala)硬件门槛无特定 GPU 要求。框架本身是编排逻辑计算负载取决于接入的模型如本地大模型需 GPU调用 API 则不需要。核心概念StateGraph(状态图)、Node(节点/智能体)、Edge(边/条件流转)、Checkpointer(状态检查点)关键特性有状态执行、循环与分支、多智能体协作、持久化状态、可视化调试是否支持 API是。可将构建的 Graph 封装为 FastAPI 等 Web 服务提供异步任务接口。是否支持批量任务是。可通过异步队列或循环调度处理批量输入状态相互隔离。适合场景复杂决策流程、多步骤任务自动化、需记忆上下文的对话系统、集成 RAG 的问答引擎、多模型路由场景。启动/部署方式通过 Python 脚本启动或封装为 Docker 容器服务。简单来说如果你需要构建一个超越简单问答、需要记忆、判断、循环调用工具或不同模型的 AI 应用这个架构就是为你准备的。2. 适用场景与使用边界适合谁解决什么问题AI 应用开发者希望将多个 LLM 调用、工具使用、数据查询步骤串联成稳定工作流。产品经理/业务分析师需要将复杂的业务逻辑如审核、客服、报告生成转化为可执行的 AI 自动化流程。研究型工程师探索多智能体协作、强化学习与规划等前沿方向需要一个可靠的基础框架。典型解决场景智能客服升级用户问题 → 意图分类节点 → 分别路由到“订单查询智能体”、“技术支持智能体”或“RAG 知识库问答智能体” → 汇总回复。自动化研究助手输入研究主题 → 联网搜索节点 → 总结归纳节点 → 批判性分析节点 → 生成报告节点。内容审核流水线上传内容 → 文本敏感词检测节点 → 图片鉴黄模型节点 → 综合判定节点 → 记录日志并执行通过/驳回操作。使用边界与注意事项不是“开箱即用”的模型你需要自己定义智能体、工具和业务流程。它提供的是“脚手架”和“发动机”而不是“整车”。学习曲线存在需要理解状态State、图Graph、节点Node等概念对 Python 编程有一定要求。计算成本取决于接入的模型如果全部使用 GPT-4 等昂贵 API运行成本会随着流程复杂度增加。需做好预算管理和限流。合规与安全当智能体可以执行“联网搜索”、“发送邮件”、“操作数据库”等工具时必须设置严格的权限控制和用户输入验证防止恶意调用。3. 环境准备与前置条件开始搭建前请确保你的开发环境满足以下要求。这是一个典型的 Python 项目环境配置。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (推荐 Ubuntu)。本文示例在 Linux/Mac 环境下通用。Python 版本 3.10。强烈建议使用 3.10 或 3.11以获得最佳的库兼容性。包管理工具pip或conda。版本控制Git可选但推荐。网络能正常访问 PyPI 和互联网用于安装包和可能调用外部 API。关键依赖库核心是langgraph和langchain。我们将安装包含常用功能的完整包。# 创建并进入项目目录 mkdir langgraph-agent-project cd langgraph-agent-project # 创建虚拟环境 (可选但推荐) python -m venv venv # Windows 激活: venv\Scripts\activate # Linux/Mac 激活: source venv/bin/activate # 安装核心框架及常用扩展 pip install langgraph langchain langchain-community langchain-openai # 如果需要使用 OpenAI 模型还需配置 API Key不是安装包 # 如果需要 RAG安装向量数据库客户端例如 Chroma pip install chromadb langchain-chroma # 如果需要可视化安装 langgraph studio (目前通常通过 Docker) # 如果需要处理 PDF安装 pypdf 或 langchain 相关文档加载器 pip install pypdf模型接入准备LangGraph/Chain 本身不提供模型需要接入。你需要准备相应 API 密钥或本地模型端点。OpenAI准备OPENAI_API_KEY。DeepSeek准备DEEPSEEK_API_KEY并使用langchain-openai兼容接口因其兼容 OpenAI SDK。智谱 AI准备ZHIPUAI_API_KEY使用langchain-zhipu。本地模型通过langchain-ollama或langchain-openai兼容本地部署的 vLLM 等接入。4. 项目结构与核心概念拆解在写代码前我们先规划一个清晰的项目结构并彻底理解 StateGraph。推荐项目结构langgraph-agent-project/ ├── agents/ # 存放不同智能体的定义 │ ├── __init__.py │ ├── classifier.py # 分类智能体 │ └── researcher.py # 研究智能体 ├── tools/ # 存放自定义工具 │ ├── __init__.py │ └── web_search.py ├── graphs/ # 存放定义的工作流图 │ ├── __init__.py │ └── main_workflow.py ├── state/ # 存放自定义状态模式 │ └── agent_state.py ├── config.py # 配置文件 (API Keys, 模型设置) ├── requirements.txt # 依赖列表 └── main.py # 应用入口启动服务或运行测试核心概念深度解析State (状态)一个字典或 Pydantic 模型贯穿整个图执行的生命周期。所有节点都读取和修改这个共享状态。例如状态可能包含{question: 用户问题, category: 分类结果, answer: 最终答案}。Node (节点)图中的一个执行单元。它可以是一个简单的函数一个 LangChain Chain或一个完整的 Agent。节点接收当前 State执行操作并返回一个更新后的 State。Edge (边)连接节点的路径决定执行流程。分为两种普通边 (Conditional Edge)根据条件决定下一个节点。固定边无条件指向下一个节点。StateGraphlanggraph的核心类。你通过add_node添加节点通过add_edge或add_conditional_edges连接它们最后调用compile()将其编译成一个可执行的CompiledGraph对象。5. 实战构建一个 RAG 多模型路由的智能体工作流我们来构建一个实战项目智能问答路由系统。目标用户输入一个问题系统先判断问题类型然后路由到最合适的处理节点通用聊天、专业领域 RAG 查询、需要联网搜索的研究性问题。流程分类节点判断问题属于[general_chat, domain_qa, research]中的哪一类。路由general_chat→ 调用通用大模型如 GPT-3.5直接回答。domain_qa→ 从RAG 知识库如公司内部文档中检索并生成答案。research→ 调用联网搜索工具获取最新信息再由分析模型如 DeepSeek总结。汇总节点格式化最终答案。5.1 定义状态模式首先在state/agent_state.py中定义我们的状态结构。# state/agent_state.py from typing import TypedDict, List, Optional, Literal from langgraph.graph.message import add_messages class AgentState(TypedDict): 定义贯穿整个工作流的状态结构。 # 用户输入 question: str # 分类结果 category: Optional[Literal[general_chat, domain_qa, research]] # 检索到的文档用于 RAG documents: List[str] # 从网络搜索到的信息 search_results: Optional[str] # 最终答案 answer: Optional[str] # 消息历史如果要做多轮对话 messages: list5.2 创建工具在tools/web_search.py中创建一个简单的模拟搜索工具实际项目可接入 SerperAPI、Tavily 等。# tools/web_search.py from langchain.tools import tool import asyncio tool def web_search_tool(query: str) - str: 执行联网搜索。传入搜索查询词返回搜索结果摘要。 # 模拟网络延迟 asyncio.sleep(1) # 这里是模拟数据真实情况应调用搜索 API mock_results f 根据网络搜索关于 {query} 的最新信息 1. 相关概念 A 的解释是... 2. 近期发展包括... 3. 主流观点认为... return mock_results5.3 创建智能体节点我们在agents/下创建三个节点对应的函数。# agents/classifier.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser import os from config import OPENAI_API_KEY # 初始化一个用于分类的小模型节约成本 classifier_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyOPENAI_API_KEY) classification_prompt ChatPromptTemplate.from_messages([ (system, 你是一个问题分类器。请将用户问题分类为以下之一general_chat通用闲聊、domain_qa特定领域知识问答、research需要联网搜索的研究性问题。只返回分类标签不要解释。), (user, {question}) ]) classification_chain classification_prompt | classifier_llm | StrOutputParser() def classify_question(state: AgentState) - AgentState: 分类节点判断问题类型。 question state[question] category classification_chain.invoke({question: question}).strip().lower() # 确保分类结果在预期内 if category not in [general_chat, domain_qa, research]: category general_chat # 默认回退 return {category: category}# agents/researcher.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from tools.web_search import web_search_tool from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL # 初始化用于分析的模型例如 DeepSeek research_llm ChatOpenAI( modeldeepseek-chat, # 使用 deepseek-chat 模型名 base_urlDEEPSEEK_BASE_URL, api_keyDEEPSEEK_API_KEY, temperature0.2 ) def research_with_web(state: AgentState) - AgentState: 研究节点联网搜索并分析。 question state[question] # 1. 调用搜索工具 search_info web_search_tool.invoke(question) # 2. 用大模型分析搜索结果 analysis_prompt ChatPromptTemplate.from_messages([ (system, 你是一个研究助手。请基于以下搜索信息清晰、有条理地回答用户的问题。如果信息不足请指出。), (user, f问题{question}\n\n搜索信息{search_info}) ]) analysis_chain analysis_prompt | research_llm | StrOutputParser() answer analysis_chain.invoke({}) return {search_results: search_info, answer: answer}5.4 构建 RAG 知识库节点这部分需要先建立向量数据库。假设我们已有一些领域文档。# 假设在 graphs/main_workflow.py 中初始化 RAG from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from config import OPENAI_API_KEY import os # 1. 加载文档并构建向量库 (仅首次运行需要) def build_knowledge_base(): loader TextLoader(./knowledge/domain_docs.txt, encodingutf-8) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) splits text_splitter.split_documents(documents) vectorstore Chroma.from_documents( documentssplits, embeddingOpenAIEmbeddings(api_keyOPENAI_API_KEY), persist_directory./chroma_db ) return vectorstore # 如果数据库已存在直接加载 if os.path.exists(./chroma_db): vectorstore Chroma( persist_directory./chroma_db, embedding_functionOpenAIEmbeddings(api_keyOPENAI_API_KEY) ) else: vectorstore build_knowledge_base() # 2. 定义 RAG 检索链 retriever vectorstore.as_retriever(search_kwargs{k: 3}) rag_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyOPENAI_API_KEY) rag_prompt ChatPromptTemplate.from_template( 你是一个专业助手请严格根据以下上下文信息回答问题。 如果上下文信息不足以回答问题请直接说“根据现有资料无法回答该问题”不要编造信息。 上下文 {context} 问题 {question} 请给出专业、准确的回答 ) def rag_qa_node(state: AgentState) - AgentState: RAG 问答节点从知识库检索并生成答案。 question state[question] docs retriever.invoke(question) context \n\n.join([doc.page_content for doc in docs]) rag_chain rag_prompt | rag_llm | StrOutputParser() answer rag_chain.invoke({context: context, question: question}) return {documents: [doc.page_content for doc in docs], answer: answer}5.5 定义通用聊天和答案格式化节点# agents/general_chat.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from config import OPENAI_API_KEY general_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7, api_keyOPENAI_API_KEY) def general_chat_node(state: AgentState) - AgentState: 通用聊天节点。 question state[question] prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好且乐于助人的助手。), (user, {question}) ]) chain prompt | general_llm | StrOutputParser() answer chain.invoke({question: question}) return {answer: answer} def format_answer_node(state: AgentState) - AgentState: 答案格式化节点可选用于最终处理。 # 这里可以做一些后处理比如添加来源、格式化等 answer state.get(answer, 未能生成答案。) formatted_answer f【智能助手回答】\n\n{answer}\n\n---\n*回答生成完毕* return {answer: formatted_answer}5.6 组装 StateGraph现在在graphs/main_workflow.py中将所有节点组装成完整的工作流图。# graphs/main_workflow.py from langgraph.graph import StateGraph, END from state.agent_state import AgentState from agents.classifier import classify_question from agents.general_chat import general_chat_node from agents.researcher import research_with_web from agents.general_chat import format_answer_node # 假设 rag_qa_node 定义在同一个文件或已导入 from . import rag_qa_node def route_question(state: AgentState) - str: 路由函数根据分类结果决定下一个节点。 category state.get(category) if category general_chat: return general_chat_node elif category domain_qa: return rag_qa_node elif category research: return research_node else: # 默认路由到通用聊天 return general_chat_node # 1. 创建状态图 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(classifier, classify_question) workflow.add_node(general_chat_node, general_chat_node) workflow.add_node(rag_qa_node, rag_qa_node) workflow.add_node(research_node, research_with_web) workflow.add_node(format_answer, format_answer_node) # 3. 设置入口点 workflow.set_entry_point(classifier) # 4. 添加条件边从分类器路由到不同处理节点 workflow.add_conditional_edges( classifier, route_question, # 路由决策函数 { general_chat_node: general_chat_node, rag_qa_node: rag_qa_node, research_node: research_node, } ) # 5. 添加固定边各处理节点完成后都流向格式化节点然后结束 workflow.add_edge(general_chat_node, format_answer) workflow.add_edge(rag_qa_node, format_answer) workflow.add_edge(research_node, format_answer) workflow.add_edge(format_answer, END) # 6. 编译图 app workflow.compile()6. 运行与测试工作流创建main.py来运行和测试我们构建的图。# main.py from graphs.main_workflow import app from state.agent_state import AgentState def test_workflow(): 测试工作流 test_questions [ 你好今天天气怎么样, # 预期general_chat 我们公司的年假政策是怎么规定的, # 预期domain_qa (假设知识库有) 请帮我分析一下近期人工智能芯片的发展趋势。, # 预期research ] for q in test_questions: print(f\n{*50}) print(f输入问题: {q}) # 初始化状态 initial_state: AgentState { question: q, category: None, documents: [], search_results: None, answer: None, messages: [] } # 执行图 final_state app.invoke(initial_state) print(f分类结果: {final_state[category]}) print(f最终答案: {final_state[answer][:200]}...) # 预览前200字符 if final_state.get(documents): print(f检索到 {len(final_state[documents])} 条相关文档。) if final_state.get(search_results): print(已执行联网搜索。) if __name__ __main__: test_workflow()运行python main.py你将看到工作流根据问题类型自动路由到不同的处理分支并生成相应答案。7. 进阶持久化、可视化与 API 服务7.1 状态持久化 (Checkpointer)对于长对话或需要中断恢复的任务状态持久化至关重要。LangGraph 提供了Checkpointer接口。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 创建 SQLite 检查点存储器 memory SqliteSaver.from_conn_string(:memory:) # 内存数据库生产环境用文件路径 # 在编译图时传入检查点存储器 app workflow.compile(checkpointermemory) # 使用 configurable 来区分不同会话 config {configurable: {thread_id: user_session_123}} initial_state {question: 第一个问题, ...} # 调用时传入 config状态会被自动保存 result1 app.invoke(initial_state, configconfig) # 后续调用可以基于上次的状态继续 result2 app.invoke({question: 第二个问题, ...}, configconfig)7.2 可视化 (LangGraph Studio)LangGraph Studio 是一个用于可视化、调试工作流的工具。通常通过 Docker 运行。docker run -d -p 8502:8502 langchain/langgraph-studio访问http://localhost:8502上传你编译好的app对象需要将其保存为.json文件即可看到工作流的图形化表示并可以单步调试。7.3 封装为 API 服务要将此工作流部署为服务可以使用 FastAPI。# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from graphs.main_workflow import app from state.agent_state import AgentState import uuid app_fastapi FastAPI(title智能体工作流 API) class QuestionRequest(BaseModel): question: str session_id: str None # 支持多轮对话会话ID class AnswerResponse(BaseModel): session_id: str category: str answer: str documents: list [] search_performed: bool False app_fastapi.post(/ask, response_modelAnswerResponse) async def ask_question(req: QuestionRequest): try: session_id req.session_id or str(uuid.uuid4()) config {configurable: {thread_id: session_id}} initial_state: AgentState { question: req.question, category: None, documents: [], search_results: None, answer: None, messages: [] # 实际应用中可以从检查点恢复历史消息 } final_state app.invoke(initial_state, configconfig) return AnswerResponse( session_idsession_id, categoryfinal_state.get(category, unknown), answerfinal_state.get(answer, ), documentsfinal_state.get(documents, []), search_performedfinal_state.get(search_results) is not None ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app_fastapi, host0.0.0.0, port8000)运行python api_server.py即可通过http://localhost:8000/docs查看并测试 API。8. 资源占用、性能观察与优化由于 LangGraph 是编排框架其资源占用主要取决于接入的模型调用 GPT-4 与调用本地 7B 模型成本与延迟天差地别。工具执行联网搜索、数据库查询等 I/O 操作。状态存储如果使用持久化检查点数据库 I/O 会有开销。性能观察点延迟分解使用langsmithLangChain 官方跟踪平台可以清晰地看到每个节点LLM 调用、工具执行的耗时。Token 消耗在调用 OpenAI 等按 Token 计费的模型时需要在节点中记录输入输出 Token 数用于成本核算。并发与队列对于高并发 API 服务需要考虑使用asyncio或任务队列如 Celery来异步执行图避免阻塞。优化建议节点异步化如果节点是 I/O 密集型如网络请求将其定义为async函数并在图中使用异步调用。缓存对频繁且结果不变的 LLM 调用或检索结果实施缓存如langchain.cache。简化状态只保留必要的字段在 State 中过大的状态会影响序列化/反序列化性能。模型降级在非关键节点使用更小、更快的模型如 GPT-3.5 Turbo 代替 GPT-4。9. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 LangGraph 失败版本不兼容或未安装pip listgrep langgraph 检查版本app.invoke()报状态字段错误State TypedDict 定义与节点返回值不匹配检查节点函数返回的字典键名是否与AgentState定义一致确保节点返回{key: value}中的key是 State 中定义的字段条件路由不生效route_question函数返回值与add_conditional_edges中映射的节点名不一致打印route_question函数的返回值确保返回值字符串与add_node时注册的节点名完全一致RAG 检索结果不准嵌入模型不合适、分块策略不佳、检索参数 k 太小检查向量库中文档内容、分块大小和重叠度调整文本分割器参数、尝试不同的嵌入模型、增大检索数量 kAPI 服务并发报错图应用 (app) 非线程安全或模型 API 调用超限查看错误日志是否为线程冲突或 API 限流1. 为每个请求创建新的图实例开销大。2. 使用锁机制。3. 在 API 层实现请求队列。检查点 (Checkpointer) 无法保存数据库连接失败或状态对象不可序列化检查数据库路径权限尝试序列化状态对象确保状态中所有值都是可 JSON 序列化的使用文件路径而非:memory:测试LangGraph Studio 无法加载图保存的.json文件格式不对或版本不兼容确认保存图的代码正确app.get_graph().to_json()使用与 Studio 兼容的版本并严格按照文档导出图结构10. 最佳实践与项目落地建议始于简单迭代复杂不要一开始就设计包含 10 个节点的复杂图。从一个简单的两节点图如分类 → 回答开始验证通后再逐步添加分支和工具。为每个节点编写单元测试每个节点函数如classify_question都应该是纯函数逻辑易于单独测试。Mock 掉 LLM 和工具调用测试业务逻辑。使用 Configuration 管理参数将模型类型、API Key、温度等参数通过配置文件或环境变量管理不要硬编码在节点中。实现完整的日志与监控在每个节点的入口和出口记录日志包括输入、输出、耗时和错误。集成langsmith进行可视化跟踪。设计可复用的子图 (Subgraph)对于复杂的、重复使用的逻辑如一个完整的“检索-生成”流程可以将其封装成一个子图在主图中引用保持结构清晰。安全第一对于可以执行外部操作的工具写文件、发邮件、执行代码必须实现严格的用户授权和输入清洗。考虑在工具前增加一个“审批节点”。制定回滚和人工接管策略对于关键业务流当 AI 无法处理或置信度低时应能平滑地转移到人工处理流程。这个架构的核心价值在于它将 AI 应用的“逻辑”和“状态”清晰地抽象出来让你能用代码像设计流程图一样设计智能体的行为。从简单的分类路由到成百上千个智能体协作的仿真系统其底层模式是一致的。掌握 LangGraph 的状态管理和图编排思想是构建下一代可维护、可扩展 AI 应用的关键一步。建议从本文的示例项目开始亲手运行并修改每一个节点体会状态如何流动这是理解整个框架最快的方式。