企业官网建设流程全解析

把 Agent 的 SOP 画成 graph，它就能自己跑。这句话听起来有点像口号，但背后对应的是一套正在被越来越多的 Agent 工程团队采用的流程建模方式：Graph Engineering。如果你正在做 Agent 开发，又觉得多步骤、多分支、多状态流转的业务流程非常难维护，那这篇文章正好适合你。

本文将围绕“如何把一份标准作业流程（SOP）翻译成一张机器可执行的图，并让 Agent 在这张图上自主运行”展开。我们会先讲清楚 Graph Engineering 是什么，再拆解 SOP 变成 graph 的核心步骤，接着用一个可复跑的订单售后案例完成实战，最后补充图数据库存储、常见坑点与工程建议。

1. 背景与核心概念

1.1 什么是 Graph Engineering

Graph Engineering 可以直译为“图工程”，它并不是一个新发明，而是把图模型（Graph Model）系统化地用在工程问题上的一套方法论。

如果你接触过数据结构，很熟悉图的概念：图由节点（Node）和边（Edge）组成。但 Graph Engineering 强调的是“用图来建模并运行真实业务”，而不是停留在算法层面。意思是，业务流程中的每个步骤都看成一个节点，步骤与步骤之间的跳转关系看成边，整张图就是一条可以被计算机执行的“流程蓝图”。

与画在纸上的流程图不同，Graph Engineering 中的图必须具备可执行性：

• 每个节点都有对应的执行函数或处理逻辑；
• 每条边都有明确的跳转条件；
• 节点之间通过共享状态传递数据；
• 整张图可以像程序一样被运行、暂停、恢复和回放。

举个例子：传统客服工单处理流程写在一份 Word 文档里，人阅读后按步骤操作。如果把它转换成 graph，那么“接收工单”“判断是否退换货”“生成退款单”“通知用户”都会变成节点，节点之间的判断箭头变成条件边。Agent 运行这张图时，就不再需要人工阅读 SOP，而是按图上的节点依次执行。

1.2 SOP 与 graph 之间的关系

SOP 是 Standard Operating Procedure 的缩写，意思是标准作业程序。它描述一项工作应该按什么顺序、在什么条件下、由谁/什么系统来完成。常见的 SOP 有：

• 订单售后处理流程；
• 内容审核与驳回流程；
• 故障自愈与告警排查流程；
• 合同审批与归档流程；
• 多 Agent 协作任务流程。

SOP 的天然问题是：它通常以自然语言、表格或 PDF 形式存在，人读得懂，但机器很难直接照着执行。即使你把它写进提示词里让大模型去做，也会有稳定性问题。多步骤流程一旦出现分支、回退、超时重试，提示词就会变得臃肿，大模型也可能在中间步骤“迷路”。

把 SOP 画成 graph，本质上就是把它变成一种“机器可解释的数据结构”。此时 SOP 不再只是一段文字，而是一组节点、边、状态和条件。Agent 要做的，就是沿着图结构一步步执行，而不是凭空理解整段流程。

1.3 为什么 Agent 需要 Graph 编排

早期的 Agent 更多是“单次问答式”，用户问一句，大模型回答一句。但业务中的 SOP 往往需要多轮、多工具、多决策点，比如：

• 先判断用户问题是否属于售后范围；
• 再决定是退、换还是修；
• 然后调用订单系统、库存系统、支付系统；
• 最后生成处理结果并通知用户。

如果完全交给大模型自由发挥，也就是 Auto Agent 风格，执行路径不可控，容易出现“跑偏”或“幻觉”。如果完全用传统 if-else 写死，业务分支一变就要改代码。

Graph 编排正好提供了一条中间路线：流程的骨架是确定性的，保证 SOP 不走样；节点内部可以放 LLM 决策，保证复杂判断有智能；边的跳转可以是确定分支，也可以根据 LLM 输出动态路由。这样既保留了流程的稳定，又给了 Agent 足够的灵活性。

同时，在多 Agent 设计中，主从模式非常常见：主 Agent 负责拆解任务，子 Agent 完成具体子任务。本质上，可以把它理解为把 subagent 当作另一种 tool 来调用。而这个“主 Agent 调子 Agent”的关系，同样可以建模成一张 graph，只是节点上的执行体不再是一个函数，而是一个完整的 Agent。

