☰
Graph Engineering:把SOP画成图,让Agent流程不再跑飞
2026/10/14 13:15:20 网站建设 项目流程

这次我们来看一个 Agent 工程里的经典痛点:业务侧明明有一份写得很清楚的 SOP,落到 Agent 里却变成了一段 Prompt 里的“你按流程来”,然后 Agent 跑着跑着就自由发挥、重复调工具、该确认的地方不确认。把 Agent 的 SOP 画成 Graph,让流程变成显式的节点和边,Agent 才能真正“自己跑”。这套方法论,业内通常叫 Graph Engineering,也就是图工程化的 Agent 编排。

先给一个直接结论:Graph Engineering 不是某一个新框架的专属名词,而是一套把业务流程显式建模为“节点 + 边 + 状态”的工程做法。它的核心价值有三个:流程可观测、步骤可控、节点可复用。跟纯 Prompt 驱动相比,Graph 方式把“下一步做什么”从模型的自由发挥里抽出来,变成拓扑结构里确定的边,模型只负责节点内的具体生成或决策,不再负责整个路径的记忆。

这篇文章会拆解四块内容:SOP 转 Graph 的核心概念、从业务文档到可运行图的完整设计方法、一份最小可运行的代码示例、以及接口 API 和批量任务怎么接。适合正在做 Agent 编排、企业内部自动化流程、或者准备把工作流平台升级成 Agent 平台的读者,也适合想避开“Agent 跑飞”这类问题的后端开发。

1. Graph Engineering 核心能力速览

能力项说明
核心思路把 Agent 的 SOP 拆成节点,用边定义流转,用状态贯穿整个执行过程
主要解决的问题长任务迷路、步骤不可控、过程不可观测、难以人工介入
关键要素节点(Node)、边(Edge)、条件路由(Conditional Edge)、共享状态(State)、人机协同(Human-in-the-Loop)
运行方式可以是代码定义的图,也可以是可视化画布导出的工作流
接口能力通常可以把整个图暴露为 HTTP API,输入初始状态,返回最终结果
批量任务适合,同一张图对多个输入反复执行,只要状态结构设计一致
显存/硬件要求取决于图中节点调用的模型;纯编排层 CPU 即可,涉及大模型推理才需要 GPU
适合场景客服工单处理、内容审核流水线、数据分析流程、文档处理 SOP、多人协作流程的自动化
不适合场景完全开放式聊天、没有固定流程的探索型任务、对延迟极敏感的实时交互(如果图节点过多)

需要强调一点:Graph Engineering 的显存和性能开销不是由“图”本身决定,而是由节点内的模型决定。如果一个节点是 GPT 级别的 LLM 调用,显存和 API 成本都按那个模型算;如果节点只是普通 Python 函数,开销可以忽略。

2. 适用场景与使用边界

Graph Engineering 最适合的场景有两个共同特征:流程相对固定,但节点内部需要智能。典型例子是客服工单处理:先做意图识别,再查知识库,再判断是否需要转人工,最后生成答复。意图识别和答复生成需要模型能力,但“先识别、再查询、再判断”这个顺序是确定的,完全适合画成图。

数据分析流程同理。常见的 SOP 是:接收数据 -> 清洗 -> 统计 -> 出报告,中间可能有“数据量超过阈值就分批处理”的条件分支。这类流程用图来描述,比让模型自己临场发挥要稳得多。

边界也很清楚:

第一,完全没有固定流程的探索型任务不适合。比如“帮我研究一个陌生领域并给出建议”,这种任务连人类都很难提前画出节点,硬套 Graph 反而会限制模型的探索能力。

第二,节点过多会导致编排成本上升。一张图如果超过几十个节点,维护边和状态的工作量会显著增加,这时候更稳妥的做法是先按子流程拆分,再用子图组合。

第三,合规和授权边界必须提前设计。Agent 在图里流转时,可能经过用户隐私数据、人脸信息、声音样本、版权素材或内部业务文档。任何涉及这些内容的节点,都要在进入节点前做权限校验和数据脱敏,并保留完整的执行日志用于审计。换脸、声音克隆、绕过权限校验这类用途,不管画成多漂亮的 Graph,都不能碰,这一点没有商量余地。

