1. 为啥 Agent 编排非得走向 Graph 架构
做了两年多 AI Agent 的实际项目,从最早的单 Agent 聊天机器人,到后来多 Agent 协作系统,再到给某公司的跨平台系统搭了一套完整的 Agent 构建框架,我踩过的坑不说几百个也有几十个。最深的感受是:单 Agent 是"写"出来的,多 Agent 是"编排"出来的,而编排这件事,用 Graph 做底子是最接近直觉的方案。
先说背景。我在做一个叫 Harness 的 Agent 构建框架,核心思路就一句话:把大模型、工具调用、条件判断、状态流转全部抽象成"图上的节点和边",运行时按图执行。听起来不复杂,但真正落地的时候牵扯到大量细节,包括状态管理、环路控制、超时熔断、上下文窗口管理,哪一个没想清楚都会在线上出问题。
为什么选择 Graph,而不是最常见的 Chain(链式)或者 Pipeline(管道)?因为真实业务里的 Agent 流程根本不是线性的。你做一个客服助手,用户问"我的订单到哪里了",Agent 要先判断用户有没有登录,没登录要走登录引导,登录了要查订单系统,订单系统返回异常要走重试或转人工,这中间有分支、有条件、有回路,用 Chain 做只能祈祷所有路径都是直线。Graph 允许你在任意节点之间连边,天然支持分支和循环,逻辑上更接近人类的"流程图思维"。
另外一个关键点是可观测性。Graph 的每个节点和边都可以埋点,Agent 跑完整个链路之后,你能把"用户输入哪个节点"、"哪一步调了哪个工具"、"返回了什么东西"一五一十地还原出来。这在传统的函数调用堆栈里非常难做到,但在 Graph 架构下,几乎等于白送。后面我会详细讲我是怎么做的。
这个内容适合谁看?如果你正在做多 Agent 系统,或者打算给团队搭一套 Agent 开发框架,又或者只是对"Agent 到底怎么从单线程进化到复杂编排"感兴趣,这篇文章值得读完。我尽量把设计思路、代码结构、踩坑记录都写得直白,毕竟我也是从小白一点点摸过来的。
2. 三种"图"不是一种图——先搞清楚你要哪种
很多刚接触 Graph 架构的同学,一上来就陷入概念混战:知识图谱、任务流程图、函数调用图、依赖图,看起来都带 Graph 这个词,实际上完全不是一个层面的东西。我在设计 Harness 时把它们彻底分开了,这一点建议所有准备做类似框架的人都先想清楚。
2.1 知识图谱管的是"记忆",不是"执行"
先说知识图谱(Knowledge Graph)。它解决的是"Agent 的知识从哪来、怎么组织"的问题。比如一个法律咨询 Agent,想让它在回答前先回忆法条、案例、司法解释,这些实体之间天然有语义关系,放到图里很自然。你问"合同违约赔偿",图里能查出与合同、违约金、实际损失相关的所有节点,作为上下文注入提示词。
我最早误以为 Harness 也要内置知识图谱,后来发现这是错误的耦合。Agent 的执行流程和它读取的外部知识,是两个独立的问题域。执行流程关注的是"下一步该做什么",知识图谱关注的是"做之前该知道什么"。你现在完全可以先用 Neo4j 或者简单的向量库存知识,再在 Agent 的 Graph 节点里动态查询,两者通过 context 传递数据,没必要做成一个东西。
2.2 工具调用图管的是"技能路由"
工具调用图(Tool/Function Call Graph)是最容易和"Agent 图"混淆的。它描述的是"哪个工具的输出会作为哪个工具的输入",本质是数据依赖关系。典型例子是数据分析场景:检索工具产出原始数据,清洗工具吃下原始数据产出结构化表格,可视化工具吃下表格产出图表,这种一条线的路径用 DAG 就够。
但真实场景远比这复杂。同一个输入,可能同时需要走文本摘要和情感分析两条路径,再把结果合并;一个工具失败,可能要换一个备用工具重跑。如果工具调用图只支持 DAG,这类分叉合并就只能手写代码硬拗,非常痛苦。
我在 Harness 里把工具调用做成图上的"技能节点",每个技能节点内部可以有多个实现分支(比如一个"天气查询"技能节点,接三个不同的天气 API),运行时自动路由到当前可用且响应最优的那个。这个设计后面会详细展开。
2.3 执行流图才是 Harness 的本体
真正决定 Harness 架构走向的,是执行流图(Execution Flow Graph),也就是我把 Agent 的状态机、控制逻辑、编排策略画成的那张图。它会把一次完整的 Agent 任务定义成:
用户输入节点→ 接收用户请求,写入会话上下文规划器节点→ 大模型决定下一步调什么工具、访问什么子 Agent工具执行节点→ 调用具体工具,捕获结果,处理异常条件分支节点→ 根据上下文中的关键字段,决定走哪条边聚合节点→ 合并多个并行路径的结果终止节点→ 输出最终答案,清理会话状态
所有节点之间通过边连接,边的语义包括顺序、条件、并行、循环。整个 Harness 的运行时就干一件事:沿着图走,遇到节点执行,遇到边做路由判断。
建议大家在动手写代码之前,花半天时间把你自己的业务场景画到纸上。不需要特别规范,就用方框和箭头。我当时的图就画了六七版,画的过程里很多模糊的决策都自动清晰了。这也是为什么我喜欢用 Graph 而不是别的——它逼你把"流程"变成"看得见的东西"。
3. Harness 核心模块到底怎么设计
定好了"执行流图"这个底座,接下来就是拆分 Harness 的模块。我的做法是四层结构:图定义层、上下文管理层、运行时调度层、观测埋点层。每一层都对应一套独立的接口,彼此尽量解耦。
3.1 图定义层:让每个环节都成为一个节点
所有节点统一抽象成一个接口,不管是 LLM 调用、工具执行还是条件判断,都实现同样的生命周期。我用的 Python,接口长得像这样(伪代码):
class Node: name: str # 全局唯一节点名 inputs: List[str] # 需要从上下文中取的字段 outputs: List[str] # 执行后写回上下文的字段 def execute(self, ctx: Context) -> NodeResult: # ctx 是全局状态,节点从这里读数据、写数据 # NodeResult 包含:状态(成功/失败/跳过)、输出值、分支路由信息 pass这样做的好处非常多。比如统一抽象之后,工具调用和模型调用长得一样,图定义完全不用区分它们。你做调试的时候,可以像看普通程序一样沿着图遍历,每个节点都输入输出明确。做可视化也容易,直接把所有节点遍历出来渲染就行。
具体到实现,我定义了几种内置节点类型:
| 节点类型 | 职责描述 | 典型场景 |
|---|---|---|
LLM_NODE | 调用大模型,支持自定义 prompt 模板、模型参数 | 意图识别、内容生成、规划决策 |
TOOL_NODE | 调用外部工具/API,内置超时、重试、降级 | 天气查询、数据库访问、搜索 |
CONDITION_NODE | 根据上下文逻辑判断,返回branch路由 | 用户是否登录、错误是否可重试 |
AGGREGATOR_NODE | 合并多个父节点的结果,供后续节点使用 | 多路检索结果汇总 |
MEMORY_NODE | 读写长期记忆或短期状态 | 保存对话历史、关键信息抽取 |
每一种节点都要处理幂等和重入。Graph 架构里同一个节点可能会因为循环或重试被多次执行,如果节点内部状态不干净,就会出暗毛病。我的建议是节点对象保持无状态,用 Context 传数据,这样重入时只需要重新拉一次上下文,不会因为内存里的脏数据导致结果漂移。
3.2 上下文管理层:让状态在 Graph 上流动起来
这是整个 Harness 最容易翻车的部分。很多 Agent 系统出问题,不是大模型不够聪明,而是上下文里的数据要么没写对、要么写重复了,要么写进去之后没人清理。
我的方案是:Context 对象承载整张图的状态,分为几层:
session:会话级数据,包含用户 ID、会话 ID、长期目标request:当前请求级数据,包含本轮输入、中间结果、临时变量global:系统级常量,比如模型配置、环境变量、全局超时
每个节点执行前,调度层根据inputs从 Context 里取数据;执行后,根据outputs把结果写回 Context。取和写都走统一接口,这样就能非常容易地在上层做拦截,记录每一步的状态变化。
这里有一个非常关键的细节:Context 里的字段名必须全局唯一。我吃过亏,两个不同模块都往 Context 写了一个叫data的字段,结果下游节点读的时候拿到了错误的数据且没有任何报错,排查了一整个下午。后来我把所有字段命名统一成模块名.业务名.字段名这种格式,比如order.check.billing_address,再也没出过这类问题。
另外建议给 Context 加一个快照能力。每次节点执行前后保存一份上下文快照,一旦下游发现异常,可以快速回溯到上一个稳定状态,把整条链路"回放"一遍。对调试和线上排查来说,这个功能帮助极大。
3.3 运行时调度层:Graph 怎么被真实驱动起来
图定义好了,状态有了,还差一个"引擎"——沿图执行的那个调度器。这层看似简单,实际最难。
最朴素的做法是直接深度优先遍历,调用节点的execute,然后根据返回结果选择下一条边。这种实现能跑,但真实场景很快会崩溃:循环节点可能死循环,并行节点没有并发能力,某个节点挂了整张图就停摆。所以我在调度层加了三个机制:
- 并行执行:当一个节点有多个后继节点且边标记为
PARALLEL时,用线程池并发跑所有子节点,等全部完成后继续。比如"查天气 + 查日历 + 查航班"可以一次并发出去,把等待时间从 3 秒压到 1 秒。 - 超时与熔断:节点可以声明自己的超时时间(默认 10 秒)。超时后触发降级:可以重试一次,或者走一条"降级分支"(比如大模型超时后直接返回缓存命中结果)。
- 循环上限:给循环边加一个
max_epochs,用一个计数器记录当前循环次数,触顶直接跳出,防止 Agent 在一个死胡同里无限打转。
调度器还有个容易被忽略的职责是动态规划。当图里有"要做决策"的节点(就是 LLM_NODE 返回某个特定 action),调度器必须根据返回结果实时展开后续路径,而不是死板地走编译期确定的边。说白了,图是静态的、可解释的"地图",但 Agent 的实际路径是动态地沿着地图走出来的。上一轮模型决定走 A 分支,这一轮可能走 B 分支,调度器都要接得住。这是 Harness 和普通工作流引擎最大的区别:普通工作流完全静态,Agent 工作流必须留出模型决策的"活口"。
深度和优雅都比不上"能跑"。我建议调度器第一版不用做得太复杂,能按图走通普通顺序路径就够了,后面再逐步加并行、重试、动态跳转。不然一开始就整一个分布式调度引擎,大概率又是过度设计。
3.4 观测埋点层:把 Graph 上的每一步都变成日志
没有观测的 Agent 系统就是黑盒。用户在线上问一个问题,Agent 中间做了什么,为什么不调用这个工具,为什么改了主意,全靠猜。在 Graph 架构里,观测不是附加功能,成本天然低,你只需要在调度器里埋几个点就行。
具体我做了三类埋点:
- 节点生命周期日志:每个节点从开始到结束,记录耗时、输入输出摘要、返回状态。
- 边路由日志:记录从节点 A 到节点 B 的跳转原因,是顺序走、条件满足还是循环迭代。
- 上下文变更日志:每次 Context 写入,记录字段名、变化前后的值、当前节点名。
这三类日志合在一起,可以拼出一张完整的"Agent 行为时序图"。线上出了诡异问题,我一般先看边路由日志,看 Agent 在哪个节点上做了预期之外的选择,基本就能定位到是模型决策问题还是数据问题。如果发现模型频繁做出非预期决策,我会把该节点之前的上下文快照捞出来,直接喂给模型做复盘分析。
4. 实操实录:我给某跨平台系统搭的 Graph Harness
光说不练没有意义,我拿一个实际的模拟项目 X 来把上面的东西串起来。项目背景是某公司需要给内部客服团队做一个 Agent 系统,支持用户查订单、办退款、改配送地址。之前用 Chain 做的版本,一路"顺拐",用户问 A 它就答 A,用户在对话中突然换话题就直接崩了。所以这次我直接用 Graph 重写。
4.1 场景拆解:客服 Agent 的真实流程
把业务逻辑画成图之后,我发现客服流程本质上是一个"三层漏斗":
- 第一层:意图识别,判断用户到底要做什么
- 第二层:风险判断,比如退款要查黑名单、配送改址要查订单状态
- 第三层:工具执行,调具体系统的接口,收尾回复
三层之间都有条件跳转。比如查订单,用户没登录就要走"登录引导";登录了但订单 ID 不存在,要走"无订单提醒";订单存在但状态已取消,要走"异常流程"。如果中间某一步系统超时,还得能跳"降级话术",别让用户干等。
4.2 图定义:直接上代码
我定义的核心节点大概如下(Python 伪代码,去掉了业务细节,保留骨架):
graph = Graph(name="customer_service") graph.add_node(LLM_NODE( name="intent_detect", inputs=["user_query"], outputs=["intent"], prompt_template="识别用户意图,输出以下之一: order_query, refund_request, address_change" )) graph.add_node(TOOL_NODE( name="check_login", inputs=["user_id"], outputs=["login_status"], tool="auth_service.check_status", timeout_sec=3, )) graph.add_node(CONDITION_NODE( name="route_by_login", inputs=["login_status"], branches={ "logged_in": "route_by_intent", "not_logged_in": "login_guide", } )) graph.add_node(LLM_NODE( name="route_by_intent", inputs=["intent", "login_status", "session.memory"], outputs=["selected_handler"], prompt_template="根据意图选择handler..." )) graph.add_edge("start", "intent_detect") graph.add_edge("intent_detect", "check_login") graph.add_edge("check_login", "route_by_login") graph.add_edge("route_by_login", "route_by_intent", condition="login_status == 'logged_in'") graph.add_edge("route_by_login", "login_guide", condition="login_status != 'logged_in'")写图定义时我控制住了一种冲动:把所有小分支都堆到一个函数里。Graph 的最佳粒度是"一个步骤一个节点",把步骤拆得太粗(比如一个大节点同时做意图识别、参数提取、结果生成),后期调试就会来回在超长代码块里翻,累得半死。宁可多定义几个节点,每个节点职责单一。
4.3 跑通全链路:从模拟输入到最终输出
定义完图,直接用 Harness 的调度引擎跑了一把。输入是"你好,我想改一下昨天那个订单的配送地址"。预期路径是:意图识别节点判断为"改地址" → 检查登录状态(模拟已登录) → 根据意图选择 handler → 调用改地址工具 → 返回成功话术。实际跑下来,路径正确,单轮耗时 1.8 秒左右,后续我加了并发优化后能再降一点。
跑的过程中我刻意做了一个坏输入测试:把用户输入改成"我在你们这买的东西坏了,要赔钱"(恶意退款倾向)。这时意图识别判断为"refund_request",在后续风险判断节点里,从黑名单接口查到该用户 ID 有风险记录——Harness 立刻跳转到一个"转人工"节点,走的是和正常退款完全不同的路径。这一步测试既往逻辑验证了条件分支的可靠性,也验证了 Graph 架构对这种"业务逻辑异常分支"的可控性,这在链式实现里写起来相当别扭。
4.4 效果对比:Graph vs 之前的 Chain 版本
同样这个场景,旧 Chain 版本的代码大约 300 行,里面全是手写的 if-else 加状态变量,新增一个分支要把整条函数重新捋一遍。换成 Graph 之后,代码量降到约 150 行,新增一个分支只需要在图上加一个节点和两条边,功能清晰很多。更重要的是,每一次路由决策都有图的结构作参照,问题出现在哪个环节,一目了然,不需要靠打印日志一步步猜。
5. 工程化落地里的硬坑和排查实录
说点实战里最疼的部分。Graph Harness 不是写完就能跑得好,真正痛苦的是线上各种"站着都能踩"的坑。
5.1 死循环比你想的更容易出现
最典型的坑是"模型在循环里反复做同一个决策"。比如我那个客服 Agent,退款申请失败后,模型可能会重复尝试"重新提交退款",而不是跳转人工。如果不限制循环次数,这个节点就会无限转圈,把上下文窗口撑爆,费用也能在几分钟内烧掉几百块。
我的解法是给所有能形成回路的边加循环计数元数据,每次经过该边时递增计数并传入上下文。计数超过阈值时,调度器强制把模型决策的优先级调低,或者直接断掉这条路径,跳到兜底节点(比如"请求人工客服介入")。所有循环边在定义时就必须默认带上 max_epochs,不允许裸奔。
5.2 上下文越滚越大,模型越来越傻
Graph 里跑几轮任务后,上下文会积累大量历史数据,比如意图识别结果、工具返回的 JSON、中间步骤的思考过程。这些东西本来只在当前环节有用,却一直留在上下文里,导致每轮请求的 token 数量爆炸,模型注意力被噪音分散,回答质量明显下降。
我在 Harness 里加了一套上下文"减肥"机制:
INTERMEDIATE:本节点可用,执行完立刻标记过期,从待发送上下文中剔除PERSIST:后续可能需要的历史信息,保留一定轮次RESET:任务完成或会话切换时清空,防止串味
每次向模型发送 Prompt 前,运行时会按配置过滤上下文,只保留"当前节点需要的 + 必要的会话记忆"。一句话总结:不是所有上下文都该进模型,你必须显式区分"当前指令"与"长期记忆"。
5.3 排查问题全靠"图反演"
线上出问题时,我的标准排查流程是:把日志里的节点执行轨迹拉出来,对照图定义还原成一张实际走过的路径图(我管这叫"图反演"),然后标注哪些节点超时、哪些节点走了非预期分支。90% 的问题,看了这张实际轨迹图就能肉眼定位。
有一次用户报"Agent 答非所问",我反演出来的路径是:意图识别节点把"查询退款进度"误判成了"发起退款申请",导致下游调了一个写操作接口。根因不是 Graph 的问题,而是意图识别节点的 prompt 模板里没有加"查询与申请必须区分"的指令。我改了一版 prompt,加了 few-shot 示例,第二天误判率直接下降了一半。这种问题,如果不用图反演的方法,很可能要翻几十条日志才能想到是模型分类错了。
5.4 节点故障的隔离与降级
Graph 上任意一个节点崩了,不能整个系统跟着崩。每个节点执行时,我强制包了一层"舱壁",把异常捕获、重试策略、降级策略全部做在框架层,节点业务代码不允许自己 try-catch 后吞掉异常。工具调用超时了,框架自动走降级分支;模型接口 5xx,框架自动重试一次;重试还不行,走"兜底话术节点"。
我把所有节点的故障处理统一成以下策略:
| 故障类型 | 处理策略 | 兜底路径 |
|---|---|---|
| 工具超时 | 重试一次,间隔 500ms | 降级话术,或换备用工具 |
| 模型 API 返回 5xx | 重试两次,指数退避 | 走缓存答案,或转人工 |
| 上下文异常/缺字段 | 抛出ContextError,中止路径 | 返回到最近的可信节点 |
| 节点抛出未捕获异常 | 框架捕获并标记节点失败 | 跳 "failure_handler" 节点 |
6. 性能调优与后续扩展方向
Graph Harness 跑稳了之后,我开始琢磨性能。主要有三条路:并行化、缓存、子图嵌套。
6.1 并行节点:把等待变成并发
客服场景里,"查订单 + 查用户会员等级 + 查最近售后记录"这三个操作其实互相独立。我把它们标成PARALLEL,用线程池同时跑,时间从 2.5 秒压到 0.9 秒。实现不复杂,但注意要处理好线程安全和 Context 写入的原子性。
6.2 节点级缓存:别让模型重复烧钱
很多工具节点返回的数据在短时间内不会变,重复调用完全浪费。我给节点加了一个cache_ttl参数,比如check_login的结果缓存 60 秒,get_order的结果缓存 5 分钟。命中缓存时直接跳过节点执行,把 token 费和响应时间都省下来了。
6.3 子图嵌套:让图保持可读性
图越来越大的时候,所有节点摊在一层就变成意大利面了。我把"退款流程"、"改地址流程"各打包成一个子图,在顶层图里用一个"子图节点"引用。顶层图只关注宏观路由,细节进子图去查。这样既保住了 Graph 的可解释性,也不会因为图太大没法看。
7. 写在最后的经验
Graph 架构的 Harness 做下来,我的最大体会是:它帮你把"不可见的逻辑"变成"可检查的路径"。Agent 系统最怕的不一定是模型笨,而是连你自己都不知道它每一步在做什么。Graph 解决了这个问题的底层结构,这是一个好框架该有的样子。
给准备动手做的同学三个建议。第一,不要一上来就追求完美,先用最朴素的图结构跑通一个真实场景,感受一下"原来流程可以画出来"的感觉。第二,上下文字段命名一定要加前缀,全局唯一,这是无数血泪换来的教训。第三,日志一定要带着节点名和路径信息,不然 Graph 的优势根本发挥不出来,排查问题还是会退化回大海捞针。
做 Agent 这件事,天花板不在模型能力,而在你把"编排"这件事做得有多扎实。Graph 是一个极好的思维工具。