2. 把 SOP 翻译成图模型的核心原理

2.1 基本术语

在 Graph Engineering 中，有几个术语需要先统一：

这里最容易混淆的是“流程图”和“可执行图”。流程图只描述逻辑，可执行图除了描述逻辑，还包含节点函数、状态对象、运行环境。换句话说，流程图是给人看的，graph 是给机器跑的。

2.2 从 SOP 到 Graph 的建模步骤

假设你手里有一份线下 SOP，想把它变成可执行的 graph，建议按以下四步拆解。

第一步，把 SOP 拆成最小动作。注意不要拆得太粗，比如“处理订单”就不是最小动作，它更像一个流程。最小动作应该是“生成退款单”“调用物流查询接口”“发送短信通知”这种无法再往下拆的步骤。

第二步，明确每个动作的输入与输出。这一步非常重要。也就是说，每个节点需要哪些字段，执行完之后会给 State 增加或修改哪些字段。输入输出清晰，节点之间的关系才会清晰。

第三步，标出分支条件与回退点。这一步直接决定 graph 的复杂度。分支条件包括数值判断、用户输入、状态判断；回退点则是在某个环节失败时，流程回到哪个节点重试。

第四步，定义全局状态。状态是 graph 的“共享内存”。状态设计得不好，节点之间就会出现数据依赖混乱。建议把状态定义成明确的类或者字典结构，每个字段都有清晰含义。

2.3 与普通代码流程的区别

很多同学会问：这不就是状态机吗？用 if-else 也能实现。确实是状态机，但 if-else 实现的流程有几个明显问题。

第一个问题是不可见。所有分支逻辑都写在代码里，运行到哪个分支完全靠日志推测，业务方想看流程走到哪一步非常困难。

第二个问题是不可配置。业务要调整分支顺序，必须改代码、走发布流程，无法快速响应。

第三个问题是不可复用。每个流程都各自写一套 if-else，处理退款写一套，处理审核写一套，公共逻辑很难抽出来共享。

Graph 方案把“流程定义”和“流程执行”分开。流程定义是数据，可以放在配置中心；流程执行是通用引擎，负责读取图、路由节点、更新状态。这样流程可以可视化、可配置、可回放，这是普通 if-else 很难做到的。

3. 环境准备与版本说明

3.1 技术选型建议

这里需要根据你的项目实际情况调整技术栈。目前 Graph Engineering 的落地方式主要有三类：

第一类是专门的 Agent 编排框架，比如 LangGraph、Spring AI Alibaba Graph。这类框架提供了图定义、状态管理、节点注册、条件路由等能力，适合与 LLM 深度结合。

第二类是通用工作流引擎，比如 Temporal、Camunda，但它们在 LLM 节点支持上不如 Agent 编排框架方便。

第三类是自研轻量状态机。如果你的场景比较简单，或者想深入理解运行原理，可以自己写一个几十行的小引擎，这也是本文实战部分采用的方式。

本文示例以 Python 3.10+ 环境为例，重点演示“流程定义与执行分离”的完整思路。示例代码不依赖第三方框架，可以直接复制运行。如果你打算在生产环境使用 LangGraph，请注意它的 0.x 版本 API 变化较快，需要锁定版本并参照官方文档迁移。

3.2 安装依赖

如果只运行本文的轻量示例，不需要额外安装依赖：

# 可选：使用 LangGraph 时可安装 pip install langgraph # 可选：连接 Neo4j 时使用 pip install neo4j

如果你要使用 Neo4j Community 版本，需要注意一个常见的坑：Community 版本默认不包含 Graph Data Science（GDS）插件，你从 Products 下载的 Neo4j 安装包里面不会自带 graph-data-science.jar。如果需要使用图算法，需要根据 Neo4j 版本单独下载对应版本的 GDS 插件，放到 plugins 目录并重启数据库。

3.3 项目结构

本文示例项目的目录结构如下：

graph-sop-demo/ ├── sop_definition.json # SOP 流程定义 ├── graph_engine.py # 轻量图执行引擎 ├── nodes.py # 节点函数定义 ├── agent_llm.py # 模拟 LLM 决策接口 └── main.py # 运行入口