3. SOP 转 Graph 的核心概念拆解

3.1 节点(Node)

节点是图里最小的执行单元。一个节点可以是一段 Python 函数、一次 LLM 调用、一个工具调用,也可以是一个子图。判断一个步骤能不能拆成节点的标准很简单:它是否有明确的输入、明确的输出、是否可以被单独测试。

例如工单处理 SOP 里的“意图识别”是一个节点,输入是工单文本,输出是意图标签。这个节点单独拉出来测试,换模型、调 Prompt 都不影响其他节点。这是图编排的第一个好处:节点级替换和节点级测试。

3.2 边(Edge)与条件路由(Conditional Edge)

边决定执行顺序。普通边是“上一个节点执行完,直接进下一个节点”。条件路由则是“根据上一个节点的输出,决定走哪条分支”。

这是 Graph 和线性脚本最大的区别。线性脚本用 if/else 也能做分支,但条件路由是把分支判断本身作为一个可观测的步骤,执行到哪条边、为什么走这条边,日志里一目了然。对 Agent 工程来说,这个“可观测性”比省几行代码重要得多。

3.3 共享状态(State)

状态是贯穿整张图的上下文对象。每个节点从状态里读取自己需要的字段,也把自己的输出写回状态。状态设计是 Graph Engineering 最容易低估的部分。

常见的状态字段包括:输入数据、中间结果、当前步骤编号、重试次数、错误信息、人工审批结果。状态对象必须尽量结构化,避免把大量原始文本塞在里面。状态越乱,节点之间的隐式耦合越强,图就越难维护。

3.4 人机协同(Human-in-the-Loop)

真实业务 SOP 里经常有“需要人工确认”的步骤。Graph 里通常的做法是把它建模成一个特殊节点:执行到该节点时暂停,把当前状态推送到审批队列,等待外部回调写入审批结果,再继续往下走。这比让模型自己判断“要不要问人”要可靠得多。

4. SOP 图的设计方法与落地步骤

把一份真实的 SOP 文档转成 Graph,不是直接把文档段落抄成节点,而是需要经过一轮信息提取和结构设计。

4.1 第一步:从 SOP 文档里抽取“步骤清单”

逐条读 SOP,抽出每个步骤的动作、输入、输出、负责人、异常处理。这一步的输出是一张普通表格,先不要想图结构。

步骤动作输入输出异常处理
S1接收用户申请原始申请单结构化申请数据格式错误则退回
S2校验资质结构化申请数据通过/不通过材料缺失标记补交
S3生成初审意见校验结果初审意见LLM 超时重试一次
S4人工复核初审意见复核结论超时转人工优先队列

4.2 第二步:识别节点、顺序边和条件边

步骤清单里每个“动作”就是一个候选节点。先后关系明确的,画成顺序边;“根据 A 的结果决定走 B 还是走 C”的,画成条件路由。“退回”“补交”“转人工”这类动作,往往是条件路由的出口。

这一步最容易犯的错是把异常处理也当成普通步骤节点。更合理的做法是:正常路径只保留主干节点,异常分支单独成边。这样图的主干清晰,异常分支也不会把图撑乱。

4.3 第三步:定义共享状态结构

给整张图定义一个状态 Schema。建议至少包含:

@dataclass class SOPState: # 输入 raw_input: str parsed_data: dict | None = None # 中间结果 validation_result: str | None = None review_comment: str | None = None # 控制信息 current_step: str = "start" retry_count: int = 0 error_log: list = field(default_factory=list) # 人工审批 human_approval: dict | None = None

状态字段尽量是确定类型和确定语义。状态字段命名混乱,是后期扩展子图时最主要的返工原因。

4.4 第四步:先画主干,再补分支,最后加人机协同

第一次设计不要追求完整。先把主路径跑通,再逐步补充条件路由、异常分支和人工审批节点。Graph 工程和传统软件开发一样,第一版的目标是“最小可运行”,不是“完整覆盖所有异常”。异常覆盖应该在运行日志和线上反馈里迭代出来。

5. 主流实现框架与选型对比

Graph Engineering 的落地方式大致分三类:通用编排框架、可视化工作流平台、自研图执行器。

