公司动态

LangGraph实战:用图式状态机编排复杂AI Agent流程

📅 2026/8/31 11:01:07
LangGraph实战:用图式状态机编排复杂AI Agent流程
这次我们直接看 LangGraph。它不解决 Prompt 怎么写、模型怎么选而是解决一个更麻烦的问题当一个 Agent 需要多个分支、循环、多轮记忆、并行调用工具时代码怎么组织才不乱。LangChain 团队在 LangChain 之后推出了这个图式编排框架核心思路是把 Agent 流程画成一张有向图节点执行实际逻辑边决定执行顺序Conditional Edge 决定走哪条路。你可以暂时把它理解成一个“带状态机能力的 Agent 流程引擎”。我建议所有正在接触智能体开发、想从固定 Chain 转向复杂 Agent 的开发者都先花时间把 LangGraph 的核心概念过一遍。原因有三个第一LangGraph 本身是一个库不是 IDE 也不是平台学习成本低安装了就能跑第二它对硬件没有强制要求纯 Python 环境就能运行只有在接入实际 LLM 时才需要 API Key 或本地模型第三它支持状态持久化、条件路由、并行节点和子图这些正是复杂 Agent 最常遇到的问题。本文会把环境准备、第一个图、State 修改、条件路由、并行分支、子图、多轮记忆、HTTP 接口封装这些环节全部带一遍代码都放在正文里可以直接复制运行。1. 核心能力速览能力项说明项目类型LLM 应用编排框架 / Agent 框架开源团队LangChain 团队编程语言Python主要功能StateGraph、Conditional Edge、Checkpointer、并行分支、子图、循环控制、状态持久化硬件要求CPU 即可运行框架本身接入 LLM 时需要 API 或本地推理环境显存要求无固定要求取决于实际使用的模型框架自身不加载大模型支持平台Windows / macOS / Linux启动方式pip 安装后通过 Python 脚本运行是否支持 API可自行封装 FastAPI 服务LangGraph 官方也有云端 Server 方案是否支持批量任务支持并行节点和循环逻辑也可接入外部任务队列配合使用适合场景复杂 Agent、多工具调用、多轮对话、条件流程、前后端集成的智能体服务从这张表可以看出来LangGraph 并不关心你用的是哪个大模型。你可以接 OpenAI 兼容接口也可以接本地模型甚至可以先不接模型用普通函数冒充节点把流程图跑通。这一点对学习特别友好排除了模型调用的不确定性让人只关注 Graph 本身的行为。2. LangGraph 是什么与 LangChain 的区别和适用场景在实战之前先把概念对齐。很多人问 LangChain 和 LangGraph 的区别其实一句话能说清LangChain 强调的是 Chain是一条顺序执行的链路LangGraph 强调的是 Graph是带状态、可以分支、可以循环的图。固定流程用 Chain 没问题但如果业务里出现“根据用户意图走不同分支”“工具调用失败要重试”“多轮对话需要跨轮记忆”这些场景纯 Chain 会变得很别扭LangGraph 就更有优势。这里整理一个对比表格维度LangChainLangGraph流程组织链式顺序调用图式状态机分支能力有限需要手动控制Conditional Edge 原生支持循环能力弱难以表达复杂循环支持节点到节点的循环边状态管理每次调用相对独立State 贯穿整个图执行过程持久化需要外部方案Checkpointer 原生支持适用场景简单管线、固定指令复杂 Agent、多轮对话、动态工具编排LangGraph 不是要替代 LangChain 的模型抽象和工具体系而是在它们之上增加了一层执行控制。实际项目里可以同时使用 LangChain 的模型封装、Tool 封装和 LangGraph 的编排能力。LangChain 官方现在对复杂 Agent 的建议也倾向于使用 LangGraph 这类图式框架。使用边界也值得说清楚。LangGraph 不负责解决模型幻觉不负责提供模型算力也不负责合规审批。它只负责让流程可控、可复现、可继续。图片生成、声音克隆、视频生成这类能力如果被集成到 Agent 里LLM 生成的内容是否合法、使用的素材是否获得授权始终是开发者自己的责任。所以在动手搭建之前先明确你的 Agent 用到什么数据、调用什么工具、输出什么内容这些边界比代码本身更重要。3. 环境准备与前置条件本地运行 LangGraph 的依赖很简单不需要 GPU不需要 Docker普通 Python 环境即可。建议使用 Python 3.9 以上版本3.10 或 3.11 更稳妥。演示项目推荐创建独立虚拟环境避免污染全局 Python。创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langgraph langchain-core如果后面要接 OpenAI 兼容接口再安装 LangChain 的 OpenAI 封装pip install langchain-openai安装完成后先验证导入是否正常from langgraph.graph import StateGraph, START, END print(LangGraph 导入成功)这里有一个常见问题LangGraph 不同版本之间的 API 命名存在差异最典型的是旧版用add_conditional_edges新版用add_conditional_edge。本文代码统一使用较新的add_conditional_edge写法如果你使用的版本报错提示找不到这个方法可以在代码里换回旧写法或者升级到最新版本。因为 LangGraph 迭代快建议以官方文档的版本说明为准。4. 从零跑通第一个 StateGraph第一个例子不接任何模型只用普通 Python 函数模拟两个节点。目标是把图的创建、编译、调用流程整个跑通。from typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): message: str def node_a(state: AgentState) - dict: return {message: state[message] - A} def node_b(state: AgentState) - dict: return {message: state[message] - B} graph StateGraph(AgentState) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({message: start}) print(result)执行后输出{message: start - A - B}这段代码说明了 LangGraph 的几个核心机制State 是一个 TypedDict描述整个图执行过程中携带的数据结构。每个 Node 接收当前 State返回一个 dict 作为更新内容。add_edge(START, a)表示从开始节点进入a。add_edge(a, b)表示a执行完进入b。add_edge(b, END)表示执行结束。compile()将图编译为可调用对象。invoke()以初始 State 启动整个流程。判断这个例子是否成功的标准只有一个输出里的 message 顺序必须是start - A - B说明边和节点的执行顺序符合预期。如果出现报错优先检查 LangGraph 版本和导入路径。从这一步开始LangGraph 的核心心智模型就建立起来了你画的每一条边都是节点之间的执行关系你定义的 State就是整个执行过程共享的“车厢”每个节点从里面拿数据再把新数据放回去。5. 在节点函数中修改 State 状态值这是 LangGraph 使用中最容易踩坑的地方。节点函数修改 State 并不是直接往 State 里赋值而是通过返回值来描述更新。返回值里的每个键会被合并到当前的 State 中。普通的字符串字段直接返回新的值即可def change_message(state: AgentState) - dict: return {message: new message}如果返回的键不存在LangGraph 会把它作为新字段加到 State 中。如果返回的键存在则会覆盖原值。这种机制默认是覆盖式更新。问题出在列表、字典这类可变类型上。假设 State 里有一个列表你希望节点往列表里追加一条数据from typing import TypedDict, Annotated import operator class ListState(TypedDict): messages: Annotated[list[str], operator.add] def add_one(state: ListState) - dict: return {messages: [new message]}这里关键点在Annotated[list[str], operator.add]。LangGraph 看到这个类型注解后不会把新值直接覆盖到旧值上而是调用operator.add把新旧两个列表拼接起来。也就是说State 中的字段如何更新由 reducer 决定。如果你的需求不是追加而是彻底覆盖整个列表就要自定义 reducerfrom typing import TypedDict, Annotated def overwrite_reducer(old: list[str], new: list[str]) - list[str]: return new class OverwriteState(TypedDict): messages: Annotated[list[str], overwrite_reducer]自定义 reducer 的本质很简单它是一个接收旧值和新值、返回最终值的普通函数。你可以在这里写任何合并逻辑比如去重、排序、限制长度。还有一点容易被忽略如果节点函数什么都不想改直接返回None即可LangGraph 会认为这个节点没有产生状态更新。如果不确定某个节点是否应修改状态尽量保持最小返回只返回需要变更的字段。这样排查问题时能快速定位是哪一步改变了 State。理解透这个机制LangGraph 的很多诡异行为都能解释通了。比如并行分支返回结果丢失、列表被整体替换而不是聚合十有八九都是 reducer 没有配置或者配置错了。6. 条件路由 Conditional Edge 实战条件路由是 LangGraph 比普通 Chain 明显更强的地方。它的核心是一个节点执行完之后根据当前 State 的内容动态决定下一个进入哪个节点。先看一个简单的路由场景输入文本里包含“代码”就走代码节点包含“写作”就走写作节点否则走兜底节点。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): input_text: str route: str def parse_intent(state: RouteState) - dict: text state[input_text] if 代码 in text: return {route: coder} if 写作 in text: return {route: writer} return {route: fallback} def coder(state: RouteState) - dict: return {input_text: state[input_text] [CODE]} def writer(state: RouteState) - dict: return {input_text: state[input_text] [WRITE]} def fallback(state: RouteState) - dict: return {input_text: state[input_text] [FALLBACK]} def decide_route(state: RouteState) - Literal[coder, writer, fallback]: return state[route] graph StateGraph(RouteState) graph.add_node(parse_intent, parse_intent) graph.add_node(coder, coder) graph.add_node(writer, writer) graph.add_node(fallback, fallback) graph.add_edge(START, parse_intent) graph.add_conditional_edge( parse_intent, decide_route, { coder: coder, writer: writer, fallback: fallback, }, ) graph.add_edge(coder, END) graph.add_edge(writer, END) graph.add_edge(fallback, END) app graph.compile() print(app.invoke({input_text: 帮我写一段代码})) print(app.invoke({input_text: 帮我写一篇博客})) print(app.invoke({input_text: 今天天气怎么样}))add_conditional_edge的三个关键部分是源节点先执行这个节点然后进入路由判断。路由函数接收当前 State返回一个字符串键。映射表把路由函数返回的键映射到实际节点名。这个机制非常清晰但也很容易出错。最常见的问题是路由函数返回的键没有出现在映射表里此时会抛异常。建议在任何条件路由里都留一个 fallback 键作为接受一切输入的兜底分支。条件路由的另一个常见应用是循环检测。很多 Agent 场景下模型需要反复调用工具直到拿到足够信息才结束。LangGraph 表达这个逻辑就是让条件路由的某一个分支指回当前节点或前序节点从而实现循环同时在 State 里维护一个计数器到达上限后退出。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): count: int message: str def step(state: LoopState) - dict: new_count state[count] 1 return {count: new_count, message: fstep {new_count}} def should_continue(state: LoopState) - Literal[next, end]: if state[count] 3: return end return next graph StateGraph(LoopState) graph.add_node(step, step) graph.add_edge(START, step) graph.add_conditional_edge( step, should_continue, {next: step, end: END}, ) app graph.compile() print(app.invoke({count: 0, message: }))输出结果{count: 3, message: step 3}这个示例对应了 LangGraph 实战里非常经典的主题条件路由与分支控制。循环不是把一张图无限跑下去而是通过状态里的计数器控制退出条件。实际项目里的退出条件可以换成“工具调用次数达到上限”“结果已经满足要求”“用户主动取消”等。7. 并行分支与子图实战复杂 Agent 中经常需要同时执行多个独立任务。LangGraph 天然支持从同一个节点分出多条边实现 fan-out 并行然后再汇合到一个聚合节点实现 fan-in。先看一个并行分支的例子from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): results: Annotated[list[str], operator.add] def task_a(state: ParallelState) - dict: return {results: [task_a done]} def task_b(state: ParallelState) - dict: return {results: [task_b done]} def join(state: ParallelState) - dict: merged , .join(state[results]) return {results: [fjoin: {merged}]} graph StateGraph(ParallelState) graph.add_node(task_a, task_a) graph.add_node(task_b, task_b) graph.add_node(join, join) graph.add_edge(START, task_a) graph.add_edge(START, task_b) graph.add_edge(task_a, join) graph.add_edge(task_b, join) graph.add_edge(join, END) app graph.compile() print(app.invoke({results: []}))输出结果{results: [join: task_a done, task_b done]}这里必须强调Annotated[list[str], operator.add]的作用。并行节点返回的是列表LangGraph 需要知道如何处理多个节点对同一个字段的更新。没有 reducer 时后执行完的分支会直接覆盖前面的结果导致丢失数据。用operator.add作为 reducer多个并行分支的返回值会按顺序合并成一个大列表。如果并行分支之间完全独立还可以对每条边做更细的自由度控制。你完全可以让task_a和task_b分别结束到不同节点或者在join之前再各自串一条后续处理节点。这说明 LangGraph 的表达能力不是“只能并行然后聚合”而是完全按照你画的边来决定执行流。子图是另一个常见需求。当某个流程已经编译成一个可用图后它可以作为另一个图的节点被直接调用。这种嵌套方式对模块化特别有用例如把“获取用户资料”的子图封装好主图的一个节点只需要调用它。from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): sub_input: str sub_output: str def sub_process(state: SubState) - dict: return {sub_output: state[sub_input] processed} sub_graph StateGraph(SubState) sub_graph.add_node(sub_process, sub_process) sub_graph.add_edge(START, sub_process) sub_graph.add_edge(sub_process, END) compiled_sub_graph sub_graph.compile() class MainState(TypedDict): main_input: str final_output: str def call_subgraph(state: MainState) - dict: sub_result compiled_sub_graph.invoke({sub_input: state[main_input]}) return {final_output: sub_result[sub_output]} main_graph StateGraph(MainState) main_graph.add_node(call_subgraph, call_subgraph) main_graph.add_edge(START, call_subgraph) main_graph.add_edge(call_subgraph, END) main_app main_graph.compile() print(main_app.invoke({main_input: hello}))子图的优势在于隔离和复用。子图内部的状态、边、条件路由都可以独立修改只要保持调用入口和返回值接口稳定主图不需要改动。这在团队协作时非常有用每个人负责一个子图最后在主图里组合。并行和子图组合起来可以应对大多数复杂流程。用并行提升吞吐用子图降低单个文件的复杂度再加上条件路由做动态控制LangGraph 能覆盖的场景已经相当完整。实际使用时不要一上来就把整张图画得太大先跑通最小闭环再逐步加并行和嵌套。8. 多轮对话与 Checkpointer 记忆LangGraph 的另一个关键能力是 Checkpointer它负责在不同轮次调用之间保存状态。没有 Checkpointer 时每次invoke()都是一个独立过程Agent 无法记住上一轮对话内容。有了 Checkpointer配合thread_id就能实现跨轮记忆。from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class ChatState(TypedDict): messages: list[str] def update_messages(state: ChatState) - dict: return {messages: state[messages]} graph StateGraph(ChatState) graph.add_node(update_messages, update_messages) graph.add_edge(START, update_messages) graph.add_edge(update_messages, END) checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: session-1}} app.invoke({messages: [第一轮你好]}, config) app.invoke({messages: [第二轮我还在吗]}, config) result app.invoke({messages: [第三轮请记住前面内容]}, config) print(result)同一个thread_id在多次invoke()之间共享状态。实际应用里每个用户可以固定一个thread_id这样 Agent 在处理新消息时能访问到同一线程下的历史状态。这个设计比自己在外面缓存对话记录更简洁因为状态由 LangGraph 统一管理。内存版MemorySaver适合开发和测试但进程重启后记忆就丢失了。生产环境建议使用外部持久化方案例如 LangGraph 官方支持的 Postgres Store、Redis Checkpointer或者自研存储。选择标准主要看你的部署环境单机小规模用内存没问题多实例部署必须用共享存储否则不同实例之间无法访问同一份会话状态。时间旅行是 Checkpointer 带来的另一个有趣能力。因为每个节点的状态都被保存下来理论上可以回退到任意一步重新执行。这个功能调试时很实用但需要配合正确配置才能使用。初学者先记住两点第一编译时传入checkpointer第二调用时传入thread_id。这两步做对多轮记忆就通了。9. 接入 LLM最小 ReAct Agent 示例前面所有例子都没有接真实模型。现在把它和 LLM 串起来做一个能调用工具的 Agent。这里使用 OpenAI 兼容接口你需要准备一个可用的 API Key推荐通过环境变量加载不要硬编码在代码里。import os from typing import TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, operator.add] tool_result: Annotated[list, operator.add] llm ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), temperature0, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} def call_tool(state: AgentState) - dict: return {tool_result: [工具执行结果]} graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.add_node(call_tool, call_tool) graph.add_edge(START, call_model) graph.add_edge(call_model, call_tool) graph.add_edge(call_tool, END) app graph.compile() result app.invoke({messages: [{role: user, content: 你好}]}) print(result[messages][-1].content)这个示例不是完整的工具调用循环但它演示了 LangGraph 节点里如何调用 LLM以及消息列表如何在节点之间流动。真正生产级的 ReAct Agent 需要在call_model和call_tool之间加入条件路由判断模型返回的是普通回答还是工具调用指令然后根据结果决定继续执行还是结束。这种循环逻辑完全可以结合第 6 节的add_conditional_edge来实现。接入模型之后要注意几个问题模型调用耗时通常远高于普通节点调试时要区分是图执行问题还是模型响应慢。消息列表需要使用Annotated[list, operator.add]之类的 reducer否则多轮叠加后会互相覆盖。工具调用的结果应该独立放在一个字段里方便后续节点读取也能避免污染对话历史。如果接入的是本地模型同样可以走 OpenAI 兼容协议只需替换base_url和模型名即可。10. 封装 HTTP API 与批量任务本地脚本能跑通之后下一步通常是把 Agent 暴露成 HTTP 服务。LangGraph 本身不强制你使用什么 Web 框架用 FastAPI 包一层即可。from fastapi import FastAPI from pydantic import BaseModel from langgraph.checkpoint.memory import MemorySaver app FastAPI() agent graph.compile(checkpointerMemorySaver()) class ChatRequest(BaseModel): thread_id: str message: str app.post(/chat) def chat(request: ChatRequest): config {configurable: {thread_id: request.thread_id}} result agent.invoke({messages: [{role: user, content: request.message}]}, config) return {reply: result[messages][-1].content}启动服务uvicorn main:app --host 0.0.0.0 --port 8000调用示例curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {thread_id: user-001, message: 你好}这里有几个工程化建议服务启动后需要通过thread_id区分不同会话。不要把不同用户的对话塞进同一个线程否则状态会串。如果使用内存版 Checkpointer服务重启后所有会话丢失。生产环境必须替换为共享存储。接口层要做输入校验限制消息长度防止超大输入打爆上下文。如果同一个 Agent 实例要被多个请求并发调用要注意 Checkpointer 的线程安全和 I/O 性能必要时引入请求队列。批量任务方面LangGraph 本身提供并行节点能力可以用来处理一个请求内部的并发工具调用。而大量独立会话的批处理更适合放在外部任务队列里例如 Celery、Redis Queue每个任务调用一次/chat接口并指定自己的thread_id。LangGraph 不负责队列调度但作为单次执行引擎它的接口足够干净和任务队列配合没有障碍。批量处理时还要注意重试策略。LLM 调用可能因为限流、超时等原因失败建议在任务队列层配置重试而不是在 LangGraph 图里无限循环。图内部的循环主要用于流程控制例如“工具结果不满意再调用一次模型”这种循环是功能和业务逻辑的一部分不是容错机制。11. 常见问题与排查方法问题现象可能原因排查方式解决方案导入 langgraph 失败虚拟环境中未安装包或 Python 版本过低执行pip show langgraph查看版本安装 langgraph 并使用 Python 3.9 以上版本invoke 后结果没有变化节点返回 None 或返回的键拼写错误打印节点返回值与 State 对比确认节点返回 dict 且键名与 State 字段一致列表字段被整体覆盖而不是追加字段类型没有配置 reducer检查 TypedDict 中是否使用 Annotated使用Annotated[list, operator.add]或自定义 reduceradd_conditional_edge 报错路由函数返回了映射表中不存在的键打印路由函数返回值添加兜底分支或修正映射表同一 thread_id 下历史状态丢失Checkpointer 未配置或外部存储失效确认 compile 时是否传入 checkpointer编译时传入 checkpointer生产环境用持久化存储条件路由死循环循环退出条件永远不满足在循环节点里打印计数或状态设置最大迭代次数并在条件函数中判断上限模型调用一直失败API Key、接口地址或网络问题单独用一条 llm.invoke 测试检查环境变量、base_url 和网络连通性并行分支返回结果缺失多个分支同时更新一个字段却没有 reducer检查并行更新字段的 Annotated 配置为并行更新的字段配置合并 reducer端口被占用本机已运行其他服务执行lsof -i:8000查看占用进程更换端口或释放原进程版本升级后代码报错LangGraph API 命名变化查看官方 Changelog 或文档按新版 API 调整代码这些是新手入门最常见的拦路虎。多数问题不是逻辑复杂造成的而是对 State 更新机制和版本差异不够熟悉。遇到报错时先看报错信息指向哪个 API再看它影响的是 State 定义还是 Graph 构建通常很快能定位。12. 最佳实践与使用建议从项目工程化的角度看有几点建议值得记下来。第一个建议是最小可运行闭环优先。不要一开始就设计大而全的 Agent 图先定义一个小 State、两个节点、一条边跑通后再增加条件路由、子图、并行。LangGraph 的错误很多时候在大型图里特别难排查保持小图能让问题快速浮出水面。第二个建议是 State 字段保持精简。State 是整个图的公共上下文字段越多节点之间的依赖越复杂reducer 冲突的概率也越高。尽量只放节点间需要共享的数据临时变量放在节点内部局部处理。第三个建议是条件路由必须有兜底分支。真实用户的输入不可控路由函数可能返回预期之外的内容映射表里没有对应键就会直接报错。兜底分支可以进入一个安全回复节点或者直接结束至少不会让整个流程崩溃。第四个建议是加入日志和追踪。在生产环境里Agent 执行过程涉及模型调用、工具调用、状态更新每一步都可能出错。建议在节点之间打印或记录 State 的关键变化或者集成可观测性工具把每一次图执行的节点顺序和耗时记录成结构化日志。没有日志的 Agent 系统几乎没法维护。第五个建议是合规先行。如果 Agent 涉及真实用户人脸、声音、隐私数据或版权素材必须确认已经获得相应授权并在正式使用前完成效果复核。不要因为“只是测试一下”就绕过授权和隐私保护流程。技术可以做得很灵活但使用边界要自己守好。13. 总结与下一步LangGraph 最值得花时间研究的地方就是状态管理和条件路由。这两个机制掌握了Agent 的复杂流程就不再是散乱的 if-else而是可以画成图、可以调试、可以持久化的结构化模型。本文从零开始带了一个最小的图又依次过了 State 更新、条件路由、循环、并行、子图、Checkpointer、模型接入和 HTTP 封装这套链路就是大多数 Agent 项目的通用骨架。建议你现在就动手做三件事第一把第 4 节的第一个例子复制到本地跑通第二给这个图加一个条件路由让不同输入走到不同节点第三给节点配置一个计数器实现三次循环后退出。这三个练习做完LangGraph 的核心用法基本就掌握了。最容易踩的坑就是 reducer 和 API 版本差异。看到异常先看自己的 State 字段有没有用Annotated配置 reducer再看代码里的 API 名称和新版文档是否一致。这两个点检查完大部分问题都能解决。后续扩展方向也有很多接入更多工具、接入外部向量数据库、部署 LangGraph Server、对接前端页面。等基础图跑通后这些都可以按需加进来。建议把本文代码保存成一份可运行的实验脚本以后开发新 Agent 时直接基于这份骨架扩展。如果你正在准备 Agent 相关项目LangGraph 这套图式编排值得深入。先把核心概念吃透再结合真实业务去扩展比直接堆功能要稳妥得多。