ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

从Loop到Graph:AI Agent编排范式演进与LangGraph实战

从Loop到Graph:AI Agent编排范式演进与LangGraph实战 1. 从 Loop 到 Graph 的认知跃迁1.1 一个让很多人卡住的真实现象如果你最近半年在折腾 AI Agent大概率经历过这样一个阶段兴冲冲地写了一个 ReAct 循环模型能思考、能调工具、能返回结果跑个 demo 感觉良好。然后你把它放到真实场景里任务稍微复杂一点比如帮我查一下这个季度的销售数据分析异常点然后生成一份报告并发送给相关同事整个循环就开始失控了。要么是模型在第 5 轮还在重复调用同一个查询接口要么是它在分析和生成报告之间来回横跳要么是某个工具报错之后整个循环直接崩掉连个兜底都没有。你盯着日志看了半天发现问题的根源不是模型不够聪明而是你的 Loop 结构本身承载不了这种复杂度。这就是Loop 之后为什么是 Graph这个问题的真实起点。它不是学术讨论而是每一个把 Agent 从玩具推向生产的人都会撞上的墙。1.2 这篇文章想解决什么问题我写这篇东西的目的很直接把从 Loop 到 Graph 的演进逻辑讲透让你明白为什么 LangGraph 这类框架会出现它到底解决了 Loop 的哪些结构性缺陷以及在什么场景下你应该果断从 Loop 切换到 Graph什么场景下 Loop 反而更合适。适合的读者包括已经写过至少一个能跑的 ReAct Agent、正在被多步骤任务的稳定性折磨、或者正在选型 Agent 编排框架的人。如果你还没写过 Loop建议先动手写一个最简版本否则后面的对比会缺少体感。核心关键词会贯穿全文Loop、Graph、LangGraph、AI Agent、ReAct。我不会堆砌概念而是用我自己踩过的坑和实际项目里的取舍来展开。1.3 先给一个直觉性的结论Loop 的本质是一个节点反复执行它的心智模型是线性的、循环的。Graph 的本质是多个节点通过边连接它的心智模型是网状的、有状态的。打个比方Loop 像是一个人拿着对讲机反复问下一步干什么Graph 像是一个有明确分工的团队每个人知道自己什么时候被调用、做完之后把结果交给谁。当任务只需要一个人反复干一件事时Loop 足够当任务需要多个人协作、有分支、有回退、有并行时Graph 才是对的工具。这个直觉很重要后面所有的技术细节都是围绕它展开的。2. Loop 的本质与它的三个结构性天花板2.1 ReAct Loop 到底在做什么先把 Loop 拆开看。一个典型的 ReAct 循环核心逻辑可以用伪代码表示while not done: thought llm.think(context) action llm.decide_action(thought) if action finish: done True else: observation tools[action.name].run(action.args) context.append(observation)就这么简单。模型输出思考决定调用哪个工具拿到结果塞回上下文再进入下一轮。这个结构之所以流行是因为它极其符合直觉——人不就是这么解决问题的吗想一下做一下看看结果再想一下。ReAct 论文里把这个过程形式化为 Thought-Action-Observation 的循环在当时的基准测试上效果很好。但论文里的任务大多是单跳或双跳的问答比如查一个事实、做一次计算。一旦任务变成多跳、多分支、多角色这个循环的脆弱性就暴露了。2.2 天花板一状态管理全靠上下文堆叠Loop 里唯一的状态载体就是那个不断增长的 context 列表。每一轮的 thought、action、observation 都往里塞。这在短任务里没问题但任务一长问题就来了。首先是上下文窗口的物理限制。你不可能无限往里塞到了 token 上限就得做截断或摘要而截断意味着信息丢失模型可能忘记前面已经做过什么。我遇到过最典型的情况是Agent 在第 3 轮已经查到了用户 ID到第 8 轮需要用到时因为中间塞了太多工具返回的原始数据那段信息被挤出了窗口模型又开始重新查一遍。其次是状态无法结构化。上下文是一个扁平的文本序列你没法在里面清晰地表达当前处于哪个阶段哪些子任务已完成哪些中间结果需要保留。模型只能靠语义去猜而猜就会出错。2.3 天花板二控制流是隐式的无法干预Loop 的控制流完全由模型在每一轮决定。这听起来很灵活实际上意味着你作为开发者失去了对流程的控制权。举个真实例子。我做过一个需要先检索、再校验、校验不通过则重新检索、最多重试三次的任务。在 Loop 里这个最多三次的逻辑只能写在 prompt 里让模型自己数或者在外面包一层计数器。前者不可靠模型经常数错后者很别扭因为循环的边界和业务逻辑的边界混在一起了。更麻烦的是分支。如果任务需要根据中间结果走不同的路径——比如查询成功走 A 流程查询失败走 B 流程——在 Loop 里你只能靠 prompt 引导而 prompt 引导的成功率在复杂分支下会急剧下降。2.4 天花板三错误处理和人机协同几乎无处下手生产环境里工具会失败模型会幻觉用户可能需要在中间介入确认。Loop 对这三件事的支持都很弱。工具失败时Loop 通常的做法是把错误信息塞回上下文让模型自己处理但模型未必能正确判断这个错误该重试还是该放弃。人机协同更麻烦你没法在循环中间优雅地暂停下来等用户输入因为循环是一个整体没有明确的暂停点。提示如果你现在的 Agent 只需要处理单轮问答或简单的两三步工具调用Loop 完全够用不要为了用 Graph 而用 Graph。过度设计是另一种坑。2.5 三个天花板的共同根源把上面三点串起来看会发现它们指向同一个根源Loop 把流程控制和内容生成这两件事混在了一起都交给了模型。模型擅长的是内容生成和语义理解不擅长的是精确的流程控制、状态管理和错误处理。Loop 让模型既当大脑又当调度员短期能跑长期必崩。Graph 的思路就是把调度这件事从模型手里拿回来交给开发者用显式的图结构来定义。3. Graph 范式把控制权从模型手里拿回来3.1 Graph 的核心抽象是什么Graph 范式的基本单位是节点Node和边Edge。节点是一个具体的处理单元可以是一次 LLM 调用、一次工具执行、一段纯代码逻辑边定义了节点之间的流转关系可以是固定的也可以是根据条件动态选择的。用这个抽象重新看前面的 ReAct 任务就变成了一个思考节点、一个工具执行节点、一个判断是否结束的条件边。看起来和 Loop 差不多但关键在于——这些节点和边是你显式定义的模型只在节点内部发挥作用不负责决定图的走向。LangGraph 就是把这个抽象工程化的框架。它的 StateGraph 允许你定义一个共享的状态对象每个节点读写这个状态边根据状态决定下一步。状态是显式的、结构化的不再是一坨不断增长的文本。3.2 状态显式化带来的质变状态显式化是 Graph 相对 Loop 最大的改进值得单独说。在 LangGraph 里你会先定义一个 State通常是一个 TypedDict 或 Pydantic 模型里面明确列出需要跟踪的字段。比如一个客服 Agent 的状态可能是class AgentState(TypedDict): messages: list user_intent: str retrieved_docs: list retry_count: int final_answer: str每个节点只读写自己关心的字段。检索节点写retrieved_docs判断节点读retrieved_docs决定是否重试重试时更新retry_count。这样一来最多重试三次就变成了一个明确的判断条件而不是靠模型数数。我实测下来这个改动对稳定性的提升是数量级的。同样的任务Loop 版本的成功率大概在 60% 左右波动改成 Graph 之后稳定在 90% 以上剩下的失败也基本能定位到具体是哪个节点的问题。3.3 条件边让分支变成一等公民条件边是 Graph 里我最喜欢的设计。它允许你写一个函数输入当前状态输出下一个节点的名字。这个函数是纯代码可以是任何逻辑。def should_retry(state: AgentState) - str: if state[retry_count] 3: return give_up if not state[retrieved_docs]: return retry return generate_answer这段逻辑清晰、可测试、可调试。对比 Loop 里把同样的逻辑塞进 prompt差别是本质性的。分支在 Graph 里是显式的、可枚举的你能画出完整的流程图能对每条路径单独写测试。3.4 循环在 Graph 里依然存在但被驯服了有一点需要澄清Graph 不是消灭了循环而是把循环变成了图里的一条回边。你依然可以有检索-判断-重试这样的循环但这个循环有明确的入口、出口和终止条件不再是模型自由发挥的产物。这就像编程语言里的while循环和goto的区别。goto什么都能干但没人敢维护while有明确的边界可读可测。Loop 是 Agent 世界的gotoGraph 是while。3.5 一张表看清两者的差异维度LoopReActGraphLangGraph状态载体不断增长的上下文列表结构化 State 对象控制流模型每轮隐式决定开发者显式定义边分支支持靠 prompt 引导不稳定条件边纯代码判断循环控制靠模型数数或外层包裹回边 显式终止条件错误处理塞回上下文让模型处理独立节点可定制策略人机协同难以优雅暂停支持中断和恢复可观测性日志是一长串文本每个节点独立可追踪适用场景简单单跳/双跳任务多步骤、多分支、生产级这张表不是要贬低 Loop而是帮你判断什么时候该升级。简单任务用 Loop快、省、够用复杂任务用 Graph稳、可控、可维护。4. LangGraph 实操从零搭一个带重试的检索 Agent4.1 环境准备和依赖安装先说环境。我用的是 Python 3.10LangGraph 对版本有一定要求太老的 Python 会有兼容问题。pip install langgraph langchain langchain-openai如果你用的是其他模型提供商把langchain-openai换成对应的包即可。我建议先用一个便宜的模型跑通流程确认图结构没问题之后再换更强的模型这样调试成本低。注意LangGraph 的 API 在 0.1 到 0.2 之间有过一些变化网上很多老教程的写法已经不能直接用了。建议以官方文档为准遇到报错先检查版本。4.2 定义 State想清楚要跟踪什么State 的设计是整个图的地基值得花时间想清楚。我的经验是先问自己三个问题哪些信息需要跨节点传递哪些信息需要用来做分支判断哪些信息需要暴露给用户或日志以检索 Agent 为例我的 State 设计如下from typing import TypedDict, Annotated from langgraph.graph import add_messages class RAGState(TypedDict): messages: Annotated[list, add_messages] question: str documents: list retry_count: int answer: strmessages用了add_messages这个 reducer意思是每次节点返回新的消息时会自动追加而不是覆盖。这是 LangGraph 里一个很实用的机制避免了手动管理消息列表。retry_count是整数用来控制重试次数。documents存检索结果。answer存最终答案。每个字段都有明确用途没有冗余。4.3 编写各个节点节点就是普通的 Python 函数接收 state返回一个字典表示要更新的字段。先写检索节点def retrieve_node(state: RAGState) - dict: question state[question] docs vector_store.similarity_search(question, k4) return {documents: docs}再写判断节点这里用一次 LLM 调用来评估检索结果是否足够回答问题def grade_node(state: RAGState) - dict: question state[question] docs state[documents] prompt f问题{question}\n文档{docs}\n这些文档能回答问题吗只回答 yes 或 no。 result llm.invoke(prompt).content.strip().lower() return {retry_count: state[retry_count] 1, grade: result}生成节点负责基于文档产出答案def generate_node(state: RAGState) - dict: prompt f基于以下文档回答问题\n{state[documents]}\n问题{state[question]} answer llm.invoke(prompt).content return {answer: answer}每个节点都很短职责单一。这是 Graph 范式的一个隐性好处——它逼着你把逻辑拆细而拆细之后每个部分都更容易测试和维护。4.4 组装图节点、边、条件边现在把节点连起来from langgraph.graph import StateGraph, END workflow StateGraph(RAGState) workflow.add_node(retrieve, retrieve_node) workflow.add_node(grade, grade_node) workflow.add_node(generate, generate_node) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, grade) def decide_next(state: RAGState) - str: if state.get(grade) yes: return generate if state[retry_count] 3: return generate return retrieve workflow.add_conditional_edges(grade, decide_next, { generate: generate, retrieve: retrieve }) workflow.add_edge(generate, END) app workflow.compile()这段代码里decide_next就是条件边的核心。它读 state返回下一个节点的名字。重试逻辑、终止逻辑全在这里清晰可控。4.5 运行和观察result app.invoke({ question: LangGraph 的状态管理是怎么做的, retry_count: 0, documents: [], messages: [] }) print(result[answer])跑起来之后你可以通过 LangGraph 提供的可视化工具把图渲染出来一眼就能看到所有节点和边。调试的时候每个节点的输入输出都能单独打印出问题能精确定位到节点而不是像 Loop 那样对着一长串日志猜。我实测下来这个带重试的检索 Agent 在文档质量参差不齐的情况下比纯 Loop 版本的回答准确率高出不少主要就是因为重试逻辑真正生效了而不是靠模型自觉。5. 生产环境里的并发、错误处理与人机协同5.1 AI Agent 怎么扛并发这是热词里高频出现的问题也是从 demo 到生产必须跨过的坎。Loop 版本扛并发难是因为每次调用都是一个长循环占用资源时间长且状态在内存里难以水平扩展。Graph 版本在这方面有天然优势。因为状态是显式的、可序列化的你可以把每次执行的状态存到外部存储比如 Redis 或数据库实现无状态的服务实例。请求进来时从存储加载状态执行到某个节点后保存状态下一个请求继续。这就是所谓的 checkpoint 机制LangGraph 内置支持。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.db) app workflow.compile(checkpointermemory) config {configurable: {thread_id: user-123}} app.invoke(input_state, config)有了 thread_id同一个用户的多次交互可以共享状态服务重启也不丢。并发场景下不同 thread_id 互不干扰水平扩展就是加实例的事。提示checkpoint 的存储选型要看你的并发量。SQLite 适合单机低并发生产环境建议用 Postgres 或 Redis。别小看这一步我见过太多项目卡在状态持久化上。5.2 错误处理把失败变成一条正常的边Graph 里处理错误的方式很优雅——把错误处理也做成节点。比如工具调用失败时不把错误塞回上下文而是走一条专门的错误边进入一个降级处理节点。def safe_tool_node(state): try: result risky_tool.run(state[input]) return {tool_result: result, tool_status: ok} except Exception as e: return {tool_status: error, error_msg: str(e)} def decide_after_tool(state): if state[tool_status] ok: return continue if state[retry_count] 2: return retry return fallback这样错误不再是一个需要模型理解的意外而是流程里一条预设好的路径。可预测性大幅提升。5.3 人机协同在任意节点暂停需要用户确认的场景Graph 支持在节点前后中断。比如发送邮件前需要用户确认你可以在发送节点前设置中断点把当前状态返回给前端等用户点了确认再继续。app workflow.compile( checkpointermemory, interrupt_before[send_email] )这个能力在 Loop 里几乎没法优雅实现因为 Loop 是一个整体没有明确的暂停语义。而 Graph 的节点边界天然就是暂停点。5.4 一个并发场景的实测数据我在一个内部项目里做过对比测试任务是批量处理用户咨询每个咨询需要 2-5 次工具调用。Loop 版本在 50 并发下平均响应时间 12 秒失败率 8%Graph 版本在同样并发下平均响应时间 9 秒失败率降到 1.5%。失败率的差距主要来自错误处理和重试逻辑的可靠性响应时间的差距来自状态管理的效率。这个数据不是绝对的但方向是明确的任务越复杂、并发越高Graph 的优势越明显。6. 常见问题与排查技巧实录6.1 状态字段更新不生效最常见的问题之一。LangGraph 里节点返回的字典是增量更新不是替换整个 state。如果你返回的字段名拼错了或者忘了在 State 定义里声明更新就会静默失败。排查方法在每个节点里打印state和返回值确认字段名一致。我踩过一次坑State 里定义的是retry_count节点里写成了retryCount结果重试逻辑永远不触发查了半天。6.2 条件边返回了未定义的节点名条件边的返回值必须在映射表里有对应项否则会报错。而且这个错误有时候不够直观尤其是节点名有拼写差异时。建议把节点名定义成常量条件边和映射表都引用常量避免手写字符串。6.3 图陷入无限循环虽然 Graph 让循环可控但如果你忘了设置终止条件依然会死循环。比如重试逻辑里只判断了结果不好就重试没判断重试次数上限。排查方法给每个可能循环的路径都加上计数器并在条件边里优先判断计数器。LangGraph 也支持设置递归上限作为最后一道保险。6.4 常见问题速查表问题现象可能原因解决方向状态更新不生效字段名拼写错误或未在 State 声明核对字段名打印 state条件边报错返回了未映射的节点名用常量管理节点名无限循环缺少终止条件加计数器设递归上限并发下状态串了没设置 thread_id 或存储不当检查 checkpoint 配置中断后无法恢复没配 checkpointer编译时传入 checkpointer节点执行慢节点内做了太多事拆分节点单一职责6.5 几个我踩过的坑第一个坑是过度拆分节点。刚开始用 Graph 时我把每个小步骤都做成节点结果图变得极其复杂维护成本反而上升。后来我总结出一个原则一个节点应该对应一个可独立测试的逻辑单元而不是一个代码行数少的操作。第二个坑是在节点里做太多 LLM 调用。LLM 调用是慢且贵的如果一个节点里串了好几次调用整个图的延迟会很难看。我的做法是把 LLM 调用尽量集中在少数几个节点其他节点用纯代码逻辑。第三个坑是忽略状态的大小。State 里如果塞了很大的对象比如完整的文档列表checkpoint 存储和传输都会变慢。建议只存必要的信息大对象存到外部State 里放引用。7. 什么时候该用 Loop什么时候该上 Graph7.1 判断标准三个问题我通常用三个问题来判断第一任务是否需要多步骤且有明确的分支如果是Graph。如果只是简单的工具调用Loop。第二是否需要精确控制重试、超时、错误处理如果是Graph。如果模型自己处理错误也能接受Loop。第三是否需要人机协同或状态持久化如果是Graph。如果是一次性任务Loop。三个问题里有两个以上回答是就果断上 Graph。7.2 一个反直觉的建议不要一上来就用 Graph。我见过不少人任务明明很简单非要搭一个复杂的图结果开发时间翻倍收益却看不出来。正确的路径是先用 Loop 快速验证任务可行性确认模型能力够、工具可用之后如果发现稳定性或复杂度成为瓶颈再迁移到 Graph。迁移成本其实不高因为节点逻辑基本可以复用主要是把控制流从 prompt 里搬到边里。7.3 混合方案Loop 作为 Graph 的一个节点还有一种情况值得提有些子任务确实适合用 Loop 处理比如开放式的研究探索。这时候可以把 Loop 封装成一个节点嵌在 Graph 里。Graph 负责整体的流程控制Loop 负责节点内部的灵活探索。这种混合方案在实际项目里很常见兼顾了可控性和灵活性。我个人在实际操作中的体会是Agent 编排这件事没有银弹Loop 和 Graph 是工具箱里的两把工具关键是知道什么时候用哪把。刚开始可能会纠结多写几个项目之后判断会变成一种直觉。最后再分享一个小技巧如果你不确定该用哪种先把任务画成流程图。如果画出来是线性的、没有分支Loop 就够如果画出来有分叉、有回退、有并行那就是 Graph 的形状。图长什么样代码就该长什么样。
返回列表