ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

LangGraph核心概念与实战:从零搭建可控AI Agent

LangGraph核心概念与实战:从零搭建可控AI Agent LangGraph 是当前做 AI Agent 开发时绕不开的一个编排框架。很多人一开始以为它是 LangChain 的升级版或者只是 LangChain 里的某个模块实际接触之后才发现LangGraph 负责的是 Agent 的状态流转、循环控制、分支路由和任务编排而 LangChain 更偏模型调用、提示词组装和工具封装。如果你准备认真做 Agent 开发不只是调一个 prompt 接口LangGraph 这一套值得花时间学。下面按我第一次从零跑通 LangGraph 的顺序来写尽量把环境准备、核心概念、条件路由、子图、持久化和调试方式都讲清楚。1. LangGraph、LangChain、Agent 三者到底怎么分工很多人一开始会在这三个词之间绕晕。先花两分钟把定位理清楚后面接触代码才不会把概念混在一起。1.1 从一次 Agent 执行看 LangChain 的位置一个典型的 Agent 任务大概是这样的用户问“上海今天需要带伞吗”大模型先判断自己不知道实时天气于是决定调用一个天气查询工具拿到工具返回的结果之后再整理成自然语言回答用户。这个过程中LangChain 主要承担模型接口封装、提示词模板、工具定义、文档加载、向量检索这些偏“零件”的事情。Agent 是一种架构思路不是某一个具体的类。它的核心是“让模型决定下一步做什么然后反复执行”。LangGraph 负责把“模型决策、工具调用、结果回填、再次决策”这一整个循环用图的方式编排出来。如果你只用 LangChain 自带的高层 Agent 封装也能跑通上面的流程。但问题是框架内部把控制流写死了。你想加一个“如果工具返回异常就换一个工具重试”、想加“超过两轮工具调用就强制结束”、想加“用户中断之后恢复会话继续执行”这些需求在高层封装里很难改。1.2 LangGraph 补上的是流程控制层LangGraph 的核心思想很简单把任务执行过程画成一张图。图里有节点节点之间用边连接边上可以加条件整个任务会维护一份共享状态。这个设计带来的好处是控制流完全可见、可改。比如“调用模型”是一个节点“调用工具”是另一个节点。模型节点执行完之后不是直接进入工具节点而是先走一个路由函数由当前状态决定下一步去哪个节点。对比一下 LangChain 和 LangGraph 的侧重点维度LangChainLangGraph定位模型交互与组件封装Agent 执行流程编排核心单元ChatModel、Tool、RetrieverState、Node、Edge状态管理链式传递弱显式状态支持持久化循环和分支高层封装不易改图结构完全可控适合场景RAG、Prompt 组装、简单调用Agent、多步任务、复杂生产流程这里要补一句LangGraph 不依赖 LangChain 才能跑。你可以只给 LangGraph 写普通函数当节点不用任何 LangChain 组件。实际开发中大多数人会选择 LangGraph LangChain 的模型封装和工具库一起用因为组件能省很多重复工作。1.3 选型建议什么时候用 LangGraph给一个相对保守的判断标准只是做 RAG、知识库问答、单轮工具调用LangChain 的高层接口就够了。需求里有明确的“多轮循环、条件判断、失败重试、会话恢复、人机协作”优先考虑 LangGraph。团队需要把 Agent 流程可视化、加日志埋点、做单元测试图结构会比封装好的 Agent 类更容易维护。很多人是被“Agent 开发必须用某框架”这种说法带着走其实先理解自己的业务复杂度更重要。简单场景用简单方案复杂场景再上编排框架这样项目返工最少。2. 环境准备先跑通一个最小 LangGraph 流程LangGraph 的安装没有太多坑但环境问题会影响后续调试。先按最小步骤把图跑起来再去加模型和工具。2.1 依赖准备与项目结构我的建议是 Python 3.10 或 3.11。3.9 也能用但一些新版依赖开始要求更高版本。先创建一个虚拟环境再装核心包python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install langgraph如果你准备接大模型通常会装pip install langchain-core langchain-openai如果要用本地部署的大模型比如 Ollama、vLLM 这类服务它们大多数会暴露一个 OpenAI 兼容接口。你只需要把base_url指向本地服务地址把api_key填成任意占位字符串然后在ChatOpenAI里指定模型名即可。这个方式在开发环境里很实用。2.2 一个最小可运行的图先不接模型让两个普通节点跑一趟。这样可以先确认 LangGraph 基本流程没问题再往上加复杂度。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): query: str answer: str def process(state: State): return {answer: 你问的是 state[query]} builder StateGraph(State) builder.add_node(process, process) builder.add_edge(START, process) builder.add_edge(process, END) app builder.compile() result app.invoke({query: LangGraph 怎么入门}) print(result)这段代码做了几件事State是一个 TypedDict代表整张图的共享状态。process是普通函数输入当前状态返回一个字典。返回的 key 会更新到状态里。builder.add_node把函数注册成图里的节点。builder.add_edge(START, process)从入口节点连到process。builder.add_edge(process, END)表示这个节点执行完任务结束。compile()编译得到可执行对象之后用invoke传入初始状态。2.3 确认运行结果的判断标准跑通之后不要只看到“没有报错”就结束。建议检查三点返回值是一个字典而且包含answer字段。如果对同一个State里的其他 key 不做修改它们应该保留原值。节点如果返回了State中不存在的 keyLangGraph 并不一定立刻报错但后续代码可能取不到这个字段排查起来会绕。注意节点的返回值才是更新状态的唯一途径。你在节点内部改传入的 state 字典虽然可能生效但不符合图的数据流设计容易在并行分支或持久化时出现奇怪问题。跑通这个最小示例之后再往下学核心概念会轻松很多。3. 核心概念State、Node、Edge 是 LangGraph 的三块积木LangGraph 的所有复杂能力最终都由这三个概念组合而来。理解它们的边界比背 API 重要。3.1 State 保持上下文State 是整张图共享的数据容器通常用 TypedDict 定义。每执行到一个节点节点可以读取 State然后返回一个字典来“部分更新”它。举个例子如果 State 里有messages和query两个字段某个节点只返回{query: 新的问题}那messages不会丢。这种设计比每次全量覆盖 State 适合在复杂流程里保留上下文。设计 State 时要注意只放需要跨节点共享的数据。临时变量、大段中间结果、调试信息不要一股脑塞进去否则图跑起来之后很难查状态和定位问题。3.2 Node 负责执行Node 不需要继承任何基类就是一个普通函数。它接收当前状态返回更新结果。这个设计让节点非常好测试你可以直接手动构造一个状态字典调函数看返回值。我一般会建议节点函数保持单一职责。一个节点只做一件事调模型、调工具、做字段拼接、判断分支。如果节点内部又调模型又写文件又做日志排查问题时就要翻很多代码。3.3 Edge 决定下一步走向Edge 有三种常见形式普通边从节点 A 直接到节点 B。条件边节点 A 执行完之后进入一个路由函数根据状态决定去哪个节点。入口和出口START是图开始节点END是图结束节点。普通边适合固定流程条件边适合需要模型或者业务逻辑做判断的场景。一个最基本的带条件路由示例可以这样写def router(state: State): if len(state[query]) 5: return short return long builder.add_conditional_edges( process, router, { short: short_handler, long: long_handler, }, )路由函数返回的值必须对应路径映射表里的 key。如果返回了一个不在映射表里的字符串运行时就会报错。这个报错很常见排查时先看路由函数返回值。3.4 为什么状态设计决定 Agent 质量刚开始写 Agent 时最常见的错误是把所有东西都塞进 State。比如把模型原始返回、工具执行日志、用户输入的原始格式、中间计算的临时值全部放进 State表面上方便实际上会让状态变得非常难追踪。更合理的做法是把 State 当作“任务的当前事实”。比如messages对话历史。tool_results工具调用结果。plan当前整体计划。retry_count重试次数。status当前任务状态。这些字段是真正需要跨节点共享的。临时变量放在节点内等节点返回时再决定是否对外暴露。4. 从单流程升级成带工具调用的 Agent最小流程跑通之后下一步就是做一个真正会调用工具的 Agent。这里最核心的不是“怎么调工具”而是“怎么让模型决定要不要调工具、调完工具之后怎么回到模型继续生成”。4.1 先定义一个工具用 LangChain 的工具装饰器可以快速定义一个带类型和说明的工具from langchain_core.tools import tool tool def get_weather(city: str): 查询指定城市的当前天气。 Args: city: 城市名称例如 上海 return f{city}: 晴25 度工具函数本身的注释非常重要。大模型会根据函数名、参数描述和 docstring 来判断什么时候调用这个工具。如果说明写得不清楚模型就可能在该调用的时候不调用或者把参数传错。4.2 让 Agent 决定是否调用工具模型是否调用工具取决于它是否被绑定到这个工具上。常见写法from langchain_openai import ChatOpenAI llm ChatOpenAI( base_urlhttp://localhost:8000/v1, api_keylocal, modelyour-model-name, ) llm_with_tools llm.bind_tools([get_weather])bind_tools之后模型在需要时会返回工具调用请求但这个请求不会自动执行需要你自己写节点处理。这也是 LangGraph 和 LangChain 高层 Agent 的一个区别LangGraph 把执行权交给你。4.3 用条件路由控制“思考-调用-继续”循环手写一个最简单的 ReAct 循环至少需要两个节点call_model把当前消息列表交给模型模型可能返回普通回答也可能返回工具调用。call_tool把工具调用请求交给对应的工具执行把结果追加到消息列表。然后需要一个路由函数判断如果模型返回了工具调用就进入call_tool然后回到call_model如果没有工具调用就结束。def call_model(state): messages state[messages] response llm_with_tools.invoke(messages) return {messages: [response]} def call_tool(state): messages state[messages] last messages[-1] tool_results [] for tool_call in last.tool_calls: tool_result get_weather.invoke(tool_call[args]) tool_results.append( { role: tool, tool_call_id: tool_call[id], content: tool_result, } ) return {messages: tool_results} def should_continue(state): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END这段代码有几个关键点每次模型调用完messages都会多一条消息所以状态必须设计成能追加不能覆盖。should_continue检查最后一条消息里有没有tool_calls。有就继续去执行工具没有就结束。工具执行结果要带tool_call_id并且以tool角色追加回去模型才能知道这次工具调用的结果。4.4 工具调用循环里的常见问题实际跑这个循环时最容易遇到的是死循环。原因是模型每次都可能返回新的tool_calls只要它一直想调用工具循环就不会结束。常见处理方式在 State 里加一个step_count或max_iterations超过阈值后强制走结束分支。在call_tool里对同一个tool_call_id做去重避免重复执行。给工具调用总体加超时防止某个工具卡住整个图。注意不要一上来就开最大并发先让单个对话完整跑通确认工具结果能正确回传模型再考虑多轮并发和批量任务。5. 条件路由、分支控制、子图和并行LangGraph 的进阶价值不在单个循环而在分支、并行和复用。这一节是很多 LangGraph 实战教程里最常被忽略的部分但对于生产级 Agent 非常重要。5.1 条件路由conditional_edges 的细节条件路由的完整结构是一个节点执行完后由一个路由函数返回分支名然后根据映射表跳到对应节点。builder.add_conditional_edges( analyze_query, route_by_intent, { weather: weather_agent, news: news_agent, unknown: fallback_agent, }, )这里有一个容易踩的坑路由函数返回的是字符串必须能在路径映射表里找到。如果你在路由函数里拼错了字符串比如多了一个空格LangGraph 会直接报错而且报错信息不一定直接指出这个原因很多时候要自己打印路由返回值才能确认。所以我在写路由函数时会顺手加一行日志把返回值打出来def route_by_intent(state): intent state.get(intent, unknown) print(当前路由目标:, intent) return intent等流程稳定之后再删掉或者改成logger.debug。5.2 分支节点如何写分支节点本身也是一个普通节点。它不负责最终执行只负责“判断”。判断依据可以是用户输入的意图。模型返回的分类结果。某个工具返回的错误码。重试次数是否达到上限。把判断逻辑单独抽成一个节点比直接在边里写 lambda 更清晰因为 lambda 里写复杂判断很难测试。5.3 子图把复杂任务拆成可单独调试的模块子图就是把一个已经编译好的图当作另一个图的节点来使用。sub_weather weather_builder.compile() builder.add_node(weather_agent, sub_weather) builder.add_edge(weather_agent, END)子图的优势有三个独立复用。一个天气查询子图可以在多个主图里使用。独立测试。子图可以单独传自己的输入跑通后再挂到主图排查范围大幅缩小。隔离状态。子图内部的状态和主图可以不同你不需要把子图内部所有中间字段都放到主图 State 里。我一般习惯先顺手把子图单独写成模块。如果后面需要调整天气逻辑完全不用动主图。5.4 并行分支与状态合并并行分支的写法比较直观从一个节点分出多条边最后汇聚到一个合并节点。builder.add_edge(START, fetch_news) builder.add_edge(START, fetch_weather) builder.add_edge(fetch_news, merge_result) builder.add_edge(fetch_weather, merge_result)但这里有一个非常隐蔽的问题多个并行节点要同时更新同一个 State key最后结果可能互相覆盖。比如fetch_news和fetch_weather都想写result后结束的节点会覆盖先结束的节点。解决方式是在 State 里使用带合并策略的字段from typing import Annotated, TypedDict import operator class State(TypedDict): results: Annotated[list, operator.add]这种写法表示results字段用operator.add来合并。并行节点返回的列表会拼在一起而不是互相覆盖。如果你需要更复杂的合并逻辑可以传自定义函数。并行分支虽然能提升速度但也会让状态变化更难追踪。在新手阶段我建议先把多分支写成串行等每一步输出都稳定了再改并行最后用日志确认合并结果。6. 持久化与会话记忆让 Agent 记住上下文LangGraph 和普通链式调用最大的区别之一是支持持久化和断点恢复。这个能力不仅影响记忆还影响生产环境里的任务恢复。6.1 为什么要 checkpoint简单场景里每次invoke都从空状态开始对话历史全靠手动拼消息。但是生产场景下你往往需要用户关掉页面后回来还能继续之前的任务。Agent 执行到最后一步崩溃能够从检查点恢复而不是从头重跑。两个用户之间的状态不能串线。LangGraph 用 checkpointer 保存每次执行的状态快照。只要在编译图时传入 checkpointer并在每次调用时指定一个thread_id框架就会把状态隔离开来。6.2 在 LangGraph 中启用简单记忆开发环境里用内存型 checkpointer 最方便from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-001}} result app.invoke({messages: []}, configconfig)同一个thread_id的多次调用状态会持续累积。不同thread_id之间互相隔离不会串数据。生产环境通常不能只依赖内存因为进程重启就丢了。可以考虑接 SQLite、Postgres 这类外部存储。具体配置方式会根据你的存储服务不同而变化这里给的是思路落地时按官方文档选择对应的 checkpoint 类即可。6.3 记忆不是越多越好很多人一听到“能记住上下文”就把整段历史全部塞给模型。实际上消息列表越长模型推理耗时越长费用也越高。一些模型的重复内容会让 Agent 决策变差。用户明确要求清空历史时直接在状态里重置消息即可。我比较推荐的做法是在状态里同时维护一个“完整消息历史”和一个“最近消息窗口”。需要做摘要时用摘要节点把旧消息压缩成一小段再放回状态。这个思路同样可以用 LangGraph 的一个子图来实现。注意如果你发现同一个用户在不同会话里能看到彼此的历史先检查thread_id是不是没有正确隔离。这通常是配置问题不是模型问题。7. 开发调试与运行方式langgraph dev 和直接启动服务写 LangGraph 图的过程中调试方式决定效率。很多人习惯所有代码写完之后再invoke一报错就开始猜其实 LangGraph 提供了更好的调试路径。7.1 langgraph dev 解决了什么问题langgraph dev是 LangGraph 官方提供的一个开发服务命令。它和你自己写 Python 脚本直接运行图最大的区别是提供图形化调试界面能查看每次节点执行前后的状态。支持设置断点在某个节点执行前暂停。能直接查看输入输出、路由走向和状态变化。适合在开发阶段完整观察 Agent 的每一次思考、工具调用和状态跳转。使用langgraph dev之前项目里通常要定义一个langgraph.json配置文件声明图在哪里。下面是一个示意{ dependencies: [.], graphs: { agent: ./agent.py:graph } }不同版本的字段可能略有调整建议以官方文档为准。这个文件的作用是告诉开发服务器“入口图对象在哪里”这样它才能把节点、边和状态加载到调试界面里。7.2 和 uv run uvicorn app 有什么区别你可能见过uv run uvicorn app这种启动方式。它能直接启动一个 FastAPI 或者类似服务把 Agent 包装成 HTTP 接口。两者定位完全不同启动方式主要用途特点uv run langgraph dev开发调试有图形化界面支持断点和状态查看uv run uvicorn app提供服务适合把 Agent 发布成 API供其他系统调用脚本直接compile()后invoke快速验证最轻量适合本地测试核心逻辑实际项目里我建议的开发节奏是用脚本直接调用快速验证单步逻辑。用langgraph dev观察完整流程定位状态或路由问题。确认稳定后用 uvicorn 启动接口服务或者直接用 LangGraph 的 Cloud/Server 方案做部署。7.3 我的调试顺序我自己排查 LangGraph 问题时一般按这个顺序先看是启动报错还是运行报错。启动报错先看langgraph.json、依赖环境和入口对象命名。运行报错先打印每个节点返回的 state 变化这个节点返回了什么下一步路由到了哪里再用图形化界面看整张图的执行轨迹。如果某个节点没有按预期执行优先确认路由函数返回值。如果状态里的字段太多最好在调试时专门打印关键字段而不是打印整个状态否则日志会非常长反而不容易定位问题。8. 常见报错排查链路LangGraph 的报错分两类。一类是使用方式错误一类是业务逻辑问题。不管是哪类不要急着改参数先按固定顺序排查。8.1 不管什么报错先按这个顺序查第一层看现象。是报错中断、任务卡住还是输出结果不对三种现象的处理路径完全不同。第二层看输入。状态里传的字段类型对不对消息列表里的每条消息是不是都有正确的 role工具参数是否符合函数签名第三层看实现。节点是否返回了 State 中不存在的 key路由函数返回值是否在映射表里工具是否抛了异常第四层看环境。依赖版本是否兼容Python 版本是否过老端口是否被占用模型服务地址是否能连通第五层看设计。是不是该用子图的地方没有拆是不是并行节点写同一个 key 却没有合并策略循环退出条件是否真的能在有限步内满足很多时候看起来是 LangGraph 报错实际是工具函数写错了、模型 base_url 不通、或者路由函数拼错了一个字符串。8.2 常见报错现象与处理现象优先排查提示找不到 graph 或入口节点langgraph.json路径错误入口对象名不一致invoke返回空或字段缺失节点没有返回该 key返回 key 没在 State 中定义工具调用一直不结束没有设置最大迭代次数模型消息里tool_calls没被正确处理并行任务结果互相覆盖并行节点写同一个 key没有用 reducer 合并会话记录串到别的用户thread_id设置错误或未隔离全局复用了同一个 checkpointer模型接口超时或无效请求base_url、api_key、模型名、网关地址有问题运行卡住但没报错节点内可能有死循环外部依赖如数据库、模型服务卡住8.3 几个实用的避坑建议第一先跑通最小图再往上加工具。不要第一次写图就堆十个节点。我见过很多项目一上来就搭一个几十个节点的图最后连路由在哪里走错都找不到。最小图能跑通之后每个新节点都单独验证一次排查成本会低很多。第二工具函数一定要写清楚 docstring。模型判断什么时候调用工具依据的主要是函数名、参数和描述。描述模糊的工具在真正运行时表现会非常不稳定。第三路由函数和节点函数都要留日志。这是排错最重要的线索。不用写太多每个节点返回时打一行关键数据就够了。第四谨慎使用全局可变变量。LangGraph 的 State 是数据流不是让你在节点里不断修改全局字典。如果遇到并发执行全局变量会带来很难复现的 bug。第五如果你在本地测试时发现模型响应很慢不要急着优化图结构先确认是不是本地模型服务本身吞吐有限。低配置机器跑小模型做实验是没问题的但批量任务和生产部署就要单独考虑资源、队列和超时。LangGraph 真正难的地方不是 API 记不记得住而是状态怎么设计、循环怎么退出、分支怎么收敛。很多问题看起来是模型不行实际上图结构没有给模型留出足够的判断空间。建议新手按照“最小图 - 单条件路由 - 工具循环 - 子图 - 持久化”这个顺序练习。先把状态设计和日志排查理顺再考虑并发、任务队列和正式部署比一开始就追求复杂架构要靠谱得多。
返回列表