下面我们按这个结构依次实现。

4. 完整实战：把售后 SOP 画成 graph 并让 Agent 自己跑

4.1 需求定义

我们以电商平台“订单售后处理”为业务背景。原始 SOP 如下：

1. 接收用户售后申请；
2. 判断该订单是否在售后时效内，如果超时则直接驳回；
3. 判断是否需要人工审核，如果商品金额超过 1000 元则转人工；
4. 调用 LLM 判断用户申请类型，可能属于“退款”“换货”或“维修”；
5. 根据类型执行对应处理动作；
6. 通知用户处理结果，流程结束。

这个流程的特点是：有确定性判断，也有人工经验判断；有数据分支，也有 LLM 动态分支。非常适合用来演示 Graph Engineering。

4.2 定义一个简单但完整的 Graph Engine

我们先用 Python 写一个轻量图执行引擎。它支持两种边：

• 普通边：上一个节点执行完，无条件进入下一个节点；
• 条件边：根据节点返回结果或 State 内容，动态选择下一个节点。

# 文件路径：graph-sop-demo/graph_engine.py from typing import Any, Callable, Dict, List, Optional class GraphEngine: """ 极简图执行引擎。 核心设计： - 每个节点是一个函数，签名为 fn(state: dict) -> dict - 节点返回值会合并进全局 state - 条件边根据节点返回值中的 next 字段决定下一步 """ def __init__(self, start_node: str = "start"): self.nodes: Dict[str, Callable[[Dict], Dict]] = {} self.edges: Dict[str, str] = {} self.conditional_edges: Dict[str, Callable[[Dict], str]] = {} self.start_node = start_node self.end_node: Optional[str] = None def add_node(self, name: str, fn: Callable[[Dict], Dict]): """注册一个节点。fn 接收 state，返回需要更新的字段。""" self.nodes[name] = fn def add_edge(self, source: str, target: str): """添加无条件边：source 执行完后直接进入 target。""" self.edges[source] = target def add_conditional_edge(self, source: str, router: Callable[[Dict], str]): """添加条件边：根据 router 返回的字符串决定下一个节点。""" self.conditional_edges[source] = router def set_end(self, node: str): """设置流程结束节点。""" self.end_node = node def run(self, initial_state: Dict) -> Dict: """ 启动流程，直到到达结束节点或某个节点没有后续边。 """ state = dict(initial_state) current = self.start_node # 运行时记录每一步，方便日志和后续审计 trace: List[Dict] = [] max_steps = len(self.nodes) * 5 + 10 step_count = 0 while current is not None: step_count += 1 if step_count > max_steps: raise RuntimeError("图执行步骤超过上限，疑似存在循环") node_fn = self.nodes.get(current) if node_fn is None: raise RuntimeError(f"节点 {current} 未注册") print(f"[STEP {step_count}] 执行节点: {current}") result = node_fn(state) state.update(result) trace.append({ "node": current, "result": result, }) # 检查是否是结束节点 if current == self.end_node: print("[FLOW] 流程结束") break # 优先走条件边 if current in self.conditional_edges: next_node = self.conditional_edges[current](state) print(f"[ROUTE] {current} -> {next_node}") current = next_node continue # 走普通边 if current in self.edges: next_node = self.edges[current] print(f"[EDGE] {current} -> {next_node}") current = next_node continue # 没有边也没有条件边时默认结束 print("[FLOW] 没有后续节点，流程结束") break state["_trace"] = trace state["_step_count"] = step_count return state

这段代码的核心并不复杂：从 start 节点开始，不断执行当前节点的函数，然后根据条件边或普通边找到下一个节点，直到遇到结束节点。每一步都打印了执行日志，相当于把流程图“动态跑起来”的关键能力。

4.3 定义节点函数

接下来我们定义真正表达业务的节点。为了不依赖真实业务系统，这里用字典模拟接口调用。