5.1 通用编排框架

以 LangGraph 为代表的一类框架,把图结构作为一等公民,用代码定义节点、边、状态和条件路由。优点是灵活,节点可以是任意函数或模型调用,状态类型可控,适合深度定制。缺点是需要写代码,对纯业务同学不友好。选它的前提是团队里有能维护图定义的后端开发。

5.2 可视化工作流平台

Dify、Coze 这类平台提供可视化画布,把节点拖拽成图,平台负责执行和日志。优点是上手快,业务同学可以直接参与编排;缺点是部分场景调试受限,节点类型必须用平台提供的能力,遇到平台没覆盖的工具,扩展成本会比较高。

5.3 自研图执行器

如果团队已经有成熟的任务队列系统,或者对执行语义有特殊要求,自研一个极简图执行器也是常见选择。自研的核心压力不在“遍历图”本身,而在持久化、重试、并发、审计和人工审批回调这些工程细节上。下文给出的代码示例属于这一类,用于演示核心机制,不代表生产级完备性。

选型建议:团队以研发为主、流程复杂且需要细粒度控制,优先考虑通用编排框架;团队有业务运营参与编排、流程中等复杂,优先考虑可视化平台;已有基础设施完善、图执行语义特殊,再考虑自研。

6. 最小可运行实现:代码级拆解

下面给一个极简图执行器示例,只演示核心思路:节点注册、顺序边、条件路由、共享状态。这不是某个特定框架的代码,而是通用的教学实现,实际项目请按所选框架重写。

6.1 定义图执行器

# graph_runner.py # 演示用极简图执行器,仅用于理解 Graph Engineering 的核心机制 from dataclasses import dataclass, field from typing import Callable, Any END = "__end__" @dataclass class Node: name: str fn: Callable[[dict], dict] next: str = END @dataclass class Graph: nodes: dict[str, Node] = field(default_factory=dict) def add_node(self, name, fn, next=END): self.nodes[name] = Node(name=name, fn=fn, next=next) def add_conditional_node(self, name, fn, router): # router: Callable[[dict], str],根据状态返回下一个节点名 def wrapped(state: dict) -> dict: result = fn(state) next_node = router(result) self.nodes[name].next = next_node return result self.nodes[name] = Node(name=name, fn=wrapped, next=END) def run(self, start: str, state: dict) -> dict: current = start while current != END: node = self.nodes[current] print(f"[graph] enter node: {node.name}") state = node.fn(state) current = node.next print(f"[graph] exit node: {node.name}, next: {current}") return state

这段代码的关键点在于:节点函数只负责处理状态,下一步去哪由边的定义决定。对于普通边,注册节点时直接指定 next;对于条件路由,用一个 router 函数根据节点结果决定 next。

6.2 定义节点函数

# nodes.py # 每个节点是一个普通函数,输入整个状态字典,返回更新后的状态字典 def receive_request(state: dict) -> dict: # 模拟接收申请,做基础格式校验 raw = state.get("raw_input", "") state["parsed_data"] = {"text": raw, "valid": bool(raw.strip())} return state def validate_application(state: dict) -> dict: # 模拟资质校验 parsed = state.get("parsed_data", {}) state["validation_result"] = "pass" if parsed.get("valid") else "fail" return state def generate_review_comment(state: dict) -> dict: # 模拟 LLM 生成初审意见,实际项目中这里替换为模型调用 state["review_comment"] = "自动初审通过,建议进入复核" return state def human_review(state: dict) -> dict: # 模拟人工复核,实际项目中应暂停并等待外部审批回调 state["human_approval"] = {"approved": True, "operator": "demo"} return state def route_by_validation(state: dict) -> str: if state.get("validation_result") == "pass": return "generate_review_comment" return "human_review"

6.3 组装并运行图

# main.py from graph_runner import Graph from nodes import ( receive_request, validate_application, generate_review_comment, human_review, route_by_validation, ) g = Graph() g.add_node("receive_request", receive_request, next="validate_application") g.add_node("validate_application", validate_application, next="route") g.add_conditional_node("route", lambda s: s, route_by_validation) g.add_node("generate_review_comment", generate_review_comment, next="human_review") g.add_node("human_review", human_review, next=END) initial_state = {"raw_input": "示例工单内容"} final_state = g.run(start="receive_request", state=initial_state) print(final_state)

