Graph Engineering实战：将SOP建模为图，让Agent自动执行

最近在做一个内部告警机器人时碰到了老问题业务方给的 SOP 写得清清楚楚但交给 Agent 执行时要么顺序写死没法处理分支要么改一条流程要翻代码找半天。后来换了个思路把 SOP 直接建模成 graphAgent 沿着图走一遍流程之前很多麻烦都自然消解了。这篇文章会围绕 Graph Engineering 这个思路拆解“把 Agent 的 SOP 画成图它就能自己跑”的完整做法。会覆盖从概念、图结构设计到 Python 最小运行器的落地其中一些代码示例可以直接套到自己的项目里改着用。1. 为什么 SOP 画成图Agent 就能自己跑1.1 传统 SOP 文档的三个痛点很多团队里的 SOP 不是一个系统而是文档。这带来三个很明显的问题第一文字描述有歧义。比如“异常较多时优先处理”什么叫较多阈值是多少不同的人会有不同理解Agent 更没法执行这种模糊指令。第二流程顺序是隐式的。文档里写“先做 A再做 B如果失败则 C”这种逻辑藏在自然语言里机器没有稳定的解析方式。第三流程改动成本高。每次 SOP 更新要重新让 Agent 理解全文或者改代码里的 if-else容易改漏也没法做版本对比。SOP 画成 graph 之后流程逻辑从“字面理解”变成了“结构数据”。节点的下一步是谁、在什么条件下走哪条边都是显式定义的这就是 Graph Engineering 最直接的收益。1.2 Graph Engineering 是做什么的Graph Engineering 不是某一个特定框架而是一套把“关系、流转、依赖”用图结构显式建模并加以工程化管理的方法论。它和传统工作流引擎有相似之处但也有明显区别传统工作流通常以 DAG 或 BPMN 图为主节点固定运行器负责推进。Graph Engineering更强调图本身作为一等公民节点可以挂状态、动态选路还能和知识图谱、图数据库、GraphRAG 这类技术叠加使用。在 Agent 场景里SOP 图就是 Agent 的“运行地图”。Agent 不需要在一大段提示词里记住全部流程只需要知道当前位置、当前节点的处理函数、以及下一步的候选边就行。1.3 从“线性流程”到“图流程”以前写 Agent 自动化很多人习惯用一个list表示流程steps [读取数据, 清洗数据, 分析数据, 发送报告]这种线性列表在简单场景下够用但一旦出现条件分支、并行处理、超时升级就变得很别扭。换成 graph 之后流程图长这样用文字示意start - read_logs - filter_normal - detect_anomaly | severity threshold? / \ alert end每个节点只关心自己的处理和输出怎么流转由边上的条件决定。这样“画出来的 SOP”不再是人看的文档而是机器可以直接执行的路径。2. 把 SOP 翻译成图结构想把一段文字 SOP 转成能跑起来的图需要先建立一套简单但统一的图模型。2.1 节点类型设计实战中大部分 SOP 节点可以归为四类节点类型作用例子开始/结束节点标记流程起点和终点start_node, end_node处理节点执行具体业务动作读取日志、过滤数据判断节点根据条件选择后续路径错误级别是否达到告警阈值子流程节点调用另一个 SOP 图调用通用的“堆栈解析”子流程在代码里我们不需要把节点类型拆得过于复杂。我最常用的方式是在节点上挂一个handler函数返回的输出会写入上下文判断逻辑由边上的condition完成。2.2 边的类型与条件路由边的核心职责是定义“从一个节点如何走到下一个节点”。通常包含source源节点 ID。target目标节点 ID。condition可选的条件函数输入是当前上下文输出 True/False。如果没有 condition边就是无条件执行如果有多个条件边运行时会按顺序判断走第一条满足条件的边。这里有一个容易踩的坑条件边一定要保证至少有一条默认边兜底否则走到这个节点时可能没有任何边可走整个流程就卡死了。2.3 一个日志分析 SOP 示例为了便于展开讲解下面以一个“日志异常分析并告警”的 SOP 为例。假设团队希望 Agent 每天自动完成如下流程从日志源读取新日志。过滤掉正常请求日志。统计异常并判断异常严重程度。如果严重程度超过阈值生成告警内容并推送。不管是否告警都生成一份简短分析报告。用节点表示start - read_logs - filter_normal_logs - detect_anomaly - classify_severity / \ (high) (low/normal) | | create_alert generate_report \ / - end这样Agent 要做的不再是理解大段文字而是沿着图的边依次处理每个节点之间通过上下文传递数据。3. 用 Python 写一个最小 SOP Graph Runner下面写一个最小可运行版本。这里不是完整框架核心目的是演示图运行器的工作原理。你可以把这段代码当作脚手架再往自己的业务方向扩展。3.1 定义节点和边# sop_graph.py from dataclasses import dataclass, field from typing import Callable, Any, Optional dataclass class SOPNode: node_id: str name: str handler: Callable[[dict], dict] description: str dataclass class SOPEdge: source: str target: str condition: Optional[Callable[[dict], bool]] None节点里的handler接收当前上下文返回处理后需要更新到上下文的数据。边里的condition接收当前上下文决定是否走这条边。3.2 定义 SOPGraphclass SOPGraph: def __init__(self): self.nodes: dict[str, SOPNode] {} self.edges: dict[str, list[SOPEdge]] {} def add_node(self, node: SOPNode): self.nodes[node.node_id] node def add_edge(self, edge: SOPEdge): if edge.source not in self.edges: self.edges[edge.source] [] self.edges[edge.source].append(edge) def get_edge(self, node_id: str, ctx: dict) - SOPEdge | None: 按照边的顺序找出第一条满足条件的边 for edge in self.edges.get(node_id, []): if edge.condition is None or edge.condition(ctx): return edge return None这里get_edge是关键。它负责解决“当前节点下一步去哪”的问题条件判断全部集中在这一处。3.3 实现 Runnerclass SOPRunner: def __init__(self, graph: SOPGraph, start_node_id: str): self.graph graph self.current_id start_node_id self.ctx: dict {} self.visited: list[str] [] def run(self, initial_ctx: dict | None None) - dict: if initial_ctx: self.ctx.update(initial_ctx) while self.current_id: if self.current_id in self.visited: raise RuntimeError(f检测到循环节点: {self.current_id}) node self.graph.nodes.get(self.current_id) if not node: raise RuntimeError(f节点不存在: {self.current_id}) self.visited.append(self.current_id) print(f[执行节点] {node.node_id} - {node.name}) step_result node.handler(self.ctx) if step_result: self.ctx.update(step_result) edge self.graph.get_edge(self.current_id, self.ctx) if edge is None: break self.current_id edge.target print(流程执行结束) return self.ctx运行逻辑非常简单从 start 节点开始。执行当前节点的 handler。更新上下文。找符合条件的边。走到下一个节点直到没有可走的边。3.4 给上面的 SOP 填上业务逻辑为了演示下面是日志分析 SOP 的完整代码# demo.py from sop_graph import SOPNode, SOPEdge, SOPGraph, SOPRunner def handle_start(ctx): print(流程开始准备读取日志) return {} def handle_read_logs(ctx): logs [ {level: INFO, msg: user login success}, {level: ERROR, msg: db connection timeout}, {level: ERROR, msg: redis memory usage high}, {level: INFO, msg: cron job finished}, {level: WARN, msg: api response slow}, ] return {logs: logs} def handle_filter_normal(ctx): logs ctx.get(logs, []) abnormal [log for log in logs if log[level] in (ERROR, WARN)] return {abnormal_logs: abnormal} def handle_detect_anomaly(ctx): abnormal ctx.get(abnormal_logs, []) ctx[anomaly_count] len(abnormal) ctx[high_severity] any(log[level] ERROR for log in abnormal) return {} def handle_create_alert(ctx): alert_msg f检测到 {ctx.get(anomaly_count, 0)} 条异常日志其中包含 ERROR 级别问题 return {alert: alert_msg} def handle_generate_report(ctx): report f本次分析完成异常总数: {ctx.get(anomaly_count, 0)} return {report: report} def condition_high(ctx): return ctx.get(high_severity, False) def condition_low(ctx): return not ctx.get(high_severity, False) def build_sop_graph() - SOPGraph: graph SOPGraph() graph.add_node(SOPNode(start, 开始, handle_start)) graph.add_node(SOPNode(read_logs, 读取日志, handle_read_logs)) graph.add_node(SOPNode(filter_normal_logs, 过滤正常日志, handle_filter_normal)) graph.add_node(SOPNode(detect_anomaly, 异常检测, handle_detect_anomaly)) graph.add_node(SOPNode(create_alert, 生成告警, handle_create_alert)) graph.add_node(SOPNode(generate_report, 生成报告, handle_generate_report)) graph.add_edge(SOPEdge(start, read_logs)) graph.add_edge(SOPEdge(read_logs, filter_normal_logs)) graph.add_edge(SOPEdge(filter_normal_logs, detect_anomaly)) graph.add_edge(SOPEdge(detect_anomaly, create_alert, conditioncondition_high)) graph.add_edge(SOPEdge(detect_anomaly, generate_report, conditioncondition_low)) graph.add_edge(SOPEdge(create_alert, generate_report)) return graph if __name__ __main__: graph build_sop_graph() runner SOPRunner(graph, start_node_idstart) result runner.run() print(最终上下文:, result)这里detect_anomaly节点比较特殊它没有直接返回新值而是修改了上下文中的high_severity。这也是一个常用技巧判断节点只负责更新判断条件真正路由由边上的 condition 来做。运行后会看到类似输出[执行节点] start - 开始 流程开始准备读取日志 [执行节点] read_logs - 读取日志 [执行节点] filter_normal_logs - 过滤正常日志 [执行节点] detect_anomaly - 异常检测 [执行节点] create_alert - 生成告警 [执行节点] generate_report - 生成报告 流程执行结束 最终上下文: {alert: 检测到 3 条异常日志其中包含 ERROR 级别问题, report: 本次分析完成异常总数: 3}如果想把告警逻辑从“高严重才告警”改成“低严重也要记录”只需要调整边条件不需要改 handler 函数。4. 让 Agent 在图上运行4.1 Agent 在图中扮演什么角色很多团队刚开始做 AI Agent 时会陷入一种误区让大模型从头到尾自由发挥。实际上在 SOP 已经明确的场景里更稳的做法是把 Agent 放进图里执行。有两种常见模式节点内接入 Agent某个 node 的 handler 里调用大模型让模型处理需理解的任务比如解析日志中错误类型。图驱动 Agent 整体运行把图结构提供给 Agent让 Agent 根据“当前节点 可用边”决定下一步而不是让它自己发挥。两种模式可以同时存在。业务稳定节点用确定性代码需要语义理解的节点用 LLM 调用。4.2 把大模型封装成节点函数在上面例子的基础上如果想引入大模型分析错误日志只需更换 handlerdef handle_llm_detect(ctx): from openai import OpenAI # 示例实际以自己用的 SDK 为准 client OpenAI() logs ctx.get(abnormal_logs, []) prompt 你是日志分析专家请分析以下日志并输出严重级别\n \n.join(str(log) for log in logs) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], ) analysis resp.choices[0].message.content high_severity 严重 in analysis or high in analysis.lower() return {llm_analysis: analysis, high_severity: high_severity}这样就把原来的规则判断替换成了大模型语义分析。图结构没有变但节点能力升级了。这也是 Graph Engineering 和 Agent 结合时最舒服的点图负责稳定LLM 负责智能。4.3 不想重写轮子时怎么选代码里的 Runner 是演示用的真正做生产环境常见的选择有自己维护一套轻量图运行器适合流程简单、团队希望完全可控。使用确定性工作流框架适合已有长期沉淀的企业级流程。参考 LangGraph 这类 Agent 图框架它在图结构上扩展了状态管理和消息传递做 AI Agent 落地时会省不少事。结合图数据库和知识图谱如果 SOP 图本身也要做版本管理和血缘分析可以使用 Neo4j 等图数据库保存图结构。关键点是不要为了“图”而图。如果流程没有分支、没有并行、没有状态共享那么一个普通列表就够了只有当流程复杂度上来时图结构才值得引入。5. 常见问题与排查方法在实际编码和运行时下面这些问题出现频率较高。问题现象常见原因解决思路流程执行到某个节点后卡住当前节点没有满足条件的边检查 condition 逻辑给节点加一条无条件 default 边运行时报“检测到循环节点”图上有环且没有终止条件限制最大步数或把循环改成显式的“失败重试 N 次”上下文越来越大难以排查多个节点不断往 ctx 里塞数据给上下文分命名空间如ctx[alert]、ctx[report]用完清除临时字段同一份 SOP 执行结果不稳定节点 handler 依赖外部状态保证 handler 是纯函数所有输入输出都通过 ctx 传递图结构改动后线上行为不一致没有做 SOP 版本管理每次修改 SOP 图都生成新版本号运行前明确指定版本比较容易被忽略的是“并行节点”的实现。上面的演示代码是顺序执行的如果 SOP 里有多个互不依赖的节点理论上可以并行。但并行会引入线程安全和上下文合并问题建议初期先用顺序执行等监控和日志完善后再优化为并行。6. 工程落地建议6.1 节点粒度怎么控制节点粒度太粗一个 handler 写几百行图就失去了可维护性节点粒度太细图会变得很碎调试成本上升。我的经验是一个节点只做“一个可以独立命名和独立测试的动作”。比如“读取日志”和“过滤正常日志”是分开的“解析堆栈”和“根据堆栈分类”也应该分开。6.2 SOP 版本管理不要省SOP 不是一成不变的。生产环境建议给每个 SOP 图加版本号并保存历史版本。这样即使新版本有问题也能快速回滚到旧版本。一个简单做法是在图对象里加version字段运行器在启动时输出当前版本。更正式的场景可以把图结构序列化成 JSON 存入配置中心或数据库中。6.3 可观测性图运行器最怕的是“跑到一半挂了但不知道挂在哪个节点”。建议记录以下信息节点开始时间和结束时间。输入上下文大小。输出上下文变化。选中的边的条件结果。这样在排查问题时能直接回答“这个节点产生了什么、走到了哪条边”。6.4 和知识图谱的衔接如果 SOP 里包含大量领域概念比如“告警”关联“服务”、关联“责任人”、关联“历史处理记录”可以考虑把这些关系存到图数据库这样 SOP 图中某个节点触发后可以从知识图谱里动态查询相关信息再传给 Agent。Graph Engineering 和知识图谱不是一回事前者管流程结构后者管实体关系。两者可以互补但建议先不过度设计等 SOP 本身跑稳了再叠加。7. 小结把 Agent 的 SOP 画成 graph本质上是把“流程的理解问题”转换为“流程的数据结构问题”。落地之后流程不再藏在文档和 if-else 里而是变成可执行、可测试、可版本管理的图。手写一个十几行的 Runner 其实不难真正难的是想清楚节点边界、条件路由、上下文管理和异常兜底。建议你拿一个自己团队中最常见、最重复的业务流程试着转成 SOP 图跑一次。跑通之后再考虑引入更完整的 Agent 框架这时候上手和理解成本会比直接啃框架低得多。如果你的流程也开始卡在“文档写得很清楚但 Agent 总跑偏”的问题上不妨从这张图开始。