# 文件路径：graph-sop-demo/nodes.py import random import time def receive_application(state: dict) -> dict: """ 节点：接收售后申请 真实场景会从消息队列或 HTTP 接口拿到工单。 这里我们从 state 读取入参，模拟初始化工单。 """ order_id = state.get("order_id", "A10001") apply_time = state.get("apply_time", "2026-01-10 10:00:00") order_time = state.get("order_time", "2026-01-01 12:00:00") amount = state.get("amount", 800.0) print(f"接收售后申请: 订单 {order_id}, 金额 {amount} 元") # 简单模拟时效判断 if "2026-01-01" <= order_time <= apply_time: in_time = True else: in_time = False return { "order_id": order_id, "apply_time": apply_time, "order_time": order_time, "amount": amount, "in_time": in_time, "current_state": "received", } def check_timeout(state: dict) -> dict: """节点：判断是否在售后时效内""" in_time = state.get("in_time", False) if in_time: print("售后时效校验通过") return {"timeout_result": "pass"} else: print("售后时效校验失败") return {"timeout_result": "fail"} def reject_application(state: dict) -> dict: """节点：驳回申请""" order_id = state.get("order_id") reason = state.get("reject_reason", "超出售后期限") print(f"订单 {order_id} 被驳回, 原因: {reason}") return {"final_result": "rejected", "reject_reason": reason} def check_human_review(state: dict) -> dict: """节点：判断是否需要人工审核""" amount = state.get("amount", 0) if amount > 1000: print(f"金额 {amount} 超过阈值，需要人工审核") return {"need_human": True} else: print(f"金额 {amount} 未超过阈值，无需人工审核") return {"need_human": False} def llm_judge_type(state: dict) -> dict: """ 节点：模拟调用 LLM 判断售后类型。 真实场景会调用大模型接口，让大模型根据用户描述返回结构化结果。 这里用随机结果模拟。 """ user_desc = state.get("user_desc", "收到商品有破损，想退货退款") # 模拟调用 LLM，真实情况应使用结构化工具体系来保证输出格式 llm_result = random.choice(["refund", "exchange", "repair"]) print(f"LLM 判断售后类型: {llm_result}，用户描述：{user_desc}") return {"apply_type": llm_result} def handle_refund(state: dict) -> dict: """节点：执行退款操作""" order_id = state.get("order_id") amount = state.get("amount", 0) print(f"执行退款: 订单 {order_id}, 退款金额 {amount} 元") # 模拟退款请求 time.sleep(0.2) success = random.random() > 0.2 return {"refund_status": "success" if success else "fail"} def handle_exchange(state: dict) -> dict: """节点：执行换货操作""" order_id = state.get("order_id") print(f"自动创建换货单: 订单 {order_id}") return {"exchange_status": "created"} def handle_repair(state: dict) -> dict: """节点：执行维修操作""" order_id = state.get("order_id") print(f"创建维修工单: 订单 {order_id}") return {"repair_status": "created"} def notify_user(state: dict) -> dict: """节点：通知用户处理结果""" order_id = state.get("order_id") result = state.get("final_result", state.get("refund_status", "unknown")) print(f"通知用户: 订单 {order_id} 的处理结果 {result}") return {"notify_status": "sent", "final_result": result} def human_review(state: dict) -> dict: """节点：人工审核。这里模拟人工审核结果。""" order_id = state.get("order_id") print(f"人工审核订单 {order_id}...") # 模拟人工审核通过 time.sleep(0.2) return {"human_review_result": "approved"}

这里有一个值得注意的点：我把“判断售后类型”交给了 LLM 节点，而不是写死 if-else。这正是“把 SOP 画成 graph”与“用代码写死流程”的核心差异。节点内部既可以是纯逻辑，也可以是大模型调用。

4.4 用 LLM 模拟结构化决策接口

为了让示例更接近真实 Agent 项目，我们再封装一个模拟 LLM 接口的模块。

# 文件路径：graph-sop-demo/agent_llm.py """ 模拟 Agent 的 LLM 决策接口。 在真实项目中，这里会调用 GPT、通义千问或其他大模型。 关键点是：要让模型输出结构化结果，例如 JSON，然后再由 graph 的 router 读取。 """ import json import random def llm_route(state: dict) -> str: """ 根据 LLM 判断出的售后类型返回下一个节点。 这里相当于 Conditional Edge 的 router 函数。 """ apply_type = state.get("apply_type", "refund") if apply_type == "refund": return "handle_refund" elif apply_type == "exchange": return "handle_exchange" elif apply_type == "repair": return "handle_repair" else: # 防御性兜底，避免未知类型导致流程卡死 return "handle_refund"