运行后可以从日志里看到完整的执行路径:receive_request -> validate_application -> route -> generate_review_comment -> human_review。如果校验不通过,路径会变成 receive_request -> validate_application -> route -> human_review。这就是 Graph 工程最直观的收益:路径是可观测、可复现、可测试的。

需要说明的是,真实生产环境不会用这种 while 循环直接跑。更稳妥的做法是把图编译成可持久化的执行计划,配合消息队列和数据库存储状态,保证中途宕机后可以从最近完成的节点恢复。

7. 接口 API 与批量任务接入

Graph 跑通之后,下一步通常是把它封装成 HTTP API,供上游业务系统调用。执行一次图,本质上是提交一份初始状态,等待最终状态返回。

7.1 接口设计建议

{ "workflow": "application_review", "input": { "raw_input": "示例工单内容" }, "callback_url": "https://example.com/callback", "priority": "normal" }

生产级接口建议包含:workflow 标识、input 初始状态、callback_url 异步回调地址、priority 优先级。同步返回适合短流程,异步回调适合长流程和人工审批介入的场景。

7.2 一个通用的 Python 调用示例

import requests url = "http://127.0.0.1:8000/api/v1/workflow/run" payload = { "workflow": "application_review", "input": {"raw_input": "示例工单内容"}, "callback_url": "http://127.0.0.1:9000/callback", } try: resp = requests.post(url, json=payload, timeout=10) resp.raise_for_status() print("任务已提交:", resp.json()) except requests.exceptions.RequestException as e: print("提交失败,请检查服务状态和网络:", e)

注意,这段代码只是通用模板,实际接口路径、字段名、鉴权方式需要按你自己选定的框架调整。不要假设所有平台都用这个路径。

7.3 批量任务的设计要点

批量任务的本质是“同一张图,多个状态实例,并发执行”。设计时要注意三点:

第一,每个任务实例的状态必须隔离,不能在共享状态里写公共字段。第二,要设置并发上限,避免批量任务把后端模型服务打爆。更稳妥的做法是引入任务队列,控制同时执行的图实例数量。第三,每个批量任务都要有任务 ID 和执行日志,失败后可以精确定位到具体节点。

批量任务还需要考虑幂等性。如果某一个节点在调用外部系统时超时,重试可能会造成重复下单或重复通知。建议为每个任务实例生成唯一业务 ID,并在调用外部系统时透传这个 ID,由外部系统做幂等处理。

8. 资源占用与性能观察

Graph 编排层本身的资源占用很低。纯 Python 实现的图执行器,即便节点很多,CPU 和内存开销也可以忽略。真正的资源消耗集中在节点内部,尤其是 LLM 调用节点。

8.1 显存占用如何观察

如果节点内跑本地模型,显存占用由模型参数量、推理精度、输入长度和并发数决定。观察方式建议用nvidia-smi每隔几秒记录一次显存,对比单任务和多任务时的显存曲线。注意本地推理服务的显存通常在模型加载时一次性分配,不会因为单个图节点的文本长短出现剧烈波动,波动更容易出现在并发数量变化时。

8.2 CPU 推理与 GPU 推理的差异

如果图里的模型节点走本地推理,GPU 一般能明显缩短单次推理延迟,但 CPU 推理在低并发、小模型场景下也能接受。更稳妥的判断是:小模型、无并发压力时先试 CPU,跑通逻辑后再评估是否需要 GPU;大模型或高并发必须 GPU。

8.3 影响性能的关键参数

  • 节点数量:节点越多,状态传递和日志开销越大,但通常不是瓶颈。
  • LLM 调用次数:这是最贵的部分。同一张图里,减少不必要的模型调用远比优化代码有效。
  • 并发度:并发过高会把模型服务打满,导致超时和重试风暴。
  • 状态体量:状态里塞大量长文本会拖慢序列化和日志写入。
  • 外部 API 延迟:图节点调用外部服务时,外部服务延迟会直接叠加到总耗时里。

8.4 如何降低资源占用

