尧图网页设计 
在 Agent 应用真正进入企业级落地之后很多人会发现一个尴尬的问题单 Agent 能力再强也扛不住复杂业务链路。过去我们习惯把 LangChain 当作“大模型工具箱”把多个工具塞给同一个模型但当任务包含多个状态、需要多角色协作、涉及多轮决策回退时这套思路会越来越难维护。LangGraph 的出现让 LangChain 从“工具链”升级成了一套真正面向 Agent 状态的编排框架再叠加 LangChain V1.x 对核心 API 的收敛和重写开发体验和落地方式都与 0.x 时代有了很大区别。这篇文章不会只停留在概念层面而是会从背景、核心组件、多智能体架构、完整可运行代码、常见坑位、工程实践这几个角度带你把 LangChain V1.x LangGraph 这条主路径吃透。适合已经写过几个 LangChain 脚本、想往 Agent 工程化方向走的开发者也适合刚从 0.x 迁移过来、被 API 变化搞得一头雾水的朋友。1. LangChain V1.x 与 LangGraph先搞清楚这一代的变化逻辑1.1 LangChain 到底是什么V1.x 又变了什么简单来说LangChain 是一套帮助开发者把大模型接入业务代码的“胶水层”。它把模型调用、提示词管理、工具调用、外部文档检索、记忆、Agent 决策等能力封装成模块让你不必每次从裸 API 开始手写整套逻辑。在 0.x 时代LangChain 的核心抽象是 Chain。开发者用 LLMChain、ConversationChain 等方式串联调用流程。但随着业务变复杂Chain 的缺点越来越明显链和链之间难以精细控制状态流转分支判断不直观多轮 Agent 内部循环一旦复杂起来调试成本非常高。到了 V1.xLangChain 做的事情更像一次“重新整理”。几个值得留意的方向核心包结构更清晰。langchain-core承接基础抽象langgraph成为状态编排层模型、工具、向量库等以独立包方式使用。原先大量chain概念逐步收敛为“自定义代码 LangGraph 消息循环 可复用组件”。对 OpenAI、Anthropic 等模型厂商的封装统一走langchain-openai这样的独立包避免大杂烩。Runnable接口仍是核心抽象但状态流转层面的工作更多交给 LangGraph。所以如果你看到网上很多资料还在教LLMChain需要注意那可能至少是 0.2 甚至更早的写法。现在更推荐的方式是直接用模型对象、消息对象和 LangGraph 来组织业务逻辑。1.2 LangGraph 在 LangChain 生态中的定位LangGraph 是一个基于图状态的 Agent 编排框架。你可以把它理解成“给大模型应用画的流程图”其中节点是一个个执行函数边表示执行顺序条件边表示根据某个结果决定下一步走向整张图维护一份全局 State。这样设计的价值在于Agent 的循环动作都显式表达不再隐晦地藏在一个 AgentExecutor 内部。多智能体协作有了自然的实现方式比如 Supervisor 模式、Handoff 模式。有持久化和断点续跑能力适合异步任务和人工审核这类真实生产场景。在 LangChain V1.x 架构里LangGraph 已经不是“插件的选项”而更像默认推荐的 Agent 运行时。如果你想构建一个有一定状态、有工具调用、还可能多个 Agent 协作的系统LangGraph 是目前最合理的一层。1.3 什么时候你才需要多智能体架构单 Agent 并不一定不好。很多时候一个 Agent 加多个工具完全能解决问题。多智能体架构并不是为了显得“高级”而是为了应对下面这些真实痛点职责隔离同时处理代码生成、SQL 查询、报告撰写让一个 Agent 全做完提示词会互相污染。上下文长度多轮对话场景下单个 Agent 的上下文可能被撑爆拆成多个 Agent 可以各自维护小状态。专业角色每个 Agent 拥有独立的系统提示词和工具集行为更可控。异步与审批不同 Agent 完成不同阶段任务中间可以插入人工审核。反过来如果你只是做一个“回答问题”的小工具引入多智能体只会增加延迟和成本。架构复杂度要用在真实的业务复杂度上。2. 环境准备与版本说明版本变化比较快下面配置以当前主流的 LangChain V1.x 系列为例。你在实际操作时建议先建一个干净的虚拟环境避免和其他项目的依赖冲突。2.1 创建虚拟环境python -m venv .venv source .venv/bin/activate推荐使用 Python 3.10 及以上版本过低的 Python 版本会影响较新版本库的安装。2.2 安装核心依赖pip install -U langchain langgraph langchain-openai如果要做文档检索再安装向量库相关依赖pip install langchain-chroma chromadb也可以把依赖写入 requirements.txtlangchain1.0,2.0 langgraph0.6,1.0 langchain-openai1.0 python-dotenv1.0然后执行pip install -r requirements.txt2.3 配置模型 API先用环境变量方式配置模型密钥。在项目根目录创建.env文件内容参考如下OPENAI_API_KEY你的密钥 OPENAI_BASE_URLhttps://你的网关地址/v1注意不要直接把密钥写进代码并提交到 Git 仓库。团队协作时可以用密钥管理平台或环境变量注入本地开发用.env配合python-dotenv即可。然后写一个公共模型客户端模块# model.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_chat_model(): return ChatOpenAI( modelgpt-4o-mini, temperature0.2, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), )这里把base_url留成可配置项是因为不少企业会用兼容 OpenAI 协议的私有网关。如果直接用官方接口不配置也完全没有问题。3. LangChain V1.x 核心组件拆解3.1 模型与消息在 V1.x 中消息类型的体系非常统一。一个正常的模型调用通常涉及三种角色消息from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from model import get_chat_model chat get_chat_model() messages [ SystemMessage(content你是一名资深Python开发工程师回答要简洁且务实。), HumanMessage(content请解释LangGraph中StateGraph的基本概念。), ] resp chat.invoke(messages) print(resp.content)注意invoke的参数要么是字符串提示词要么是消息列表。实际项目中更推荐使用消息列表因为消息列表可以完整保留多轮对话上下文也方便 LangGraph 做消息追加。3.2 工具定义工具是 Agent 的重要能力来源。LangChain 中定义一个工具最简单的方式是给函数加tool装饰器from langchain_core.tools import tool tool def get_server_status(server_name: str) - str: 查询指定服务器的运行状态。 Args: server_name: 服务器名称例如 api-server-01 # 真实环境这里应该去调用CMDB或监控系统 status_map { api-server-01: running, worker-server-02: down, } return status_map.get(server_name, unknown)工具函数的类型注解和 docstring 非常重要因为大模型会通过这些信息来决定什么时候调用、参数填什么。省略类型注解会让工具调用准确率明显下降。再看一个稍微复杂一点的工具示例import json import requests from langchain_core.tools import tool tool def query_order(order_id: str) - str: 根据订单号查询订单最新状态。 Args: order_id: 订单ID例如 ORD202506001 # 示例对接内部订单系统接口 url fhttps://internal.example.com/api/orders/{order_id} resp requests.get(url, timeout5) if resp.status_code 200: return json.dumps(resp.json(), ensure_asciiFalse) return 订单查询失败3.3 提示词模板与结构化输出在企业落地时结构化的输出往往比自由文本更实用。LangChain 中可以用with_structured_output让模型按 Pydantic 模型输出from pydantic import BaseModel, Field from langchain_core.messages import HumanMessage from model import get_chat_model class BugReport(BaseModel): severity: str Field(description严重程度P0/P1/P2/P3) component: str Field(description故障组件) summary: str Field(description一句话总结) suggestion: str Field(description修复建议) chat get_chat_model() structured_model chat.with_structured_output(BugReport) resp structured_model.invoke( 接口 /api/login 在并发200时出现大量502超时当前线上服务不稳定。 ) print(resp.severity) print(resp.component) print(resp.summary) print(resp.suggestion)这种方式特别适合后续把 Agent 结果直接转成告警工单、数据库记录或者发送给下游系统。3.4 Agent 与 LangGraph 的关系很多刚从 0.x 迁移过来的同学会有疑问LangChain 自己的 Agent 和 LangGraph 有什么区别简单理解LangChain V1.x 已经明确 LangGraph 作为 Agent 的运行时。你可以继续使用 LangChain 提供的高层 Agent 创建方法它底层其实也是把模型、工具、消息队列包装成语义图。如果只是做简单工具调用高层方法就行如果要做分支、循环、人工审批、多角色协作直接写 LangGraph 会更清楚。4. 多智能体架构的核心模式与原理4.1 为什么多智能体需要用“图”来编排多智能体不是简单地“创建多个 Agent 然后调来调去”核心问题是控制权转移。以常见的 Supervisor 模式为例一个主控 Agent 负责接收用户请求判断应该由哪个子 Agent 处理。子 Agent 完成自己的专业任务后把结果交回给主控。如果中间出现了子 Agent 无法解决的问题主控需要重新做任务规划。这种流程用代码 if-else 写一旦任务类型增多逻辑会非常混乱。而 LangGraph 把每一步抽象为节点把“下一个节点是谁”的决策作为一条条件边整张图就能变得清晰可维护。4.2 三种常见的多智能体架构4.2.1 Supervisor 模式一个总控 Agent多个子 Agent。总控查看任务进度动态决定将任务分配给谁。适用于任务类别分明、每个 Agent 有独立工具集的场景。用户请求 ↓ Supervisor Agent ↓ 分流 Researcher Agent → 完成 Coder Agent → 完成 Reviewer Agent → 完成 ↓ 汇总结果4.2.2 Handoff 模式Agent 之间直接转交控制权。比如客服机器人先由前台 Agent 接单发现需要退款时直接转交给售后 Agent售后 Agent 处理后如果涉及技术问题再转交技术 Agent。这种模式更适合“流程固定、单人负责到底”的业务。4.2.3 Hierarchical 模式多个团队形成层级关系高层 Agent 管理底层 Agent 组。适合大型企业系统例如一个“项目经理 Agent”管理着“后端研发 Agent”“前端研发 Agent”“测试 Agent”。层级关系可以降低单个图的复杂度但也会增加延迟。4.3 LangGraph 状态机的核心机制LangGraph 的核心是 State。你可以自定义状态的结构每个节点执行完会返回一份字典LangGraph 会把返回值合并进全局 State。一个最简单的 StateGraphfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): total: int def add_one(state: State): return {total: state[total] 1} def add_two(state: State): return {total: state[total] 2} graph_builder StateGraph(State) graph_builder.add_node(add_one, add_one) graph_builder.add_node(add_two, add_two) graph_builder.add_edge(START, add_one) graph_builder.add_edge(add_one, add_two) graph_builder.add_edge(add_two, END) graph graph_builder.compile() result graph.invoke({total: 0}) print(result)执行顺序是开始 → add_one → add_two → 结束。最终状态为{total: 3}。这里START和END是 LangGraph 内置的特殊节点分别表示入口和出口。节点名可以自定义但建议用语义化的字符串方便后续调试和查看轨迹。5. 完整代码实战构建企业级多智能体系统下面用一个贴近真实业务的场景来演示目标是把一个“IT 工单智能处理系统”落地为 LangGraph 多智能体应用。系统包含三个子 Agent日志分析 Agent分析故障日志定位错误类型。数据库诊断 Agent检查慢查询和数据库状态。工单回复 Agent汇总信息并生成处理建议。生产环境中的日志数据获取、数据库指标查询应该接真实的监控系统这里为了聚焦 LangGraph 代码本身用模拟函数代替但代码结构和调用方式与真实系统一致。5.1 项目结构agent-demo/ ├── .env ├── requirements.txt ├── model.py ├── tools.py ├── agents.py ├── graph.py └── main.py5.2 定义公共状态# graph.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] logs: str db_metrics: str final_answer: str next: strAnnotated[list, add_messages]表示 messages 字段使用 LangGraph 内置的追加逻辑每次节点返回新消息时会自动合并到已有列表而不是整体覆盖。这种写法在 Agent 消息流中一定要保留。其他字段如logs、db_metrics是普通字段返回时会直接覆盖。5.3 定义工具节点# tools.py import random from datetime import datetime from langchain_core.tools import tool tool def get_recent_error_logs(service: str) - str: 获取指定服务最近一小时的错误日志。 Args: service: 服务名称如 order-service logs [ [ERROR] connection pool exhausted, wait timeout 3000ms, [ERROR] Redis cluster max memory reached, [WARN] retry 2 times, then success, ] return f{service} 最近日志:\n \n.join(logs) tool def query_slow_queries(db_name: str) - str: 查询指定数据库近一小时的慢查询。 Args: db_name: 数据库名称如 order_db count random.randint(1, 10) return f{db_name} 慢查询数量: {count}最长耗时: 2.3s5.4 实现各智能体节点# agents.py from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from model import get_chat_model from tools import get_recent_error_logs, query_slow_queries from langgraph.prebuilt import ToolNode chat get_chat_model() log_analyzer_tools [get_recent_error_logs] db_analyzer_tools [query_slow_queries] log_analyzer_tool_node ToolNode(log_analyzer_tools) db_analyzer_tool_node ToolNode(db_analyzer_tools) LOG_ANALYZER_SYSTEM 你是一名SRE日志分析工程师。需要调用工具获取日志识别异常原因。 分析结果保持结构化输出三部分错误类型、可能原因、疑似组件。 DB_ANALYZER_SYSTEM 你是一名数据库DBA。需要调用工具查询慢SQL和数据库状态。 分析结果输出三部分数据库风险、SQL建议、优先级。 REPLY_AGENT_SYSTEM 你是一名工单回复工程师。根据前序分析结果输出最终解决方案。 内容面向值班人员要求简洁、可执行不超过200字。 def log_analyzer_node(state): messages [ SystemMessage(contentLOG_ANALYZER_SYSTEM), *state[messages], ] response chat.invoke(messages) return {messages: [response], logs: response.content} def db_analyzer_node(state): messages [ SystemMessage(contentDB_ANALYZER_SYSTEM), *state[messages], ] response chat.invoke(messages) return {messages: [response], db_metrics: response.content} def reply_node(state): combined_info f日志分析结果:\n{state.get(logs, )}\n\n数据库诊断结果:\n{state.get(db_metrics, )} messages [ SystemMessage(contentREPLY_AGENT_SYSTEM), HumanMessage(contentcombined_info), ] response chat.invoke(messages) return {messages: [response], final_answer: response.content}这里有一个容易踩坑的地方state[messages]中可能包含前面 Agent 生成的中间结果直接用扩展运算符拼接时要确保节点之间消息类型兼容。上面示例为了让三个 Agent 的职责隔离更清晰第二个、第三个 Agent 使用的是前序节点写入的自定义字段而不是把全部对话历史重放一遍这样可以显著减少模型上下文体积。5.5 用 Supervisor 节点做任务路由在真实项目中并非每个工单都需要走完三个 Agent。我们可以加一个 Router Agent根据工单内容决定是否需要数据库诊断。def supervisor_node(state): last_message state[messages][-1].content if state[messages] else prompt f根据工单内容决定处理路径。 工单内容: {last_message} 如果涉及数据库、慢SQL、锁表输出 ONLY_DB。 如果涉及日志分析、服务异常、报错排查输出 ONLY_LOG。 如果两者都涉及输出 BOTH。 现在请直接输出一个词。 resp chat.invoke([HumanMessage(contentprompt)]) decision resp.content.strip().upper() if ONLY_DB in decision: return {next: db_analyzer} elif ONLY_LOG in decision: return {next: log_analyzer} else: return {next: both}这个 Supervisor 节点本身没有使用工具它只是一个决策函数。判决结果写入next字段供条件边读取。5.6 组装 LangGraph# graph.py from langgraph.graph import StateGraph, START, END from agents import ( log_analyzer_node, db_analyzer_node, reply_node, supervisor_node, ) from agents import log_analyzer_tool_node, db_analyzer_tool_node def build_graph(): builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(log_analyzer, log_analyzer_node) builder.add_node(db_analyzer, db_analyzer_node) builder.add_node(reply, reply_node) builder.add_node(log_tools, log_analyzer_tool_node) builder.add_node(db_tools, db_analyzer_tool_node) builder.add_edge(START, supervisor) # supervisor 之后走条件路由 builder.add_conditional_edges( supervisor, lambda state: state[next], { only_log: log_analyzer, only_db: db_analyzer, both: log_analyzer, }, ) # 日志分析 Agent 需要执行工具 builder.add_edge(log_analyzer, log_tools) builder.add_edge(log_tools, db_analyzer) # 数据库分析 Agent 需要执行工具 builder.add_edge(db_analyzer, db_tools) builder.add_edge(db_tools, reply) builder.add_edge(reply, END) return builder.compile()注意这里为了方便演示把条件路由的目标做了简化。如果想在 “both” 下同时执行两个 Agent更好的做法是先跑日志分析再跑数据库分析或者拆出两个并行分支。LangGraph 本身支持并行分支但那需要更稳妥的状态合并策略真实项目要按任务依赖关系决定。生产项目中更常见的做法是把工具执行和 Agent 节点分开Agent 节点先把 “需要调用工具” 的信息写入状态ToolNode 统一执行然后再把工具结果返回给同一个 Agent 节点循环。你可以用下图理解这个循环Agent 节点 - Tool 节点 - 继续 Agent 节点判断是否继续调用工具 ↓ 不再需要工具 回复/结束如果你希望 Agent 能多次调用工具建议用langgraph.prebuilt里的create_react_agent或者自己写带 LLM 循环的节点。5.7 运行入口# main.py from graph import build_graph from langchain_core.messages import HumanMessage def main(): app build_graph() inputs { messages: [HumanMessage(content订单服务频繁超时日志显示连接池耗尽同时订单库有大量慢查询。)], logs: , db_metrics: , final_answer: , next: , } result app.invoke(inputs) print(\n 最终工单回复 ) print(result[final_answer]) if __name__ __main__: main()运行命令python main.py预期输出会是一段由回复 Agent 生成的处理建议。由于大模型输出非确定性每次结果不会完全一样但结构应该保持稳定。5.8 扩展加入人工审核断点企业级系统往往需要人工确认后 Agent 才能执行最终变更。LangGraph 支持断点机制from langgraph.checkpoint.memory import MemorySaver def build_graph_with_checkpoint(): builder StateGraph(AgentState) # ... 添加节点和边的逻辑与上面相同 ... graph builder.compile(checkpointerMemorySaver()) return graph app build_graph_with_checkpoint() config {configurable: {thread_id: ticket-001}} result app.invoke(inputs, config)每次执行时传入同一个thread_idLangGraph 就能记住该会话的完整状态之后可以从中途继续执行。这个机制在生产环境非常有价值比如插入人工审核后审核通过再更新状态继续往下走。如果要持久化可以换成SqliteSaver或PostgresSaver避免进程重启后状态丢失。6. 常见问题与排查思路6.1 问题表格问题现象常见原因解决思路安装失败或版本冲突langchain、langgraph、pydantic 版本不兼容使用虚拟环境统一安装先升级 pipinvoke 报 TypeError传参格式不对传入普通字符串而不是消息列表检查参数类型统一使用消息列表工具未被调用docstring 不清晰或缺少类型注解补充详细功能描述和参数类型工具调用后 Agent 仍认为没有结果ToolNode 的返回没有正确合并进状态确认 ToolNode 返回格式检查 State 的 reducerLangGraph 执行次数过多Agent 陷入工具循环在节点中加入最大循环次数限制多分支结果互相覆盖多个并行节点同时写入同一字段使用不同的状态字段或定义自定义 reducer无法从断点继续执行没有配置 checkpointer加入 MemorySaver 或持久化 checkpoint模型上下文过大每次节点都把全部历史消息传入按节点职责过滤历史只使用自定义字段6.2 Agent 陷入循环的解决方案工具调用循环是 Agent 开发中最常见的问题之一。示例写法如下from langgraph.graph import StateGraph, START, END MAX_ROUNDS 5 def resilient_node(state): round state.get(round, 0) 1 if round MAX_ROUNDS: return {messages: [AIMessage(content已达到最大执行轮数请人工介入。)], round: round, next: END} return {messages: [...], round: round, next: tools}强烈建议所有工具型 Agent 节点都设置最大轮数。不要指望模型“自己知道什么时候应该停止”复杂任务中模型很容易反复尝试同一个工具。6.3 如何调试 LangGraph 图LangGraph 提供了可视化编译图的能力但为了让文章保持代码可复制性这里只介绍控制台调试方法# 在 build_graph 编译后打印图结构 for node_name in graph.get_graph().nodes: print(节点:, node_name) for edge in graph.get_graph().edges: print(边:, edge)更实际的办法是在每个节点函数里打印状态字段的关键内容def log_analyzer_node(state): print( 进入日志分析节点) print( 当前消息数:, len(state[messages])) # 原有逻辑...7. 企业级 Agent 落地的最佳实践7.1 状态设计字段要小职责要清多智能体系统的状态字段要克制。不要把不需要共享的临时变量全部放进 State。字段越多状态合并越复杂checkpoint 存储成本也越高。建议每个 Agent 只把自己真正要输出的结构化结果写入共享字段。7.2 工具设计可观测、可降级工具是 Agent 接触外部系统的入口必须做好日志和安全控制。所有工具调用记录日志包括入参、出参、耗时、调用方。外部接口调用设置超时避免 Agent 长时间等待。对写操作类工具默认不直接执行先返回“将要执行的 SQL”给人工确认。最小权限原则Agent 使用的数据库账号、云平台密钥权限只需要最低限度即可。7.3 提示词与上下文管理V1.x 中虽然消息列表是标准做法但多智能体场景下不建议把每个 Agent 的完整对话历史都传给其他 Agent。每个专业 Agent 应该只拿它需要的信息。上面的实战代码里reply 节点使用的是前序节点输出的分析文本而不是原始日志或完整 messages这种方式在真实环境中能明显降低 token 成本和无关信息干扰。7.4 安全边界企业级 Agent 系统最重要的合规点API Key 不允许出现在代码仓库中。Agent 不能拥有直接删除生产数据的工具。如果业务确实需要工具内部必须增加二次确认参数。审计日志完整记录 Agent 的每一次决策和工具调用。特别是多智能体系统中角色权限不同越权调用其他 Agent 的工具会带来隐患。对于涉及用户个人信息的数据Agent 节点要做脱敏处理后再传给大模型。7.5 性能与成本优化优先使用更小的模型做路由和分类大模型只做关键推理。对不常变化的知识片段使用 RAG 缓存。将 LangGraph 的长连接任务异步化避免阻塞 HTTP 请求线程。如果业务允许考虑对重复性工单结果做缓存大幅节省模型调用成本。8. 总结与学习路线这篇文章从 LangChain V1.x 核心组件讲到 LangGraph 状态编排再落到一个可运行的多智能体工单处理系统核心是想让大家建立一套新的开发心智不要再把 Agent 看成“一个会调用函数的模型”而要把它看成一张有状态、有分支、有断点、可恢复的图。学习顺序建议这样安排先熟练掌握langchain-core的模型、消息、工具、结构化输出。再学习 LangGraph 的状态、节点、边、条件边把所有概念在自己的小 Demo 中跑一遍。接着从单 Agent 升级到多 Agent先实现 Supervisor 模式再尝试 Handoff 和 Hierarchical 模式。加入持久化和断点机制模拟人工审核流程。最后做工程化改造日志、限流、密钥管理、权限控制、监控告警。如果你现在正处在从 LangChain 0.x 往 V1.x 迁移的阶段建议不要逐行改代码而是先按本文的架构重新梳理业务边界。绝大多数旧项目的核心难点不是 API 改名而是状态流和职责边界没理清。LangGraph 只是把这个问题从“隐藏的隐式逻辑”变成了“显式的图结构”这也正是它对企业级项目最有价值的地方。如果这篇文章对你理解 LangChain V1.x 和 LangGraph 有帮助可以收藏备用。后续我也会继续补充分布式多智能体部署、Agent 可观测性、RAG 与多智能体组合实战等话题。 LangChain V1.x + LangGraph 多智能体编排工程实战
在 Agent 应用真正进入企业级落地之后很多人会发现一个尴尬的问题单 Agent 能力再强也扛不住复杂业务链路。过去我们习惯把 LangChain 当作“大模型工具箱”把多个工具塞给同一个模型但当任务包含多个状态、需要多角色协作、涉及多轮决策回退时这套思路会越来越难维护。LangGraph 的出现让 LangChain 从“工具链”升级成了一套真正面向 Agent 状态的编排框架再叠加 LangChain V1.x 对核心 API 的收敛和重写开发体验和落地方式都与 0.x 时代有了很大区别。这篇文章不会只停留在概念层面而是会从背景、核心组件、多智能体架构、完整可运行代码、常见坑位、工程实践这几个角度带你把 LangChain V1.x LangGraph 这条主路径吃透。适合已经写过几个 LangChain 脚本、想往 Agent 工程化方向走的开发者也适合刚从 0.x 迁移过来、被 API 变化搞得一头雾水的朋友。1. LangChain V1.x 与 LangGraph先搞清楚这一代的变化逻辑1.1 LangChain 到底是什么V1.x 又变了什么简单来说LangChain 是一套帮助开发者把大模型接入业务代码的“胶水层”。它把模型调用、提示词管理、工具调用、外部文档检索、记忆、Agent 决策等能力封装成模块让你不必每次从裸 API 开始手写整套逻辑。在 0.x 时代LangChain 的核心抽象是 Chain。开发者用 LLMChain、ConversationChain 等方式串联调用流程。但随着业务变复杂Chain 的缺点越来越明显链和链之间难以精细控制状态流转分支判断不直观多轮 Agent 内部循环一旦复杂起来调试成本非常高。到了 V1.xLangChain 做的事情更像一次“重新整理”。几个值得留意的方向核心包结构更清晰。langchain-core承接基础抽象langgraph成为状态编排层模型、工具、向量库等以独立包方式使用。原先大量chain概念逐步收敛为“自定义代码 LangGraph 消息循环 可复用组件”。对 OpenAI、Anthropic 等模型厂商的封装统一走langchain-openai这样的独立包避免大杂烩。Runnable接口仍是核心抽象但状态流转层面的工作更多交给 LangGraph。所以如果你看到网上很多资料还在教LLMChain需要注意那可能至少是 0.2 甚至更早的写法。现在更推荐的方式是直接用模型对象、消息对象和 LangGraph 来组织业务逻辑。1.2 LangGraph 在 LangChain 生态中的定位LangGraph 是一个基于图状态的 Agent 编排框架。你可以把它理解成“给大模型应用画的流程图”其中节点是一个个执行函数边表示执行顺序条件边表示根据某个结果决定下一步走向整张图维护一份全局 State。这样设计的价值在于Agent 的循环动作都显式表达不再隐晦地藏在一个 AgentExecutor 内部。多智能体协作有了自然的实现方式比如 Supervisor 模式、Handoff 模式。有持久化和断点续跑能力适合异步任务和人工审核这类真实生产场景。在 LangChain V1.x 架构里LangGraph 已经不是“插件的选项”而更像默认推荐的 Agent 运行时。如果你想构建一个有一定状态、有工具调用、还可能多个 Agent 协作的系统LangGraph 是目前最合理的一层。1.3 什么时候你才需要多智能体架构单 Agent 并不一定不好。很多时候一个 Agent 加多个工具完全能解决问题。多智能体架构并不是为了显得“高级”而是为了应对下面这些真实痛点职责隔离同时处理代码生成、SQL 查询、报告撰写让一个 Agent 全做完提示词会互相污染。上下文长度多轮对话场景下单个 Agent 的上下文可能被撑爆拆成多个 Agent 可以各自维护小状态。专业角色每个 Agent 拥有独立的系统提示词和工具集行为更可控。异步与审批不同 Agent 完成不同阶段任务中间可以插入人工审核。反过来如果你只是做一个“回答问题”的小工具引入多智能体只会增加延迟和成本。架构复杂度要用在真实的业务复杂度上。2. 环境准备与版本说明版本变化比较快下面配置以当前主流的 LangChain V1.x 系列为例。你在实际操作时建议先建一个干净的虚拟环境避免和其他项目的依赖冲突。2.1 创建虚拟环境python -m venv .venv source .venv/bin/activate推荐使用 Python 3.10 及以上版本过低的 Python 版本会影响较新版本库的安装。2.2 安装核心依赖pip install -U langchain langgraph langchain-openai如果要做文档检索再安装向量库相关依赖pip install langchain-chroma chromadb也可以把依赖写入 requirements.txtlangchain1.0,2.0 langgraph0.6,1.0 langchain-openai1.0 python-dotenv1.0然后执行pip install -r requirements.txt2.3 配置模型 API先用环境变量方式配置模型密钥。在项目根目录创建.env文件内容参考如下OPENAI_API_KEY你的密钥 OPENAI_BASE_URLhttps://你的网关地址/v1注意不要直接把密钥写进代码并提交到 Git 仓库。团队协作时可以用密钥管理平台或环境变量注入本地开发用.env配合python-dotenv即可。然后写一个公共模型客户端模块# model.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_chat_model(): return ChatOpenAI( modelgpt-4o-mini, temperature0.2, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), )这里把base_url留成可配置项是因为不少企业会用兼容 OpenAI 协议的私有网关。如果直接用官方接口不配置也完全没有问题。3. LangChain V1.x 核心组件拆解3.1 模型与消息在 V1.x 中消息类型的体系非常统一。一个正常的模型调用通常涉及三种角色消息from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from model import get_chat_model chat get_chat_model() messages [ SystemMessage(content你是一名资深Python开发工程师回答要简洁且务实。), HumanMessage(content请解释LangGraph中StateGraph的基本概念。), ] resp chat.invoke(messages) print(resp.content)注意invoke的参数要么是字符串提示词要么是消息列表。实际项目中更推荐使用消息列表因为消息列表可以完整保留多轮对话上下文也方便 LangGraph 做消息追加。3.2 工具定义工具是 Agent 的重要能力来源。LangChain 中定义一个工具最简单的方式是给函数加tool装饰器from langchain_core.tools import tool tool def get_server_status(server_name: str) - str: 查询指定服务器的运行状态。 Args: server_name: 服务器名称例如 api-server-01 # 真实环境这里应该去调用CMDB或监控系统 status_map { api-server-01: running, worker-server-02: down, } return status_map.get(server_name, unknown)工具函数的类型注解和 docstring 非常重要因为大模型会通过这些信息来决定什么时候调用、参数填什么。省略类型注解会让工具调用准确率明显下降。再看一个稍微复杂一点的工具示例import json import requests from langchain_core.tools import tool tool def query_order(order_id: str) - str: 根据订单号查询订单最新状态。 Args: order_id: 订单ID例如 ORD202506001 # 示例对接内部订单系统接口 url fhttps://internal.example.com/api/orders/{order_id} resp requests.get(url, timeout5) if resp.status_code 200: return json.dumps(resp.json(), ensure_asciiFalse) return 订单查询失败3.3 提示词模板与结构化输出在企业落地时结构化的输出往往比自由文本更实用。LangChain 中可以用with_structured_output让模型按 Pydantic 模型输出from pydantic import BaseModel, Field from langchain_core.messages import HumanMessage from model import get_chat_model class BugReport(BaseModel): severity: str Field(description严重程度P0/P1/P2/P3) component: str Field(description故障组件) summary: str Field(description一句话总结) suggestion: str Field(description修复建议) chat get_chat_model() structured_model chat.with_structured_output(BugReport) resp structured_model.invoke( 接口 /api/login 在并发200时出现大量502超时当前线上服务不稳定。 ) print(resp.severity) print(resp.component) print(resp.summary) print(resp.suggestion)这种方式特别适合后续把 Agent 结果直接转成告警工单、数据库记录或者发送给下游系统。3.4 Agent 与 LangGraph 的关系很多刚从 0.x 迁移过来的同学会有疑问LangChain 自己的 Agent 和 LangGraph 有什么区别简单理解LangChain V1.x 已经明确 LangGraph 作为 Agent 的运行时。你可以继续使用 LangChain 提供的高层 Agent 创建方法它底层其实也是把模型、工具、消息队列包装成语义图。如果只是做简单工具调用高层方法就行如果要做分支、循环、人工审批、多角色协作直接写 LangGraph 会更清楚。4. 多智能体架构的核心模式与原理4.1 为什么多智能体需要用“图”来编排多智能体不是简单地“创建多个 Agent 然后调来调去”核心问题是控制权转移。以常见的 Supervisor 模式为例一个主控 Agent 负责接收用户请求判断应该由哪个子 Agent 处理。子 Agent 完成自己的专业任务后把结果交回给主控。如果中间出现了子 Agent 无法解决的问题主控需要重新做任务规划。这种流程用代码 if-else 写一旦任务类型增多逻辑会非常混乱。而 LangGraph 把每一步抽象为节点把“下一个节点是谁”的决策作为一条条件边整张图就能变得清晰可维护。4.2 三种常见的多智能体架构4.2.1 Supervisor 模式一个总控 Agent多个子 Agent。总控查看任务进度动态决定将任务分配给谁。适用于任务类别分明、每个 Agent 有独立工具集的场景。用户请求 ↓ Supervisor Agent ↓ 分流 Researcher Agent → 完成 Coder Agent → 完成 Reviewer Agent → 完成 ↓ 汇总结果4.2.2 Handoff 模式Agent 之间直接转交控制权。比如客服机器人先由前台 Agent 接单发现需要退款时直接转交给售后 Agent售后 Agent 处理后如果涉及技术问题再转交技术 Agent。这种模式更适合“流程固定、单人负责到底”的业务。4.2.3 Hierarchical 模式多个团队形成层级关系高层 Agent 管理底层 Agent 组。适合大型企业系统例如一个“项目经理 Agent”管理着“后端研发 Agent”“前端研发 Agent”“测试 Agent”。层级关系可以降低单个图的复杂度但也会增加延迟。4.3 LangGraph 状态机的核心机制LangGraph 的核心是 State。你可以自定义状态的结构每个节点执行完会返回一份字典LangGraph 会把返回值合并进全局 State。一个最简单的 StateGraphfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): total: int def add_one(state: State): return {total: state[total] 1} def add_two(state: State): return {total: state[total] 2} graph_builder StateGraph(State) graph_builder.add_node(add_one, add_one) graph_builder.add_node(add_two, add_two) graph_builder.add_edge(START, add_one) graph_builder.add_edge(add_one, add_two) graph_builder.add_edge(add_two, END) graph graph_builder.compile() result graph.invoke({total: 0}) print(result)执行顺序是开始 → add_one → add_two → 结束。最终状态为{total: 3}。这里START和END是 LangGraph 内置的特殊节点分别表示入口和出口。节点名可以自定义但建议用语义化的字符串方便后续调试和查看轨迹。5. 完整代码实战构建企业级多智能体系统下面用一个贴近真实业务的场景来演示目标是把一个“IT 工单智能处理系统”落地为 LangGraph 多智能体应用。系统包含三个子 Agent日志分析 Agent分析故障日志定位错误类型。数据库诊断 Agent检查慢查询和数据库状态。工单回复 Agent汇总信息并生成处理建议。生产环境中的日志数据获取、数据库指标查询应该接真实的监控系统这里为了聚焦 LangGraph 代码本身用模拟函数代替但代码结构和调用方式与真实系统一致。5.1 项目结构agent-demo/ ├── .env ├── requirements.txt ├── model.py ├── tools.py ├── agents.py ├── graph.py └── main.py5.2 定义公共状态# graph.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] logs: str db_metrics: str final_answer: str next: strAnnotated[list, add_messages]表示 messages 字段使用 LangGraph 内置的追加逻辑每次节点返回新消息时会自动合并到已有列表而不是整体覆盖。这种写法在 Agent 消息流中一定要保留。其他字段如logs、db_metrics是普通字段返回时会直接覆盖。5.3 定义工具节点# tools.py import random from datetime import datetime from langchain_core.tools import tool tool def get_recent_error_logs(service: str) - str: 获取指定服务最近一小时的错误日志。 Args: service: 服务名称如 order-service logs [ [ERROR] connection pool exhausted, wait timeout 3000ms, [ERROR] Redis cluster max memory reached, [WARN] retry 2 times, then success, ] return f{service} 最近日志:\n \n.join(logs) tool def query_slow_queries(db_name: str) - str: 查询指定数据库近一小时的慢查询。 Args: db_name: 数据库名称如 order_db count random.randint(1, 10) return f{db_name} 慢查询数量: {count}最长耗时: 2.3s5.4 实现各智能体节点# agents.py from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from model import get_chat_model from tools import get_recent_error_logs, query_slow_queries from langgraph.prebuilt import ToolNode chat get_chat_model() log_analyzer_tools [get_recent_error_logs] db_analyzer_tools [query_slow_queries] log_analyzer_tool_node ToolNode(log_analyzer_tools) db_analyzer_tool_node ToolNode(db_analyzer_tools) LOG_ANALYZER_SYSTEM 你是一名SRE日志分析工程师。需要调用工具获取日志识别异常原因。 分析结果保持结构化输出三部分错误类型、可能原因、疑似组件。 DB_ANALYZER_SYSTEM 你是一名数据库DBA。需要调用工具查询慢SQL和数据库状态。 分析结果输出三部分数据库风险、SQL建议、优先级。 REPLY_AGENT_SYSTEM 你是一名工单回复工程师。根据前序分析结果输出最终解决方案。 内容面向值班人员要求简洁、可执行不超过200字。 def log_analyzer_node(state): messages [ SystemMessage(contentLOG_ANALYZER_SYSTEM), *state[messages], ] response chat.invoke(messages) return {messages: [response], logs: response.content} def db_analyzer_node(state): messages [ SystemMessage(contentDB_ANALYZER_SYSTEM), *state[messages], ] response chat.invoke(messages) return {messages: [response], db_metrics: response.content} def reply_node(state): combined_info f日志分析结果:\n{state.get(logs, )}\n\n数据库诊断结果:\n{state.get(db_metrics, )} messages [ SystemMessage(contentREPLY_AGENT_SYSTEM), HumanMessage(contentcombined_info), ] response chat.invoke(messages) return {messages: [response], final_answer: response.content}这里有一个容易踩坑的地方state[messages]中可能包含前面 Agent 生成的中间结果直接用扩展运算符拼接时要确保节点之间消息类型兼容。上面示例为了让三个 Agent 的职责隔离更清晰第二个、第三个 Agent 使用的是前序节点写入的自定义字段而不是把全部对话历史重放一遍这样可以显著减少模型上下文体积。5.5 用 Supervisor 节点做任务路由在真实项目中并非每个工单都需要走完三个 Agent。我们可以加一个 Router Agent根据工单内容决定是否需要数据库诊断。def supervisor_node(state): last_message state[messages][-1].content if state[messages] else prompt f根据工单内容决定处理路径。 工单内容: {last_message} 如果涉及数据库、慢SQL、锁表输出 ONLY_DB。 如果涉及日志分析、服务异常、报错排查输出 ONLY_LOG。 如果两者都涉及输出 BOTH。 现在请直接输出一个词。 resp chat.invoke([HumanMessage(contentprompt)]) decision resp.content.strip().upper() if ONLY_DB in decision: return {next: db_analyzer} elif ONLY_LOG in decision: return {next: log_analyzer} else: return {next: both}这个 Supervisor 节点本身没有使用工具它只是一个决策函数。判决结果写入next字段供条件边读取。5.6 组装 LangGraph# graph.py from langgraph.graph import StateGraph, START, END from agents import ( log_analyzer_node, db_analyzer_node, reply_node, supervisor_node, ) from agents import log_analyzer_tool_node, db_analyzer_tool_node def build_graph(): builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(log_analyzer, log_analyzer_node) builder.add_node(db_analyzer, db_analyzer_node) builder.add_node(reply, reply_node) builder.add_node(log_tools, log_analyzer_tool_node) builder.add_node(db_tools, db_analyzer_tool_node) builder.add_edge(START, supervisor) # supervisor 之后走条件路由 builder.add_conditional_edges( supervisor, lambda state: state[next], { only_log: log_analyzer, only_db: db_analyzer, both: log_analyzer, }, ) # 日志分析 Agent 需要执行工具 builder.add_edge(log_analyzer, log_tools) builder.add_edge(log_tools, db_analyzer) # 数据库分析 Agent 需要执行工具 builder.add_edge(db_analyzer, db_tools) builder.add_edge(db_tools, reply) builder.add_edge(reply, END) return builder.compile()注意这里为了方便演示把条件路由的目标做了简化。如果想在 “both” 下同时执行两个 Agent更好的做法是先跑日志分析再跑数据库分析或者拆出两个并行分支。LangGraph 本身支持并行分支但那需要更稳妥的状态合并策略真实项目要按任务依赖关系决定。生产项目中更常见的做法是把工具执行和 Agent 节点分开Agent 节点先把 “需要调用工具” 的信息写入状态ToolNode 统一执行然后再把工具结果返回给同一个 Agent 节点循环。你可以用下图理解这个循环Agent 节点 - Tool 节点 - 继续 Agent 节点判断是否继续调用工具 ↓ 不再需要工具 回复/结束如果你希望 Agent 能多次调用工具建议用langgraph.prebuilt里的create_react_agent或者自己写带 LLM 循环的节点。5.7 运行入口# main.py from graph import build_graph from langchain_core.messages import HumanMessage def main(): app build_graph() inputs { messages: [HumanMessage(content订单服务频繁超时日志显示连接池耗尽同时订单库有大量慢查询。)], logs: , db_metrics: , final_answer: , next: , } result app.invoke(inputs) print(\n 最终工单回复 ) print(result[final_answer]) if __name__ __main__: main()运行命令python main.py预期输出会是一段由回复 Agent 生成的处理建议。由于大模型输出非确定性每次结果不会完全一样但结构应该保持稳定。5.8 扩展加入人工审核断点企业级系统往往需要人工确认后 Agent 才能执行最终变更。LangGraph 支持断点机制from langgraph.checkpoint.memory import MemorySaver def build_graph_with_checkpoint(): builder StateGraph(AgentState) # ... 添加节点和边的逻辑与上面相同 ... graph builder.compile(checkpointerMemorySaver()) return graph app build_graph_with_checkpoint() config {configurable: {thread_id: ticket-001}} result app.invoke(inputs, config)每次执行时传入同一个thread_idLangGraph 就能记住该会话的完整状态之后可以从中途继续执行。这个机制在生产环境非常有价值比如插入人工审核后审核通过再更新状态继续往下走。如果要持久化可以换成SqliteSaver或PostgresSaver避免进程重启后状态丢失。6. 常见问题与排查思路6.1 问题表格问题现象常见原因解决思路安装失败或版本冲突langchain、langgraph、pydantic 版本不兼容使用虚拟环境统一安装先升级 pipinvoke 报 TypeError传参格式不对传入普通字符串而不是消息列表检查参数类型统一使用消息列表工具未被调用docstring 不清晰或缺少类型注解补充详细功能描述和参数类型工具调用后 Agent 仍认为没有结果ToolNode 的返回没有正确合并进状态确认 ToolNode 返回格式检查 State 的 reducerLangGraph 执行次数过多Agent 陷入工具循环在节点中加入最大循环次数限制多分支结果互相覆盖多个并行节点同时写入同一字段使用不同的状态字段或定义自定义 reducer无法从断点继续执行没有配置 checkpointer加入 MemorySaver 或持久化 checkpoint模型上下文过大每次节点都把全部历史消息传入按节点职责过滤历史只使用自定义字段6.2 Agent 陷入循环的解决方案工具调用循环是 Agent 开发中最常见的问题之一。示例写法如下from langgraph.graph import StateGraph, START, END MAX_ROUNDS 5 def resilient_node(state): round state.get(round, 0) 1 if round MAX_ROUNDS: return {messages: [AIMessage(content已达到最大执行轮数请人工介入。)], round: round, next: END} return {messages: [...], round: round, next: tools}强烈建议所有工具型 Agent 节点都设置最大轮数。不要指望模型“自己知道什么时候应该停止”复杂任务中模型很容易反复尝试同一个工具。6.3 如何调试 LangGraph 图LangGraph 提供了可视化编译图的能力但为了让文章保持代码可复制性这里只介绍控制台调试方法# 在 build_graph 编译后打印图结构 for node_name in graph.get_graph().nodes: print(节点:, node_name) for edge in graph.get_graph().edges: print(边:, edge)更实际的办法是在每个节点函数里打印状态字段的关键内容def log_analyzer_node(state): print( 进入日志分析节点) print( 当前消息数:, len(state[messages])) # 原有逻辑...7. 企业级 Agent 落地的最佳实践7.1 状态设计字段要小职责要清多智能体系统的状态字段要克制。不要把不需要共享的临时变量全部放进 State。字段越多状态合并越复杂checkpoint 存储成本也越高。建议每个 Agent 只把自己真正要输出的结构化结果写入共享字段。7.2 工具设计可观测、可降级工具是 Agent 接触外部系统的入口必须做好日志和安全控制。所有工具调用记录日志包括入参、出参、耗时、调用方。外部接口调用设置超时避免 Agent 长时间等待。对写操作类工具默认不直接执行先返回“将要执行的 SQL”给人工确认。最小权限原则Agent 使用的数据库账号、云平台密钥权限只需要最低限度即可。7.3 提示词与上下文管理V1.x 中虽然消息列表是标准做法但多智能体场景下不建议把每个 Agent 的完整对话历史都传给其他 Agent。每个专业 Agent 应该只拿它需要的信息。上面的实战代码里reply 节点使用的是前序节点输出的分析文本而不是原始日志或完整 messages这种方式在真实环境中能明显降低 token 成本和无关信息干扰。7.4 安全边界企业级 Agent 系统最重要的合规点API Key 不允许出现在代码仓库中。Agent 不能拥有直接删除生产数据的工具。如果业务确实需要工具内部必须增加二次确认参数。审计日志完整记录 Agent 的每一次决策和工具调用。特别是多智能体系统中角色权限不同越权调用其他 Agent 的工具会带来隐患。对于涉及用户个人信息的数据Agent 节点要做脱敏处理后再传给大模型。7.5 性能与成本优化优先使用更小的模型做路由和分类大模型只做关键推理。对不常变化的知识片段使用 RAG 缓存。将 LangGraph 的长连接任务异步化避免阻塞 HTTP 请求线程。如果业务允许考虑对重复性工单结果做缓存大幅节省模型调用成本。8. 总结与学习路线这篇文章从 LangChain V1.x 核心组件讲到 LangGraph 状态编排再落到一个可运行的多智能体工单处理系统核心是想让大家建立一套新的开发心智不要再把 Agent 看成“一个会调用函数的模型”而要把它看成一张有状态、有分支、有断点、可恢复的图。学习顺序建议这样安排先熟练掌握langchain-core的模型、消息、工具、结构化输出。再学习 LangGraph 的状态、节点、边、条件边把所有概念在自己的小 Demo 中跑一遍。接着从单 Agent 升级到多 Agent先实现 Supervisor 模式再尝试 Handoff 和 Hierarchical 模式。加入持久化和断点机制模拟人工审核流程。最后做工程化改造日志、限流、密钥管理、权限控制、监控告警。如果你现在正处在从 LangChain 0.x 往 V1.x 迁移的阶段建议不要逐行改代码而是先按本文的架构重新梳理业务边界。绝大多数旧项目的核心难点不是 API 改名而是状态流和职责边界没理清。LangGraph 只是把这个问题从“隐藏的隐式逻辑”变成了“显式的图结构”这也正是它对企业级项目最有价值的地方。如果这篇文章对你理解 LangChain V1.x 和 LangGraph 有帮助可以收藏备用。后续我也会继续补充分布式多智能体部署、Agent 可观测性、RAG 与多智能体组合实战等话题。