需要说明的是，真实场景里让 LLM 输出类型并不安全，因为 LLM 可能返回任意字符串。更稳妥的做法是使用 Function Calling 或 JSON Schema 约束输出格式，同时增加枚举校验。上面代码中的 else 兜底就是一种防御性写法。

4.5 把 SOP 定义成 JSON

我们希望能做到“流程定义与代码分离”。下面把整条 SOP 定义在一个 JSON 文件中。

{ "name": "order_after_sale_sop", "description": "订单售后处理 SOP", "start": "receive_application", "end": "notify_user", "edges": [ {"source": "receive_application", "target": "check_timeout"}, {"source": "check_timeout", "target": "check_human_review", "condition": "timeout_result == pass"}, {"source": "check_timeout", "target": "reject_application", "condition": "timeout_result == fail"}, {"source": "check_human_review", "target": "llm_judge_type", "condition": "need_human == false"}, {"source": "check_human_review", "target": "human_review", "condition": "need_human == true"}, {"source": "human_review", "target": "llm_judge_type"}, {"source": "llm_judge_type", "target": "llm_router", "condition": "special_router"} ] }

严格来说，这个 JSON 的边条件还需要一个“条件表达式解析器”才能真正零代码运行。完整的通用流程引擎往往会集成表达式引擎，比如 SpEL、Aviator、或者简单的 Python eval。为了保持代码精简，本文示例里不实现 JSON 自动解析，而是直接在建图时显式声明条件边。这样做的目的是让你看到“图由数据和函数共同构成”这一核心思想，而不是陷入表达式解析器本身的实现细节。

4.6 组装完整流程并运行

下面把整个流程串起来，创建可运行的入口文件。

# 文件路径：graph-sop-demo/main.py from graph_engine import GraphEngine from nodes import ( check_human_review, check_timeout, handle_exchange, handle_refund, handle_repair, human_review, llm_judge_type, notify_user, receive_application, reject_application, ) from agent_llm import llm_route def build_graph() -> GraphEngine: graph = GraphEngine(start_node="receive_application") graph.set_end("notify_user") # 注册节点 graph.add_node("receive_application", receive_application) graph.add_node("check_timeout", check_timeout) graph.add_node("reject_application", reject_application) graph.add_node("check_human_review", check_human_review) graph.add_node("llm_judge_type", llm_judge_type) graph.add_node("handle_refund", handle_refund) graph.add_node("handle_exchange", handle_exchange) graph.add_node("handle_repair", handle_repair) graph.add_node("human_review", human_review) graph.add_node("notify_user", notify_user) # 无条件边 graph.add_edge("receive_application", "check_timeout") graph.add_edge("human_review", "llm_judge_type") # 条件边 def timeout_router(state: dict) -> str: if state.get("timeout_result") == "pass": return "check_human_review" return "reject_application" def human_review_router(state: dict) -> str: if state.get("need_human") is True: return "human_review" return "llm_judge_type" # LLM 判断后的路由由专门的 router 函数完成 graph.add_conditional_edge("check_timeout", timeout_router) graph.add_conditional_edge("check_human_review", human_review_router) graph.add_conditional_edge("llm_judge_type", llm_route) return graph if __name__ == "__main__": # 模拟一条售后单 initial_state = { "order_id": "A10001", "order_time": "2026-01-05 12:00:00", "apply_time": "2026-01-10 09:30:00", "amount": 800.0, "user_desc": "收到商品破损，申请退款", } graph = build_graph() final_state = graph.run(initial_state) print("\n========== 最终状态 ==========") for key, value in final_state.items(): if not key.startswith("_"): print(f"{key}: {value}")

运行这个脚本，预期会看到类似下面的执行日志：

