news 2026/9/1 6:07:22

Graph Engineering:用图控制Agent执行SOP的工程实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
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 大致是:

  1. 用户发起退款申请。
  2. 检查订单是否在可退款时间内。
  3. 如果在,自动同意退款并通知仓库拦截发货。
  4. 如果不在,引导用户走售后单流程。
  5. 如果用户投诉升级,转人工客服。

这样一条 SOP,用文字写出来是一段话,用图表示就是一个有分支的流程网络。图的好处是:每个判断点都清楚,每个出口都有归宿,不会出现模型“自由发挥”的中间态。

2.3 Agent 的编排模式:从 Chain 到 Graph

早期 LangChain 的经典抽象是 Chain——一条线性的链,A 做完传给 B,B 做完传给 C。对于线性流程够用,但真实业务几乎都有分支和循环。

Graph 编排的核心表达能力在于:

  • 分支:根据条件选择不同的下一个节点。
  • 循环:处理结果不合格时重试。
  • 并行:多个相互独立的节点同时执行。
  • 汇聚:多个分支的结果汇总后继续。

用一个对比表格更清楚:

维度ChainGraph
流程结构线性、无分支有向无环或有循环
表达力简单流程够用适合复杂真实业务
可观测性执行路径单一路径多样,需要追踪
可扩展性新需求常需要重构新节点可灵活插入
调试难度定位快需要用路径追踪

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,再安装openairequests即可。

用标准库的好处是:你能把注意力放在“图工程”本身的逻辑上,而不是被框架 API 带偏。

5. 核心流程拆解:把 SOP 变成可执行图

下面拆解整个实现流程。这个流程不是一次性写完的,而是按照“定义图 -> 实现节点 -> 实现引擎 -> 运行验证”的顺序推进。

5.1 定义 SOP 图

我们以一个“电商售后退款自动处理”的简化 SOP 为例。完整流程如下:

  1. 用户提交退款申请。
  2. 检查订单是否超过退款期限。
  3. 如果超时,直接进入“拒绝退款”流程。
  4. 如果未超时,检查订单是否已发货。
  5. 如果未发货,自动同意退款并通知仓库取消发货。
  6. 如果已发货,引导用户填写退货申请。
  7. 整个流程中,如果用户情绪激烈(从文本判断为投诉),转人工客服。
  8. 结束节点汇总结果返回给用户。

这个 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 实现一个简单的图执行引擎

这是整个例子的核心。引擎做的事情是:

  1. 从当前节点开始。
  2. 根据节点类型执行对应逻辑。
  3. 如果是条件节点,根据条件函数的结果选择下一个节点。
  4. 如果不是条件节点,取配置好的next节点或边的目标节点。
  5. 记录每一步路径。
  6. 如果遇到结束节点,停止执行。
# 文件路径: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'}

如何判断运行成功:

  1. 执行路径符合预期:申请未超时、未发货,走了approve_refund分支。
  2. 最终结果的statusapproved
  3. 没有出现[引擎] 超过最大执行步数的错误提示。

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中该节点的配置,检查nextedges为该节点补充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 流程的“硬约束”,比提示词更可靠。
  • 图定义、节点实现、执行引擎三者分离,职责清晰。
  • 条件节点 + 状态共享是分支流程的核心机制。
  • 路径追踪是排障的第一手段。
  • 模型节点必须有校验和失败的兜底路径。

读完本文,建议下一步做这样几件事:

  1. 从你自己工作的业务里挑一个真实的 SOP,用本文的 JSON 结构试着画图表达。
  2. 把条件节点的判断逻辑替换成真实的 LLM 调用,体验“模型决策 + 图控制”的配合。
  3. 研究一下主流的 Agent 编排框架(比如 LangGraph、自研框架),对比它们的图模型和本文的最小实现有哪些异同。

Graph Engineering 本质上是在回答一个问题:Agent 的能力边界由模型决定,但 Agent 的可靠性边界由工程决定。把 SOP 画成图只是第一步,真正让 Agent 在生产环境稳定工作的,是你对流程中每一个分支、每一个失败出口、每一个状态字段的控制力。建立一个最小可验证的图执行引擎,是走向这条路径值得先迈出的第一步。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/1 6:04:42

瑞萨RH850F1L CAN通信驱动开发:从官方示例到实际项目调试指南

简介:瑞萨RH850F1L CAN通信驱动官方示例代码,面向汽车电子、工业自动化领域嵌入式开发者,帮助理解并实现RH850F1L微控制器上的CAN总线通信。资源共9个文件,压缩包仅1MB,包含3个c源文件、2个asm汇编文件、2个h头文件&am…

作者头像 李华
网站建设 2026/9/1 6:04:13

管道漏水检测数据集与源码实战:从声学特征到深度学习模型

简介:管道漏水检测数据集是一份面向计算机视觉目标检测任务的VOCYOLO双格式标注资源,包含2614张图像,针对裂缝、泄漏、无泄漏和水四类目标共标注2690个矩形框,可用于智慧管网、工业巡检等场景下的模型训练与算法验证。资源以源码工…

作者头像 李华
网站建设 2026/9/1 6:03:58

基于PROSAIL查找表的LAI预测Python脚本实现与验证

简介:这是一套面向遥感应用场景的LAI预测Python工具包,主要服务农业监测、生态学和气候变化研究中的科研人员,也适合具备Python基础的开发者直接使用。资源共10个文件,压缩包约97KB,包含5个Python脚本、2个Excel样本数…

作者头像 李华
网站建设 2026/9/1 6:03:39

Grok大模型驱动的机器人定制开发:从ROS2代码生成到API集成实践

这次我们不聊某个开源模型权重,而是聊一套把大模型能力真正落到机器人定制开发里的技术服务方案。Grok 这个名字在大模型圈子里热度一直很高,机器人定制开发则是另一个高频话题:从 ROS2 导航、机械臂运动学,到工业产线上的点位示教…

作者头像 李华
网站建设 2026/9/1 6:03:27

模拟智能体技术解析:从核心原理到实战应用指南

在探索前沿技术时,我们常常会经历一个从“玩梗”到“严肃应用”的过程。最近,“模拟智能体”(Simulated Agents)这个概念在开发者社区和AI圈内热度攀升,它不再仅仅是科幻电影里的酷炫概念或技术极客们的“玩具”&#…

作者头像 李华
网站建设 2026/9/1 6:00:41

Unity开发自动化:用CLI工具整合AI辅助工作流

如果你是一名Unity开发者,最近是否感觉工作流被各种AI工具和协议“包围”了?从Cursor的智能补全,到Claude的对话编程,再到各种宣称能连接一切的MCP(Model Context Protocol)服务器,新技术层出不…

作者头像 李华