Graph Engineering:用图控制Agent执行SOP的工程实践
只讲原理和概念是不解决问题的。最近在 Agent 开发社区里,Graph Engineering 这个词出现的频率越来越高,很多团队开始把自己沉淀的业务 SOP 直接画成 graph,然后让 Agent 自己沿着图跑完整个流程。这个方向值得认真拆一下。
我的核心判断是:把 SOP 画成 graph,真正的价值不在于“可视化”,而在于把 Agent 的执行路径从“模型自由发挥”变成“工程可控”。如果你的 Agent 一直停留在“demo 能跑、上线就乱”的状态,大概率是流程控制层出了问题,而不是模型能力不够。
1. 这篇文章真正要解决的问题
很多人做 Agent 会遇到一个典型困境:单轮对话或者简单任务,GPT 类模型表现很好,但一旦涉及“先做 A,再根据 A 的结果判断是否做 B,如果 B 失败则走 C 分支”这种真实业务 SOP,模型就会开始“自由发挥”——漏步骤、跳顺序、自己发明不存在的操作。
于是出现了两条技术路线:
- 提示词硬约束:把 SOP 写进 system prompt,要求模型遵守。这不是不行,但效果不稳定。长 SOP 容易被截断,模型在长上下文里容易遗忘前面的约束,出了错也难定位。
- 代码硬编码:用 if-else 把流程写死。可控性确实强,但业务一变化就要改代码发版,工程成本很高。
Graph Engineering 提供的是第三条路:把 SOP 显式建模成一张有向图,Agent 每一步做什么由图的节点决定,节点之间的跳转由图结构或模型决策决定。这样流程可控、可视、可改,而且图和代码分离,业务调整不用动代码逻辑。
这篇文章会从概念原理、环境准备、完整代码实现到常见问题排查,完整演示如何把一条电商售后 SOP 用 graph 表达,并让 Agent 自己跑起来。
适合的读者:
- 正在用 LangChain、LlamaIndex 或自研框架做 Agent,但觉得流程不可控的同学。
- 团队沉淀了很多业务 SOP,想把这些文档变成可执行自动化流程的技术负责人。
- 面试被问到 Agent 架构,需要从“只会调 API”提升到“懂编排设计”的开发者。
简单说,这篇文章帮你把“流程控制”这个 Agent 开发里最容易被忽略、却是生产环境最关键的问题讲透。
2. 基础概念与核心原理
在深入代码之前,先把几个核心概念讲清楚。这些概念在面试和实际开发中都经常被混淆。
2.1 Graph Engineering 是什么
Graph Engineering(图工程)不是一个新框架,也不是某个具体工具,而是一种把流程、关系、依赖建模为图结构,并围绕图进行设计、开发、运行和运维的工程方法。
在 Agent 场景下,图工程的核心工作包括:
- 把业务 SOP 拆解为节点和边。
- 定义每个节点的执行逻辑和数据输入输出。
- 设计节点之间的跳转条件。
- 保证图可监控、可追溯、可恢复。
通俗地说:Graph Engineering 是给 Agent 画了一张“地图”,模型负责在每个路口做判断,但路线是预先设计好的。
2.2 SOP 在 Agent 场景下的含义
传统制造业里 SOP 是标准作业程序,比如“装配工艺卡片”规定了每道工序怎么做、用什么工具、什么标准算合格。
在 Agent 场景下,SOP 同样存在,只是工序变成了:
- 调用什么工具。
- 查询什么数据。
- 根据什么条件决定下一步。
- 什么情况下需要人工介入。
比如一个电商售后 Agent 的 SOP 大致是:
- 用户发起退款申请。
- 检查订单是否在可退款时间内。
- 如果在,自动同意退款并通知仓库拦截发货。
- 如果不在,引导用户走售后单流程。
- 如果用户投诉升级,转人工客服。
这样一条 SOP,用文字写出来是一段话,用图表示就是一个有分支的流程网络。图的好处是:每个判断点都清楚,每个出口都有归宿,不会出现模型“自由发挥”的中间态。
2.3 Agent 的编排模式:从 Chain 到 Graph
早期 LangChain 的经典抽象是 Chain——一条线性的链,A 做完传给 B,B 做完传给 C。对于线性流程够用,但真实业务几乎都有分支和循环。
Graph 编排的核心表达能力在于:
- 分支:根据条件选择不同的下一个节点。
- 循环:处理结果不合格时重试。
- 并行:多个相互独立的节点同时执行。
- 汇聚:多个分支的结果汇总后继续。
用一个对比表格更清楚:
| 维度 | Chain | Graph |
|---|---|---|
| 流程结构 | 线性、无分支 | 有向无环或有循环 |
| 表达力 | 简单流程够用 | 适合复杂真实业务 |
| 可观测性 | 执行路径单一 | 路径多样,需要追踪 |
| 可扩展性 | 新需求常需要重构 | 新节点可灵活插入 |
| 调试难度 | 定位快 | 需要用路径追踪 |
LangChain 后来也意识到了这个问题,从 Chain 迁移到 LangGraph,本质上就是把流程从“线性脚本”升级为“图状态机”。其他主流 Agent 框架也都越来越多地引入图模型,这已经是行业共识。
2.4 范式转变:从“提示词控制”到“图控制”
很多开发者的第一反应是:模型能力这么强,直接把 SOP 写进提示词不就行了吗?
这里要理解一个问题:提示词是软约束,图是硬约束。
- 提示词控制:模型“应该”这么做,但不保证。
- 图控制:模型“只能”在这些节点之间跳转,没有别的路。
经典的 Agent 架构往往是“这里插入图片”和“这里插入代码”,实际上图工程的核心就是把这层控制从模型内部“搬”到模型外部。这也是为什么 LangGraph 等工具这么重视 graph 设计——生产级 Agent 的可控性来源,主要是执行框架,而非模型本身。
过去我们习惯让模型来主导流程,但工程上更稳妥的做法是让工程主导流程、模型处理局部决策,把这两个环节解耦,各自做自己最擅长的事。
3. 图的节点类型与设计原则
把 SOP 画成 graph,首先要理解图上有哪些类型的节点。不同类型的节点执行逻辑不同,职责也不同。
3.1 常用节点类型
| 节点类型 | 英文常见叫法 | 职责 | 例子 |
|---|---|---|---|
| 开始节点 | start | 接收输入,初始化状态 | 接收用户退款申请 |
| 执行节点 | node / action | 调用工具或业务逻辑 | 查询订单状态 |
| 判断节点 | condition / router | 根据条件选择分支 | 订单是否超时 |
| 模型节点 | llm | 调用大模型做生成或分类 | 判断用户情绪是否升级 |
| 工具节点 | tool | 调用外部 API 或函数 | 调用支付系统退款接口 |
| 汇合节点 | join | 合并多个分支结果 | 汇总审核结果 |
| 结束节点 | end | 组织最终输出 | 返回用户退款结果 |
不需要每个图都用上全部类型,但判断节点和工具节点是最值得花心思设计的。
3.2 设计原则
结合我看到的团队踩坑经验,设计 Agent 的 SOP 图时,以下四个原则很重要。
原则一:节点粒度适中。一个节点只做一件事。比如“检查订单状态并发送通知”应该拆成两个节点,因为发送通知可能失败,检查订单状态也可能失败,两者不应该耦合。
原则二:显式处理异常路径。很多开发者只画“正常流程”的边,忽略了失败分支。真实生产环境中,工具调用超时、API 返回异常、模型输出格式不对都会发生。图中必须显式画出这些异常路径。
原则三:避免过长的链。一段流程如果连续超过 5 个单纯串行节点,考虑是否有可以并行或合并的步骤。过长的链会让单次请求延迟变高,调试也更困难。
原则四:状态设计要清晰。图的状态(state)是所有节点共享的数据总线。要明确定义哪些状态字段是节点间传递的,哪些是全局共享的,避免节点间通过“隐式变量”传递数据。
4. 环境准备与前置条件
下面进入实操部分。我们用一个 Python 最小实现来演示 Graph Engineering 的核心思想,不依赖任何重量级框架。
4.1 运行环境
- Python 3.9 或以上版本。
- 操作系统不限,Windows、macOS、Linux 都可以。
- 不需要 GPU,纯 CPU 环境即可跑通。
- 如果需要演示真实的大模型调用节点,需要配置 OpenAI 兼容的 API Key。如果不想调用外部模型,可以用规则模拟模型节点,流程依然可以跑通。
说明:本文重点演示“图引擎 + Agent 编排”的工程思路,不绑定特定框架版本。代码中的实现思路可以迁移到 LangGraph、自研框架或 Java/Go 项目中。
4.2 项目结构
graph-agent-demo/ ├── sop.json # SOP 图定义文件 ├── engine.py # 图执行引擎 ├── nodes.py # 节点实现 ├── main.py # 入口脚本 └── requirements.txt # 依赖说明(本次最小示例不需要第三方库)这个结构本身就是 Graph Engineering 的一个基本实践:把图定义(sop.json)、引擎(engine.py)、业务节点(nodes.py)分离。
4.3 安装依赖
本次最小实现只需要 Python 标准库,不需要额外安装任何第三方包。如果你后面要对接真实的大模型 API,再安装openai或requests即可。
用标准库的好处是:你能把注意力放在“图工程”本身的逻辑上,而不是被框架 API 带偏。
5. 核心流程拆解:把 SOP 变成可执行图
下面拆解整个实现流程。这个流程不是一次性写完的,而是按照“定义图 -> 实现节点 -> 实现引擎 -> 运行验证”的顺序推进。
5.1 定义 SOP 图
我们以一个“电商售后退款自动处理”的简化 SOP 为例。完整流程如下:
- 用户提交退款申请。
- 检查订单是否超过退款期限。
- 如果超时,直接进入“拒绝退款”流程。
- 如果未超时,检查订单是否已发货。
- 如果未发货,自动同意退款并通知仓库取消发货。
- 如果已发货,引导用户填写退货申请。
- 整个流程中,如果用户情绪激烈(从文本判断为投诉),转人工客服。
- 结束节点汇总结果返回给用户。
这个 SOP 图至少包含:
- 1 个开始节点。
- 2 个判断节点(超时判断、发货判断)。
- 1 个模型判断节点(用户情绪判断)。
- 3 个执行节点(拒绝退款、同意退款、引导退货)。
- 1 个结束节点。
用 JSON 表达图结构,让图和代码分离:
// 文件路径:sop.json { "nodes": [ {"id": "start", "type": "start", "next": "check_timeout"}, {"id": "check_timeout", "type": "condition", "condition": "is_timeout", "next": {"yes": "reject_refund", "no": "check_shipped"}}, {"id": "check_shipped", "type": "condition", "condition": "is_shipped", "next": {"yes": "guide_return", "no": "approve_refund"}}, {"id": "approve_refund", "type": "action", "action": "approve_refund"}, {"id": "reject_refund", "type": "action", "action": "reject_refund"}, {"id": "guide_return", "type": "action", "action": "guide_return"}, {"id": "check_sentiment", "type": "llm", "config": {"prompt": "判断用户情绪是否为激烈投诉"}}, {"id": "end", "type": "end"} ], "edges": [ {"from": "approve_refund", "to": "check_sentiment"}, {"from": "reject_refund", "to": "check_sentiment"}, {"from": "guide_return", "to": "check_sentiment"}, {"from": "check_sentiment", "to": "end"} ] }这里有个值得注意的设计:三个执行节点执行完之后都汇聚到check_sentiment节点,这个汇聚点让“用户情绪判断”成为所有结果路径的统一出口,而不需要在每个分支里重复判断。这是图结构比线性代码简洁的典型场景。
5.2 实现状态管理
有了图定义,还要定义状态对象。状态是图中所有节点共享的数据上下文。一个简单的做法是用 Python 字典,但更规范的方式是定义状态类:
# 文件路径:engine.py from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @dataclass class GraphState: """图执行过程中的共享状态""" input: Dict[str, Any] = field(default_factory=dict) data: Dict[str, Any] = field(default_factory=dict) result: Optional[Dict[str, Any]] = None path: List[str] = field(default_factory=list) error: Optional[str] = None def update(self, key: str, value: Any) -> None: self.data[key] = value def get(self, key: str) -> Optional[Any]: return self.data.get(key) def record(self, node_id: str) -> None: self.path.append(node_id)注意path字段:它记录 Agent 实际经过了哪些节点。这个字段对排查问题至关重要——当 Agent 执行结果不符合预期时,先看 path,就知道模型“走”了哪条路,而不是只看到最终结果。
5.3 实现业务节点
每个节点就是一个普通 Python 函数,输入状态、输出更新后的状态。这样设计的好处是:业务逻辑就是纯函数,可以做单元测试,可以在不同图之间复用。
# 文件路径:nodes.py from engine import GraphState def node_check_timeout(state: GraphState) -> GraphState: """判断订单是否超时""" order_info = state.get("order_info") or {} is_timeout = order_info.get("is_timeout", False) state.update("is_timeout", is_timeout) state.record("check_timeout") print(f"[节点] check_timeout: is_timeout={is_timeout}") return state def node_check_shipped(state: GraphState) -> GraphState: """判断订单是否已发货""" order_info = state.get("order_info") or {} is_shipped = order_info.get("is_shipped", False) state.update("is_shipped", is_shipped) state.record("check_shipped") print(f"[节点] check_shipped: is_shipped={is_shipped}") return state def node_approve_refund(state: GraphState) -> GraphState: """自动同意退款""" order_id = state.input.get("order_id", "unknown") state.update("refund_result", {"status": "approved", "order_id": order_id}) state.record("approve_refund") print(f"[节点] approve_refund: 订单 {order_id} 已同意退款,通知仓库拦截") return state def node_reject_refund(state: GraphState) -> GraphState: """拒绝退款""" order_id = state.input.get("order_id", "unknown") state.update("refund_result", {"status": "rejected", "reason": "超过退款时限"}) state.record("reject_refund") print(f"[节点] reject_refund: 订单 {order_id} 已拒绝退款") return state def node_guide_return(state: GraphState) -> GraphState: """引导用户填写退货申请""" order_id = state.input.get("order_id", "unknown") state.update("refund_result", {"status": "needs_return", "order_id": order_id}) state.record("guide_return") print(f"[节点] guide_return: 订单 {order_id} 已发货,引导用户走退货流程") return state def node_check_sentiment(state: GraphState) -> GraphState: """模拟模型判断用户情绪是否激烈""" user_message = state.input.get("user_message", "") # 在真实项目中,这里应该调用 LLM 接口进行情绪判断 # 这里是简单规则模拟:包含"投诉""愤怒""差评"等关键词则升级人工 escalation_keywords = ["投诉", "愤怒", "差评", "曝光"] is_escalation = any(kw in user_message for kw in escalation_keywords) state.update("is_escalation", is_escalation) state.record("check_sentiment") print(f"[节点] check_sentiment: is_escalation={is_escalation}") return state这里有一个需要说明的点:check_sentiment节点在真实项目中应该是一个 LLM 调用节点,调用大模型判断用户情绪。但为了让大家在没有 API Key 的情况下也能完整跑通流程,我用关键词规则模拟了这个节点的行为。在工程架构上,未来替换成真正的 LLM 调用只是改一个节点的内部实现,图和引擎不需要动。
5.4 实现一个简单的图执行引擎
这是整个例子的核心。引擎做的事情是:
- 从当前节点开始。
- 根据节点类型执行对应逻辑。
- 如果是条件节点,根据条件函数的结果选择下一个节点。
- 如果不是条件节点,取配置好的
next节点或边的目标节点。 - 记录每一步路径。
- 如果遇到结束节点,停止执行。
# 文件路径:engine.py import json from typing import Dict, Any, Callable, Optional class GraphEngine: """一个简单的图执行引擎""" def __init__(self, graph_config: Dict[str, Any], node_handlers: Dict[str, Callable]): self.nodes = {node["id"]: node for node in graph_config["nodes"]} self.node_handlers = node_handlers self.start_node = self._find_start_node() def _find_start_node(self) -> str: for node in self.nodes.values(): if node["type"] == "start": return node["id"] raise ValueError("图中没有找到 start 节点") def _get_next(self, node: Dict, state: GraphState) -> str: """根据节点类型决定下一个节点""" node_type = node["type"] if node_type == "condition": condition_key = node.get("condition") condition_value = state.get(condition_key) next_map = node.get("next", {}) # 支持真/假分支和显式匹配 if isinstance(next_map, dict): return next_map.get("yes" if condition_value else "no") return next_map # 如果配置了 next 就走 next,否则广播到所有边 if "next" in node: return node["next"] # 如果没有配置 next,返回 None return None def run(self, initial_state: GraphState) -> GraphState: state = initial_state current_node_id = self.start_node max_steps = 50 step_count = 0 while current_node_id: if step_count >= max_steps: state.error = "超过最大执行步数,可能存在循环引用" print(f"[引擎] 执行超过 {max_steps} 步,强制终止") break node = self.nodes.get(current_node_id) if not node: state.error = f"未知节点: {current_node_id}" break node_type = node["type"] # 起始节点不执行特殊逻辑,仅记录状态 if node_type == "start": state.record("start") print(f"[引擎] 开始执行,进入节点: {current_node_id}") current_node_id = node.get("next") step_count += 1 continue # 结束节点 if node_type == "end": state.record("end") print(f"[引擎] 执行结束") state.result = state.get("refund_result") or state.data break # 条件节点:只做判断,不执行业务(由条件表达式决定) if node_type == "condition": print(f"[引擎] 进入条件节点: {current_node_id}") handler = self.node_handlers.get(current_node_id) if handler: state = handler(state) next_node_id = self._get_next(node, state) print(f"[引擎] 条件节点 {current_node_id} 下一步 -> {next_node_id}") current_node_id = next_node_id step_count += 1 continue # 其他节点(action、llm):执行处理器 handler = self.node_handlers.get(current_node_id) if handler: state = handler(state) else: print(f"[引擎] 节点 {current_node_id} 没有对应处理器,跳过逻辑") # 默认从 next 配置或边集合中寻找下一个节点 next_node_id = self._get_next(node, state) if not next_node_id: # 如果没有显式 next,从 edges 中寻找 for edge in self.edges: if edge["from"] == current_node_id: next_node_id = edge["to"] break print(f"[引擎] 节点 {current_node_id} 下一步 -> {next_node_id}") current_node_id = next_node_id step_count += 1 return state这里有一个生产环境中很关键的细节:最大步数限制。图结构如果设计不当,可能存在环——比如 A 出错后跳回 B,B 又跳回 A,形成死循环。加一个max_steps限制,可以让系统在异常情况下不陷入无限循环。生产级图框架基本都有类似机制,比如 LangGraph 的recursion_limit。
5.5 条件节点的实现细节
条件节点是 SOP 图里的“岔路口”。它本身不执行业务逻辑,只负责读取状态里的某个字段,然后决定下一步走向哪个分支。
在上面的_get_next方法中,条件节点的逻辑是:
if node_type == "condition": condition_key = node.get("condition") condition_value = state.get(condition_key) next_map = node.get("next", {}) if isinstance(next_map, dict): return next_map.get("yes" if condition_value else "no")也就是说,条件节点依赖某个状态字段的布尔值。在sop.json中对应的配置是:
{"id": "check_timeout", "type": "condition", "condition": "is_timeout", "next": {"yes": "reject_refund", "no": "check_shipped"}}这里的 key 是is_timeout,正好来自node_check_timeout函数中state.update("is_timeout", is_timeout)设置的值。理解这个数据流非常关键:条件节点的判断依据不是凭空来的,而是上游节点写入状态的结果。这就是“图控制”的精髓——每个节点只负责把自己的判断结果写进状态,流程跳转由图的配置决定。
6. 完整实现与运行验证
现在把前面各个部分拼起来,跑一个完整示例。
6.1 主入口文件
# 文件路径:main.py import json from engine import GraphEngine, GraphState from nodes import ( node_check_timeout, node_check_shipped, node_approve_refund, node_reject_refund, node_guide_return, node_check_sentiment, ) def load_graph(path: str = "sop.json"): with open(path, "r", encoding="utf-8") as f: return json.load(f) def build_handlers(): """把节点 ID 映射到对应的处理函数""" return { "check_timeout": node_check_timeout, "check_shipped": node_check_shipped, "approve_refund": node_approve_refund, "reject_refund": node_reject_refund, "guide_return": node_guide_return, "check_sentiment": node_check_sentiment, } def main(): graph_config = load_graph() handlers = build_handlers() # 模拟一个未超时、未发货、情绪正常的退款申请 initial_input = { "order_id": "ORD20250101001", "user_message": "你好,我刚买的东西不想要了,申请退款", "order_info": { "is_timeout": False, "is_shipped": False } } initial_state = GraphState(input=initial_input) engine = GraphEngine(graph_config, handlers) final_state = engine.run(initial_state) print("\n===== 执行结果 =====") print(f"执行路径: {' -> '.join(final_state.path)}") print(f"最终状态: {final_state.data}") print(f"最终结果: {final_state.result}") if __name__ == "__main__": main()6.2 运行结果
在命令行中执行:
python main.py预期输出:
[引擎] 开始执行,进入节点: start [引擎] 进入条件节点: check_timeout [节点] check_timeout: is_timeout=False [引擎] 条件节点 check_timeout 下一步 -> check_shipped [引擎] 进入条件节点: check_shipped [节点] check_shipped: is_shipped=False [引擎] 条件节点 check_shipped 下一步 -> approve_refund [引擎] 节点 approve_refund 下一步 -> check_sentiment [节点] approve_refund: 订单 ORD20250101001 已同意退款,通知仓库拦截 [节点] check_sentiment: is_escalation=False [引擎] 节点 check_sentiment 下一步 -> end [引擎] 执行结束 ===== 执行结果 ===== 执行路径: start -> check_timeout -> check_shipped -> approve_refund -> check_sentiment -> end 最终状态: {'is_timeout': False, 'is_shipped': False, 'refund_result': {'status': 'approved', 'order_id': 'ORD20250101001'}, 'is_escalation': False} 最终结果: {'status': 'approved', 'order_id': 'ORD20250101001'}如何判断运行成功:
- 执行路径符合预期:申请未超时、未发货,走了
approve_refund分支。 - 最终结果的
status为approved。 - 没有出现
[引擎] 超过最大执行步数的错误提示。
6.3 测试不同的分支路径
把main.py中的order_info改为以下值,再运行一次,观察执行路径变化:
"order_info": { "is_timeout": True, "is_shipped": False }预期执行路径:
start -> check_timeout -> reject_refund -> check_sentiment -> end再改为:
"order_info": { "is_timeout": False, "is_shipped": True }预期执行路径:
start -> check_timeout -> check_shipped -> guide_return -> check_sentiment -> end再把user_message改为包含“我要投诉你们”,观察check_sentiment节点的is_escalation变为True。
能跑通多条分支路径,说明图引擎对 SOP 的表达是完整的,而不是只适配了一条流程。
7. 常见问题与排查思路
结合实际开发经验,整理了 Graph Engineering 场景下最常见的问题。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| Agent 执行到某个节点后停滞,没有下一步 | 节点没有配置next,也没有在边集合中定义出边 | 查看sop.json中该节点的配置,检查next和edges | 为该节点补充next或添加对应边 |
| 条件节点总是走同一个分支 | 条件字段名写错,状态里没有对应的 key | 打印状态字典,确认state.data中的 key 名 | 统一状态字段命名,使用常量而非字符串 |
| 执行路径出现死循环 | 图中存在环且没有跳出条件 | 启用max_steps限制,打印state.path定位循环 | 重新设计图,为循环增加终止条件或计数器 |
| 节点执行顺序和预期不一致 | 节点函数写了自己修改current_node_id的逻辑 | 检查节点函数是否意外改动引擎变量 | 节点函数只操作状态,不控制流程 |
| 模型输出不符合预期 | 模型节点输出格式不稳定 | 增加输出格式校验和重试机制 | 在模型节点后增加解析与校验节点 |
| 状态数据覆盖 | 不同节点使用相同的 key 写入不同类型数据 | 打印最终状态,检查被覆盖的字段 | 使用命名空间,比如order.check_result |
| 新增节点后其他流程被影响 | 图配置更新时误改了公共节点 | 对比 git diff 中sop.json的变更 | 图配置变更走 review 流程,使用独立分支 |
| 执行速度慢 | 图中模型节点过多或工具调用串行 | 查看每一个节点的耗时日志 | 将独立节点改为并行执行 |
8. 最佳实践与工程建议
8.1 先画图,再写代码
Graph Engineering 最大的优势是设计与实现分离。建议团队在开发新 Agent 流程时,先和业务方一起把 SOP 画成图,确认每个分支、每个汇合点、每个失败出口都清楚,再开始写代码。不是直接在代码里“图着写”,那样图的作用就浪费了。
8.2 图定义与代码分离
图配置文件(JSON/YAML)应该是业务人员能看懂、能评审的。把图定义和节点实现放在不同目录、不同模块,这样才能让业务变更不需要改核心代码。
推荐目录结构:
src/ ├── graphs/ # 图定义文件(JSON/YAML) ├── nodes/ # 节点实现代码 │ ├── order_nodes.py │ ├── payment_nodes.py │ └── user_nodes.py ├── engine/ # 图执行引擎 └── common/ # 通用工具8.3 路径追踪是排障第一手段
Agent 出了问题,先看执行路径,不要重新看模型输出。一条清晰的执行路径能快速告诉你:
- Agent 走了哪条分支。
- 是哪个条件节点决策出了问题。
- 是哪个节点执行抛了异常。
生产环境建议把state.path和每个节点的输入输出都写入结构化日志,方便后续复盘。
8.4 模型节点加校验
凡是模型输出要作为下一步决策依据的节点,后面都应该加一个“校验节点”。校验规则包括:
- 输出是否为合法 JSON。
- 分类结果是否在预设候选集内。
- 文本长度是否合理。
- 必要的字段是否存在。
校验失败时,可以选择重试、按默认值降级、或转人工。一条简单的原则:模型输出永远是不可信的,直到被校验通过。
8.5 生产环境加超时和重试
每一个工具调用节点、模型调用节点都要有独立的超时设置。超时后的处理策略在图中显式建模,比如:
- 工具调用超时 -> 走“降级策略”分支。
- 模型调用超时 -> 重试一次,仍然失败则转人工。
这些异常路径要在图设计阶段就想好,而不是运行时才处理。
8.6 版本管理
图是流程的“代码”,同样需要版本管理。每次修改sop.json都要走代码 review 流程,并且要记录版本号,方便回溯。生产环境如果切换了新的图配置,最好先灰度跑一小部分流量,确认执行路径稳定后再全量。
8.7 不要用图表达所有逻辑
图适合表达流程控制,但不适合表达细粒度的业务校验、计算公式、字符串处理等确定性逻辑。这些逻辑应该封装在节点内部的工具函数里。图解决的是“下一步做什么”,而不是“这一步怎么做”。不要把图变成一张巨大的流程图,那会失去清晰度——也失去了工程可控的价值。
9. 总结与后续学习方向
这篇文章的核心是用一个最小实现演示了 Graph Engineering 在 Agent 场景中的落地思路。可以看到,真正让 Agent 按照 SOP 执行的,不是模型本身,而是图引擎对流程的硬控制。模型在每个节点做出判断,但整体流程是设计好的,是可追踪的,也是可以改的。
核心要点回顾:
- 图是 Agent 流程的“硬约束”,比提示词更可靠。
- 图定义、节点实现、执行引擎三者分离,职责清晰。
- 条件节点 + 状态共享是分支流程的核心机制。
- 路径追踪是排障的第一手段。
- 模型节点必须有校验和失败的兜底路径。
读完本文,建议下一步做这样几件事:
- 从你自己工作的业务里挑一个真实的 SOP,用本文的 JSON 结构试着画图表达。
- 把条件节点的判断逻辑替换成真实的 LLM 调用,体验“模型决策 + 图控制”的配合。
- 研究一下主流的 Agent 编排框架(比如 LangGraph、自研框架),对比它们的图模型和本文的最小实现有哪些异同。
Graph Engineering 本质上是在回答一个问题:Agent 的能力边界由模型决定,但 Agent 的可靠性边界由工程决定。把 SOP 画成图只是第一步,真正让 Agent 在生产环境稳定工作的,是你对流程中每一个分支、每一个失败出口、每一个状态字段的控制力。建立一个最小可验证的图执行引擎,是走向这条路径值得先迈出的第一步。