[STEP 1] 执行节点: receive_application 接收售后申请: 订单 A10001, 金额 800.0 元 [EDGE] receive_application -> check_timeout [STEP 2] 执行节点: check_timeout 售后时效校验通过 [ROUTE] check_timeout -> check_human_review [STEP 3] 执行节点: check_human_review 金额 800.0 未超过阈值，无需人工审核 [ROUTE] check_human_review -> llm_judge_type [STEP 4] 执行节点: llm_judge_type LLM 判断售后类型: refund，用户描述：收到商品破损，申请退款 [ROUTE] llm_judge_type -> handle_refund [STEP 5] 执行节点: handle_refund 执行退款: 订单 A10001, 退款金额 800.0 元 [EDGE] handle_refund -> notify_user [STEP 6] 执行节点: notify_user 通知用户: 订单 A10001 的处理结果 success [FLOW] 流程结束

如果你多运行几次，会因为 LLM 判断结果随机，看到不同的执行路径，比如进入 handle_exchange 或 handle_repair。这就是“图结构 + LLM 决策”的直观效果：流程骨架稳定，但分支可以由模型动态决定。

4.7 运行结果说明

上面示例最重要的收获有两点。

第一，SOP 一旦变成 graph，就变成了“数据 + 函数”的组合。节点函数是执行单元，图结构是执行路径，路由函数是分支策略。业务调整时，大部分情况下只需要调整节点或路由函数，不需要重写整个程序。

第二，Agent 的“智能”被限制在合适的粒度上。流程控制由图保证，LLM 只在“判断售后类型”这一步发挥作用。如果 LLM 判断出错，最坏情况是走了错误分支，但流程不会乱成不可控状态。这也是生产环境对 Agent 的基本要求：稳定优先，聪明其次。

5. 进阶：把 Graph 存进图数据库

5.1 为什么要存图数据库

当流程规模变大后，Graph 的运行日志、流程定义、历史状态都需要持久化。普通关系型数据库可以存，但图数据库在“查询节点之间关系”和“可视化流程轨迹”上更直观。

你可以把每次运行实例抽象成这样的图数据：

• 节点：某个流程实例执行的 Step；
• 属性：节点名称、执行时间、执行结果；
• 关系：STEP_A 执行后跳转到 STEP_B。

这样，复盘一个 Agent 为什么走错了分支时，直接查图数据库就能看到完整路径，比翻日志高效得多。

5.2 Neo4j 写入示例

下面给出一个用 neo4j 驱动写入运行轨迹的示例。这里不展开完整的集群方案，只演示核心写入逻辑。

# 文件路径：graph-sop-demo/neo4j_writer.py（可选） from neo4j import GraphDatabase class GraphWriter: def __init__(self, uri: str, user: str, password: str): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def save_trace(self, trace_id: str, execution_id: str, trace: list): """ 将一次执行的节点轨迹写入 Neo4j。 trace 是 graph_engine 中记录的执行步骤列表。 """ with self.driver.session() as session: for i, step in enumerate(trace): session.run( """ MERGE (s:Step {execution_id: $execution_id, step_index: $i}) SET s.node_name = $node_name, s.result_json = $result_json, s.trace_id = $trace_id """, execution_id=execution_id, i=i, node_name=step["node"], result_json=str(step["result"]), trace_id=trace_id, ) if i > 0: # 给相邻两步之间建立跳转关系 session.run( """ MATCH (prev:Step {execution_id: $execution_id, step_index: $prev_i}) MATCH (curr:Step {execution_id: $execution_id, step_index: $curr_i}) MERGE (prev)-[:NEXT]->(curr) """, execution_id=execution_id, prev_i=i - 1, curr_i=i, )

使用示例：

writer = GraphWriter("bolt://localhost:7687", "neo4j", "your-password") writer.save_trace( trace_id="trace_001", execution_id="exec_001", trace=final_state.get("_trace", []) ) writer.close()

注意，Neo4j Community 版本不包含 Graph Data Science 插件，如果你只是做流程轨迹存储与查询，Community 版本完全够用。如果你需要用到 PageRank、社区发现等图算法，再单独下载对应版本的 GDS 插件。

5.3 流程定义与运行实例分离

到这里可以总结出一个工程分层思想：流程定义图、运行实例图、节点能力图。

流程定义图是“模板”，描述 SOP 有哪些节点、哪些分支；运行实例图是“这一次实际跑的路径”，包含具体时间、状态、结果；节点能力图描述的是每个节点能调用哪些 API、工具、MCP 服务。三者分开存储，权限和查询也分开管理，工程上会更加清晰。

