用Graph Engineering把SOP画成可执行图：Agent流程控制实战

之前做客服工单自动处理 Agent 时需求看起来很简单把客服团队的 SOP 文档丢给大模型让它自动分类、自动退款、帮技术排查最后质检完自动回访。结果真正落地才发现最难的既不是 Prompt也不是模型能力而是流程本身——怎么让 Agent 严格按 SOP 办事中间分支不乱、循环不迷路、每一步都可追踪。试过让大模型直接读一大段 SOP 自由发挥效果不稳定同一个问题上午处理正确下午就跑偏了。后来换了个思路把 SOP 画成一张 Graph让 Agent 在这张图上按路径执行。问题一下子清晰了很多——节点就是动作边就是流转条件状态就是整张图共享的“工作台”Agent 只是在图上做一次路线明确的旅行。本文就把这套“Graph Engineering”思路完整拆解一遍从核心概念讲到可运行的实战案例最后补充生产落地时常见的问题和最佳实践。无论你是想给自己写的 Agent 增加流程控制能力还是想把团队里冗长的 SOP 文档固化成一个自动化引擎这篇文章都值得收藏备用。1. 为什么要给 Agent 画一张可执行的图1.1 直接把 SOP 塞给模型为什么不可靠先看一个典型场景。客服团队有一份处理退款的标准流程收到工单 → 判断意图 → 如果是财务问题看金额是否超过阈值 → 小金额自动退款大金额转人工审批 → 所有工单质检 → 回访关闭。如果用纯 Prompt 的方式通常是这样写的你是客服助手请严格按照以下 SOP 处理工单 1. 判断客户问题属于财务、技术还是其他。 2. 财务问题金额小于等于1000元自动退款大于1000元转人工。 3. 技术问题按技术排障流程处理。 4. 所有工单最后都要质检质检通过后关闭工单。 ...这种写法有三个非常现实的问题分支不稳定。模型可能自己“发挥”出 SOP 里没有的步骤也可能忽略某个关键判断。循环不可控。如果模型反复要求用户补充信息就可能在某个环节无限兜圈子。不可观测。出了问题你只能看到对话内容看不到它“走到”SOP 的哪一步很难排查。也就是说流程控制权在纯 Prompt 方案里实际上交给了模型而不是交给了开发者。这就是不稳定的根源。1.2 Graph Engineering 是什么Graph Engineering 的思路很简单把 SOP 从“给人看的文字”变成“给机器执行的图”。一个可执行图由以下几个部分组成节点Node某个具体动作例如“工单分类”“财务退款”“质检”。边Edge节点之间的流转关系表示“做完这一步下一步做什么”。条件边Conditional Edge根据状态决定走哪条边例如“金额大于1000转人工”。状态State所有节点共享的数据容器例如工单号、客户名、分类结果、处理结果。Agent 运行时从入口节点开始按图的结构一步步执行每到一个节点就调用对应的函数函数处理完毕后返回一个路由结果图引擎根据路由结果决定下一步走向。这样做的好处是流程显式化。SOP 是什么样图就是什么样开发和业务方都能看懂。控制权回到开发者。模型的职责被限制在“节点内部”比如只做意图分类不能自己改变流程方向。天然可观测。每一步执行了什么、路由到哪个节点都能记录下来。可测试可回滚。每个节点是独立函数可以单独写单测。1.3 一张图能解决哪些实际问题我总结下来下面几类场景最值得引入 Graph Engineering智能客服工单处理分类、处理、质检、回访每个环节都是明确的动作分支依赖业务规则。内容审核 Agent初筛、敏感词命中、模型二次审核、人工复审、结果归档。自动化运维排查监控告警 → 定位服务 → 检查日志 → 执行恢复操作 → 通知负责人。多步工具调用 Agent查天气、订酒店、发通知每一步都可能需要调用不同外部 API且顺序固定。业务流程类 RPA下载文件、清洗数据、入库、发送报表本质上也是图。当然并不是所有场景都要模型。如果流程简单到只有两三个步骤用 if-else 就够了但如果流程有多个分支、多个兜底、多个终态Graph 就是比散装代码更合适的抽象方式。2. 环境准备与工程选型2.1 运行环境说明本文核心示例是纯 Python 实现的轻量图引擎不依赖任何第三方库用的是 Python 标准库因此环境要求非常简单环境项建议操作系统Windows / macOS / Linux 均可Python 版本Python 3.10 及以上依赖包无示例只用标准库IDEVS Code、PyCharm 等任一 Python IDE 即可生产框架样例可选择 LangGraph 等框架版本以官方最新文档为准需要说明的是本文示例的重点是让你理解 Graph Engineering 的完整思路所以用一个可复制的轻量引擎演示流程控制。放到生产环境时可以迁移到成熟的图编排框架例如 LangGraph。这类框架在概念上与本文自定义引擎一一对应但 API 和特性随版本变化较快请以官方文档为准。2.2 两种落地路径在实际项目中你有两种选择路径一业务链路简单团队规模小先用轻量引擎。自己维护几十行的图运行器能解决 80% 的流程控制需求模型只负责节点内部的理解和判断。优点是依赖少、逻辑透明、方便二开。路径二流程复杂、并发量大、需要持久化和多 Agent 协作直接用生产级框架。例如 LangGraph 这类框架自带图保存、断点恢复、超时控制、并行分支等功能能省去很多自研成本。但相应的学习成本、版本迁移成本也需要评估。本文先带你把轻量引擎跑通再看生产框架的能力对照。2.3 示例项目目录为了演示清晰示例只用一个文件agent_graph_demo/ └── sop_graph_demo.py如果你想工程化一点可以按下面的目录拆分agent_graph_project/ ├── graph/ │ ├── engine.py # 图执行引擎 │ ├── nodes.py # 所有节点函数 │ └── router.py # 路由判定逻辑 ├── sop/ │ └── support_sop.py # 客服工单 SOP 定义与图构建 ├── tests/ │ └── test_support.py # 节点与路径测试 └── main.py # 入口3. 图工程的四个核心要素3.1 Node封装一个最小动作节点是图的最小执行单元。一个合格的节点函数应该满足三个原则单一职责一个节点只做一件事比如“分类”就只分类“退款”就只退款。无副作用节点不直接改数据库、不直接发消息而是把结果写入 State由专门的节点负责外部副作用。返回路由键节点处理完返回一个字符串作为路由键交给图引擎。先看一个最简单的节点写法def classify(state: dict) - str: content state[content] if 退款 in content or 发票 in content: state[intent] finance return finance if 接口 in content or 崩溃 in content: state[intent] tech return tech state[intent] other return other这里节点函数接收state读它需要的字段写它负责的字段最后返回一个字符串路由键。这个键会被图引擎用来查找下一步节点。把节点设计成“纯函数”有很多好处便于测试、便于单人维护、便于在并行节点之间隔离状态。3.2 Edge显式定义转移关系普通边表示确定性的转移A 做完必然走 B。例如“质检完成后必然进入收尾节点”。graph.add_edge(qa, end)普通边的特点是无条件。只要 A 节点正常执行完就走 B。适合表达顺序型流程。条件边则表示带条件的转移路由键由当前节点写入状态图引擎根据这个键查映射表确定下一跳graph.add_conditional_edges( classify, lambda state: state.get(intent), { finance: finance_handle, tech: tech_handle, other: human_fallback, } )3.3 State所有节点共享的“工作台”State 是图里所有节点共享的数据容器。别把它理解成一个简单的字典它更像是整张图的“工作台”任何节点处理产生的中间结果都会被放到工作台上被后续节点读取。这里有一个关键设计决策State 应该分层设计。我建议至少分成三层入参层工单号、客户内容、金额一次任务开始时就确定。中间层意图分类、处理器、退款状态由各节点写入。结果层质检评级、最终状态流程结束时沉淀下来。分层的好处是调试方便。你打印一次 State就能区分哪些是业务输入、哪些是中间过程、哪些是最终产物。state { ticket_id: T001, customer: 张三, content: 我在 App 上申请退款订单金额 800 元但一直没到账。, amount: 800, # 下面是中间状态 intent: , handler: , refund_status: , # 下面是最终输出 qa_result: , status: , }3.4 条件路由与循环终止条件路由是 Graph Engineering 的灵魂。它让图能够根据不同输入走向不同分支小金额自动退款大金额转人工审批无法识别的走兜底。不过条件路由容易引入一个经典问题循环。比如你希望节点在信息不足时返回“补充信息”路由让流程回到“询问用户”节点这实际上就构成了一个环。图工程对循环的态度是“允许存在但必须可控”。常见的控制手段有两种最大步数兜底。引擎设定最大执行步数超过即终止并告警。显式终止节点。每个分支最终必须收敛到一个终态节点不能无限循环。可以这样理解Graph 允许有环但环必须“有出口”。好图是把“正常路径”和“异常绕圈”严格区分的。4. 完整实战把客服工单 SOP 画成 Graph4.1 原始 SOP 拆解假设客服团队的退款工单处理 SOP 如下1. 客服系统收到工单后先记录工单信息。 2. 对工单内容做意图分类 - 财务问题进入财务处理流程。 - 技术问题进入技术处理流程。 - 其他问题直接转人工兜底。 3. 财务处理流程 - 退款金额 1000 元自动退款进入质检。 - 退款金额 1000 元转人工审批进入人工兜底。 4. 技术处理流程按“复现 - 定位 - 修复 - 回归”的步骤处理进入质检。 5. 人工兜底分配给人工客服进入质检。 6. 质检根据是否人工介入给评级。 7. 工单关闭发送满意度回访。这段 SOP 用自然语言写出来很长而且有多个分支直接在 Prompt 里描述很容易漏条件。把它画成图之后表达变得非常紧凑。4.2 图设计与节点清单整个图一共有 7 个节点、1 个条件分类路由、1 个财务分支路由和 4 条普通边。节点清单如下节点名职责后续路由receipt记录工单信息普通边到 classifyclassify识别意图finance / tech / other条件边到三个节点finance_handle财务处理自动退款或人工审批条件边到 qa / human_fallbacktech_handle技术排障处理普通边到 qahuman_fallback人工兜底普通边到 qaqa质检评级普通边到 endend工单关闭回访终止节点这个图的 ASCII 简图如下receipt | v classify ------------ (other) ---- human_fallback | | | (finance) v v qa finance_handle --------------------- | | | | (auto_refunded) | ----------- qa ----------------- end | | (pending_manual) ----------- human_fallback4.3 实现一个轻量 Graph Runner下面用不到 60 行代码实现一个通用图执行引擎。它支持三种要素add_node注册节点函数。add_edge注册普通边。add_conditional_edges注册条件边传入路由函数和映射表。class SOPGraph: 轻量图执行引擎通过普通边和条件边驱动流程。 用法 graph SOPGraph(entryreceipt) graph.add_node(receipt, receipt_fn) graph.add_edge(receipt, classify) graph.add_conditional_edges(classify, router_fn, mapping) graph.run(state) def __init__(self, entry: str): self.entry entry self.nodes {} self.default_edges {} self.conditional_edges {} def add_node(self, name: str, fn): self.nodes[name] fn def add_edge(self, from_node: str, to_node: str): self.default_edges[from_node] to_node def add_conditional_edges(self, node: str, router_fn, mapping: dict): self.conditional_edges[node] (router_fn, mapping) def run(self, state: dict, max_steps: int 10): current self.entry state[flow] [] print(f[Start] 入口节点: {current}) for _ in range(max_steps): if current is None: break node_fn self.nodes.get(current) if not node_fn: print(f[Error] 节点未注册: {current}) break # 执行节点函数拿到路由键 route_key node_fn(state) state[flow].append(current) print(f[Node] {current} - 返回: {route_key}) # 优先判断条件路由 if current in self.conditional_edges: router_fn, mapping self.conditional_edges[current] if route_key not in mapping: print(f[Warn] 路由键 {route_key} 不在映射表中流程终止) break next_node mapping[route_key] print(f[Router] {current} 条件路由: {route_key} - {next_node}) current next_node else: next_node self.default_edges.get(current) print(f[Edge] {current} 普通边 - {next_node}) current next_node else: print([Warn] 达到最大步数限制流程可能未正常终止) print(f[End] 执行路径: { - .join(state[flow])}) return state这个引擎很简单但已经具备真实图引擎的核心能力节点分发、普通边、条件边、兜底告警、路径记录。4.4 实现工单处理节点接下来实现 7 个节点函数。每个节点都会在 State 中写入自己的处理结果并通过返回值暴露路由键。def receipt(state: dict) - str: 节点记录工单信息 print(f 收到工单: {state[ticket_id]} | 客户: {state[customer]}) print(f 问题内容: {state[content]}) state[status] received return ok def classify(state: dict) - str: 节点识别工单意图 content state[content] if 退款 in content or 发票 in content: state[intent] finance print( 识别意图: 财务问题) return finance if 接口 in content or 崩溃 in content or 超时 in content: state[intent] tech print( 识别意图: 技术问题) return tech state[intent] other print( 识别意图: 其他问题) return other def finance_handle(state: dict) - str: 节点财务处理 amount state.get(amount, 0) if amount 1000: state[handler] 财务自动退款 state[refund_status] auto_refunded print(f 财务SOP: 金额 {amount} 元 1000自动退款) return auto_refunded else: state[handler] 财务人工审批 state[refund_status] pending_manual print(f 财务SOP: 金额 {amount} 元 1000转人工审批) return pending_manual def tech_handle(state: dict) - str: 节点技术排障 state[handler] 技术支持 state[resolution] 复现 - 定位日志 - 修复接口超时 - 回归验证 print(f 技术SOP: {state[resolution]}) return ok def human_fallback(state: dict) - str: 节点人工兜底 state[handler] 人工客服 state[status] manual print( 兜底SOP: 已转人工客服预计 1 小时内响应) return ok def qa(state: dict) - str: 节点质检 grade A if state.get(refund_status) ! pending_manual else B state[qa_result] f质检通过, 评级{grade} print(f 质检SOP: {state[qa_result]}) return ok def end_node(state: dict) - str: 节点工单关闭 state[status] done print( 收尾: 工单关闭发送满意度回访) return ok注意这里每个节点都只是修改 State 字典并没有真正调用退款 API 或发送短信。真实项目里这些外部副作用应该放在独立的能力层由节点去调用而不是把连接数据库的代码散落在流程控制逻辑中。4.5 构建图并运行四个 Case下面把节点注册到图里并配置边graph SOPGraph(entryreceipt) graph.add_node(receipt, receipt) graph.add_node(classify, classify) graph.add_node(finance_handle, finance_handle) graph.add_node(tech_handle, tech_handle) graph.add_node(human_fallback, human_fallback) graph.add_node(qa, qa) graph.add_node(end, end_node) # 普通边顺序执行 graph.add_edge(receipt, classify) graph.add_edge(tech_handle, qa) graph.add_edge(human_fallback, qa) graph.add_edge(qa, end) # 条件边意图分类 graph.add_conditional_edges( classify, lambda state: state.get(intent), { finance: finance_handle, tech: tech_handle, other: human_fallback, } ) # 条件边财务分支 graph.add_conditional_edges( finance_handle, lambda state: state.get(refund_status), { auto_refunded: qa, pending_manual: human_fallback, } )定义好图之后用四个 Case 覆盖不同的业务路径def run_case(title: str, state: dict): print( * 60) print(fCase: {title}) print( * 60) result graph.run(state) print(f最终状态: {result}) print() case1 { ticket_id: T001, customer: 张三, content: 我在 App 上申请退款订单金额 800 元但一直没到账。, amount: 800, } run_case(财务小金额自动退款, case1) case2 { ticket_id: T002, customer: 李四, content: 调用订单接口时报错服务直接崩溃急, } run_case(技术问题走技术SOP, case2) case3 { ticket_id: T003, customer: 王五, content: 我的账号最近登录有点奇怪你们能帮我看看吗, } run_case(无法识别走人工兜底, case3) case4 { ticket_id: T004, customer: 赵六, content: 发票金额 1500 元财务需要给我退款。, amount: 1500, } run_case(财务大金额人工审批, case4)4.6 运行结果与路径说明四个 Case 的预期执行路径如下Case标题执行路径Case 1财务小金额自动退款receipt - classify - finance_handle - qa - endCase 2技术问题走技术SOPreceipt - classify - tech_handle - qa - endCase 3无法识别走人工兜底receipt - classify - human_fallback - qa - endCase 4财务大金额人工审批receipt - classify - finance_handle - human_fallback - qa - end以 Case 4 为例控制台输出大致如下 Case: 财务大金额人工审批 [Start] 入口节点: receipt 收到工单: T004 | 客户: 赵六 问题内容: 发票金额 1500 元财务需要给我退款。 [Node] receipt - 返回: ok [Edge] receipt 普通边 - classify 识别意图: 财务问题 [Node] classify - 返回: finance [Router] classify 条件路由: finance - finance_handle 财务SOP: 金额 1500 元 1000转人工审批 [Node] finance_handle - 返回: pending_manual [Router] finance_handle 条件路由: pending_manual - human_fallback 兜底SOP: 已转人工客服预计 1 小时内响应 [Node] human_fallback - 返回: ok [Edge] human_fallback 普通边 - qa 质检SOP: 质检通过, 评级B [Node] qa - 返回: ok [Edge] qa 普通边 - end 收尾: 工单关闭发送满意度回访 [Node] end - 返回: ok [End] 执行路径: receipt - classify - finance_handle - human_fallback - qa - end 最终状态: {ticket_id: T004, customer: 赵六, content: 发票金额 1500 元财务需要给我退款。, amount: 1500, status: done, flow: [...], intent: finance, handler: 人工客服, refund_status: pending_manual, qa_result: 质检通过, 评级B}注意观察同一个图因为输入不同走的路径完全不同。这正是 Graph Engineering 的价值——同一套 SOP不同分支自动分流流程全程可见。四类 Case 对应的业务结论也很有意义财务小金额自动退款评分 A自动化率高。技术问题走技术处理 SOP评分 A。无法识别不硬猜转人工兜底避免误操作。财务大金额转人工审批守住资金安全边界。5. 常见问题与排查思路Graph 跑起来之后最常遇到的坑其实不在模型而在流程本身。下面把高频问题整理成表格方便对照排查。问题现象常见原因解决思路流程执行到某节点后突然停止条件路由的路由键不在映射表中打印节点返回值检查路由键拼写补全映射关系每次都走到同一个分支另一个分支永远不触发判断条件写死或输入数据没有对应字段检查 State 中判断字段是否被上层节点正确写入图陷入死循环一直重复某个节点条件边构成环路且缺少终止节点设置 max_steps 兜底在环上增加显式出口多个 Case 之间状态互相污染多个任务复用了同一个 State 引用每次任务新建 State必要时深拷贝初始数据节点函数抛异常整个流程中断节点缺少异常处理节点内捕获异常写入 state[error] 并路由到兜底节点模型节点返回“人话”而不是枚举值模型自由文本输出导致路由键不匹配用结构化输出约束模型或增加一层归一化映射并发执行时状态错乱共享全局变量或同一个 State 对象每个任务隔离 State数据库持久化中间状态质检结果和实际处理结果对不上质检节点读取的字段被后续节点覆盖约定字段只写一次质检节点只读不做覆盖下面挑两个高频问题展开一下。5.1 路由键不在映射表里这是最常见的入门问题。比如 classify 节点里你写了return tech但映射表里写的是技术那图引擎就会告警并终止。排查顺序打印route_key确认节点实际返回了什么。打印mapping确认映射表里有哪些键。两者不一致要么改节点返回值要么改映射表。这就是我前面强调“节点返回枚举值”的原因。只要是可控枚举路由就稳定一旦出现自由文本稳定性立刻下降。5.2 流程循环难以结束如果某个节点判断“信息不足”就回到上一个节点很容易形成环。例如ask_user - check_info - ask_user - check_info - ...解决方案有两种层级引擎层加 max_steps超限终止并告警。设计层设计“确认上限”。比如信息最多补充 3 次超过就转人工。正确的做法是两层同时做。引擎层兜底保护设计层避免无意义循环。6. 最佳实践从 Demo 走向生产自研轻量引擎跑通之后要上生产还需要考虑很多事情。下面的建议都来自实际踩坑经验按优先级排序。6.1 State 设计是第一优先级一句话图跑得好不好多半取决于 State 设计得好不好。我建议给 State 建一个明确的数据结构。在 Python 中可以用 dataclass也可以用 dict 加字段命名规范。关键是每个字段都搞清楚谁写入、谁读取、何时结束写入。例如from dataclasses import dataclass, field dataclass class TicketState: # 入参层 ticket_id: str customer: str content: str amount: float 0.0 # 中间层 intent: str handler: str refund_status: str # 结果层 qa_result: str status: str flow: list field(default_factorylist) error: str 用 dataclass 比裸字典更安全因为字段在创建时就固定下来不容易出现“拼错 key”这种低级问题。6.2 让每一步都可以观测生产环境中图的执行路径必须完整记录。至少要在以下几个维度打日志节点开始时间、结束时间、执行耗时。节点返回值。路由选择结果。关键 State 字段的变更。可以把 Flow 记录落到日志系统或数据库每个工单对应一条路径记录。这样出问题的时候把工单 ID 拉出来就能看到它当时走了哪条路、在哪个节点耗时最长、卡在哪个路由上。6.3 幂等、超时与重试Agent 的节点经常要调用外部 API而外部调用天然不稳定。有三个问题必须考虑超时每个外部调用都要设置超时不能无限等待。重试对“临时性错误”做重试对“业务性错误”直接转兜底。幂等同一个节点执行两次结果必须一致。例如退款节点如果重复执行不能让用户收到两次退款。解决办法是在 State 中保存每个节点的执行状态例如nodes_executed列表节点执行前先判断是否已执行过。更简单的做法是让节点函数保证“重复执行无副作用”。6.4 安全边界与人工兜底涉及资金、数据删除、权限变更这类高风险动作时一定要守住安全边界。前面案例中的“大金额转人工审批”就是最典型的一种兜底。更进一步的实践是引入“审批节点”概念图运行到审批节点后暂停等待人工在管理台点击“通过”图再从该节点继续往下走。apply_refund - approval_wait - (通过) - execute_refund - end \- (驳回) - notify_user - end在自研引擎里实现“暂停”需要 checkpoint 机制也就是把 State 持久化到数据库下次从数据库恢复节点。用生产框架如 LangGraph时会轻松很多因为框架原生支持断点恢复。6.5 不要在图中写死业务参数最后一个小建议图结构本身用于表达流程拓扑不要把业务阈值写死在节点代码里。比如“1000 元”这个退款阈值更适合放在配置中心或数据库里运行时读取。amount_threshold config.get(finance.refund_threshold, default1000)这样运营人员调整阈值时不需要改代码、不需要重新发布图修改配置即可。图的稳定性和业务的灵活性就能同时保住。7. 总结与后续学习路线这篇文章的核心是把“Agent 按 SOP 执行”这件事从纯 Prompt 方案转成了 Graph Engineering 方案。你看到的关键点可以归纳为三条节点函数负责做事返回路由键不直接决定流程走向。图引擎负责走路根据路由键和映射表显式控制每一步去向。State 负责记账入参、中间结果、最终产物都沉淀在工作台里全程可观测。示例里那个 60 行的轻量引擎已经把普通边、条件边、环路兜底、路径记录这些核心能力都实现了。你可以直接拿它跑一下四个 Case观察同一个图针对不同输入如何走出不同路径。这一步理解透彻之后再去看 LangGraph 等生产级框架的文档你会发现概念几乎可以无缝迁移StateGraph 对应图结构add_node 对应节点注册add_conditional_edges 对应条件路由checkpointer 对应断点恢复。如果接下来想继续深入建议按这个顺序练习给示例图增加一个“补充信息”节点故意制造循环再通过 max_steps 和次数限制控制它。把示例中的意图分类节点替换成真实的大模型调用但用枚举输出约束路由键。给图增加数据库存储实现 State 持久化和中途恢复。尝试在工作流里接入真实的退款 API并验证幂等性。画一张你自己业务中最复杂的 SOP 图先画在纸上再用代码实现。Graph Engineering 并不是什么高深理论它只是把“流程控制”这件事做得更工程化而已。先别急着上复杂框架最有效的方式是拿你手头最熟悉的一条 SOP按本文的 demo 先画一遍、跑一遍把每条路径的输出看清楚你自然就理解了“把 SOP 画成 Graph它就能自己跑”这句话的含义。