常见手段包括:对小模型节点做结果缓存;对大文本状态做截断或摘要后再进入下一节点;对 LLM 节点做超时控制;批量任务使用队列削峰。这里不给出具体显存数字,因为不同模型、不同量化精度差异太大,一切以本机实测为准。

9. 常见问题与排查方法

问题现象可能原因排查方式解决方案
图跑到一半停止,没有报错节点函数异常被吞掉,或进入了死循环检查每个节点的 try/except 和日志输出确保异常向外抛出,并记录到 error_log
条件路由走到了错误分支路由函数读到的状态字段不是预期值打印进入路由前一刻的状态快照在路由函数入口加状态校验,明确默认分支
人工审批节点一直不往下走回调地址错误或审批任务没有被消费检查回调接口日志和审批队列确认回调签名,给审批节点加超时提醒
批量任务部分实例失败单个实例状态污染或外部服务限流按任务 ID 查执行日志强制状态隔离,配合队列重试
接口调用超时图中 LLM 节点耗时过长看节点级耗时统计改为异步提交,通过回调接口拿结果
状态字段在节点间丢失某个节点返回了新的字典而不是在原状态上更新打印每个节点输入输出的字段 diff统一状态更新规范,禁止节点随意丢弃字段
启动时依赖安装失败Python 版本或依赖冲突检查错误栈中的包名和版本号使用虚拟环境,按项目文档锁定版本
CUDA/显卡相关报错驱动版本或 PyTorch 版本不匹配运行nvidia-smi和框架自检命令按模型要求重装匹配的 CUDA 和推理框架

最后一个建议:给图执行器加一个“回放”能力。每条执行记录保存状态变更序列,出问题时按时间顺序回放,比看散乱的日志高效得多。这是 Graph 工程相比传统脚本在可观测性上最大的优势,也是排查复杂 Agent 问题时最值得投入的方向。

10. 最佳实践与使用建议

第一,第一次跑通之前,不要追求图结构的完整覆盖。先实现主干路径,肉眼确认执行顺序正确,再逐步补条件分支和异常处理。主干不稳,分支越多越难排查。

第二,保留一套最小可运行配置。包括最小状态 Schema、最小节点集、一份示例输入。以后每次改动图结构,先用这套配置回归,确认没有破坏基础流转。

第三,模型文件、输入素材、输出结果分目录管理。Graph 工程同样适用这套规范,尤其是批量任务,必须保证输出文件按任务 ID 归档。

第四,批量任务必须加日志和失败重试。建议为每个任务实例分配唯一 ID,日志里至少记录:进入节点时间、离开节点时间、节点输出摘要、耗时。重试策略建议指数退避,并设置最大重试次数。

第五,接口服务要限制访问范围。如果图服务暴露到内网或公网,一定要加鉴权,至少使用 API Token;涉及敏感数据的接口建议加 IP 白名单和完整的调用审计。

第六,涉及人脸、声音、版权素材、用户隐私数据时,必须在图入口做数据合规检查。即使是测试环境,也要用脱敏数据或明确授权的素材。

第七,发布前做效果复核。Graph 能保证流程按图走,但不保证每个节点内部模型的输出质量。上线前对每个关键节点的输出做抽样复核,尤其是面向用户的文本和涉及审批建议的结论。

总结与下一步

Graph Engineering 最值得尝试的点,是把 Agent 从“靠模型记忆流程”变成“靠结构定义流程”。流程一旦变成节点和边,就可以测试、可以审计、可以单独替换某一个环节,这是纯 Prompt 驱动方案很难做到的。

如果现在准备上手,最先应该验证的是:把你手头最熟悉的一条 SOP 画成主干图,用最小状态跑通一次,观察执行日志里的节点路径是否符合预期。最容易踩的坑是在状态设计阶段偷懒,字段类型混乱会导致后面每加一个节点都要返工。

后续可以继续扩展的方向有两类:一是把单张图升级为子图组合,让大流程由多个小图嵌套而成;二是把人工审批彻底产品化,做成独立的审批任务中心,让图执行和人工操作彻底解耦。两条路都建立在同一个前提上:先把 SOP 画清楚,再让代码按图画跑。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询