6. 常见问题与排查思路

在 Graph Engineering 落地过程中，最容易踩到的坑如下。

排查这类问题时，建议遵循“从外到内”的顺序：先看网络与依赖，再看流程定义是否正确，最后看状态数据是否异常。不要一上来就怀疑大模型，大多数问题出在流程编排和状态管理上。

7. 最佳实践与工程建议

7.1 SOP 不要上来就全自动化

Graph Engineering 虽然漂亮，但不代表所有 SOP 都适合一次自动化到端到端。建议先选择高频、稳定、边界清晰的流程试水。那些需要大量人际沟通、临场判断的流程，可以保留人工节点，让 Agent 先完成信息收集和初步决策，再交给人工确认。

7.2 节点粒度要适当

节点太粗，比如一个节点干了十件事，排错时定位困难；节点太细，比如把一次 HTTP 调用拆成三个节点，图会变得碎片化，也不好维护。我的经验是：一个节点只做一件事，且这件事能用一个动词加一个宾语描述清楚，比如“获取订单信息”“生成退款单”“发送通知”。如果一个节点需要超过五个字段作为输入，就应该考虑拆分了。

7.3 状态设计决定流程上限

Graph 的每个节点都是“读状态、写状态”。状态字段的命名、类型、作用域需要提前设计。建议：

• 使用统一的不可变数据结构，节点返回新的状态增量；
• 状态字段命名遵循同一套命名规范，例如 prefix_字段名；
• 敏感数据不要写入节点日志；
• 对状态变化做版本号管理，方便回放。

7.4 LLM 只做必要决策，不做流程控制

这是最容易忽略的一条。很多同学把 LLM 当作“万能胶水”，希望它自己决定下一步调用什么工具。在复杂 SOP 中，这会让流程变得不可预测。更成熟的模式是：图结构负责流程控制，LLM 负责在节点内做判断、生成文本、抽取字段。如果你希望 LLM 输出结构化决策结果，一定要用 JSON Schema 或枚举约束。

7.5 日志与可观测性不能省

Graph 的运行轨迹比普通函数调用更容易出现“状态漂移”。生产环境建议记录以下内容：

• 每次实例执行的完整节点序列；
• 每个节点读入和写出的状态字段；
• 每条条件边的判断依据；
• 每个 LLM 节点的响应内容与 token 开销。

只有把运行轨迹完整记录下来，才能在做 A/B 测试、灰度发布、事故复盘时快速定位问题。

7.6 安全与权限边界

在将 Agent 流程与真实系统打通时，必须考虑最小权限原则。退款节点只能调用退款接口，不能顺便拿到用户手机号列表；人工审核节点只能看到该工单的必要字段；LLM 节点不能无限制访问内部 API。涉及金额、个人信息、生产环境变更的节点，必须设置人工审批环节，并且保留完整的操作审计。

7.7 流程版本管理与灰度

SOP 会持续变化。建议把流程定义保存为带版本号的配置，例如 sop_definition_v1.json、v2.json。新流程先走沙箱环境，确认没问题后再灰度切换。线上流程如果出现异常，要及时支持回滚到上一个版本。

8. 总结与学习路线

通过本文的拆解，我们重点掌握了三件事：一是 Graph Engineering 的本质是把 SOP 变成机器可执行的图；二是如何用“节点 + 边 + 状态”完成一个包含 LLM 判断的售后流程；三是如何让流程定义与运行实例分离，并结合图数据库做轨迹存储。

下一步，如果你对 LangGraph 感兴趣，可以对照官方文档实现一个带记忆和工具调用的 Agent graph；如果你想深入了解企业级落地，可以研究 Spring AI Alibaba Graph、Camunda 或 Temporal；再往后，可以尝试多 Agent 协作编排，例如把 subagent 当作特殊的 tool，在主从模式中通过 graph 控制主 Agent 与子 Agent 的调用关系。还可以结合知识图谱，把业务流程与业务实体、SOP 文档、历史案例统一建模。

最推荐的做法是：从一个小型、高频、有明确边界的 SOP 开始，手动建一张图，跑通后再逐步扩展。画图本身不是目的，让流程稳定、可见、可回放、可演进，才是 Graph Engineering 真正的价值所在。