很多 Agent 项目跑不好不是模型不够强而是流程没有边界。SOP 写在文档里模型执行的时候经常漏步骤、跳判断、把不该并行的事并行掉把 SOP 画成一张有向图之后每一步是什么、下一步由谁决定、异常走哪条边全部落成数据结构。Agent 不再靠“临场发挥”决定顺序而是在图上按规则前进这就是 Graph Engineering 的核心思路。这篇文章不聊概念空转直接拆解一套可以落地的做法把现有 SOP 拆成节点和条件边用代码构建可执行图跑通之后封装成 API 服务再接入批量任务。如果已经在用 LangGraph、自研编排引擎或者准备改造一个多步骤 Agent这篇文章可以直接收藏。1. Graph Engineering 是什么先解决流程失控问题Graph Engineering 用图数据结构来编排 Agent 的执行流程。图里常见的几个要素是Node一个具体工作步骤比如“识别意图”“检索知识库”“调用订单接口”“生成回复”。Edge节点之间的连接关系表达“上一步完成后进入下一步”。State所有节点共享的状态对象保存用户输入、中间结果、最终输出。Conditional Edge条件路由由规则或 LLM 判断后选择走哪条边。Loop循环边用来实现多轮修正、重试、继续追问。Checkpoint检查点把执行到一半的状态落盘异常后可以从断点恢复。之所以说“把 SOP 画成 Graph它就能自己跑”是因为图结构把流程控制权从模型手里收回来了。模型只负责节点内部的计算比如文本分类、内容生成、信息抽取下一步去哪个节点由预先定义好的边和条件路由决定。这样既保留了 LLM 的语义判断能力又不至于让它把关键步骤带偏。核心能力速览能力项说明流程表达用节点和边表达多步骤 SOP支持顺序、分支、循环、并行状态管理节点之间通过全局 State 共享数据避免口头传参条件路由支持规则路由和 LLM 路由模型判断后进入不同分支可观测性每个节点执行前、执行后都能打日志定位问题到具体步骤可恢复性结合 Checkpoint 可在失败后恢复不用整条流程重跑接口化编译后的图对象可以封装成 HTTP API供外部系统调用批量支持同一张图可并发处理多条任务配合限流和队列使用从这里开始后面所有内容都围绕这个目标展开把一个普通的客服处理流程变成一张可运行、可测试、可上线的 Agent 图。2. 适用场景与使用边界Graph Engineering 最适合“步骤清晰但中间有判断”的业务场景。比如客服工单处理先判断用户意图再决定走订单查询分支、知识库分支还是投诉升级分支。内容审核先做基础过滤再由 LLM 判断风险等级高风险进入人工复核。数据分析报告先取数再清洗再让 LLM 生成结论最后交给审查节点。内部工具编排把多个内部系统的操作串成一条固定流程每一步都有权限校验。文档处理先解析文件再分块再检索再生成摘要最后转成 Markdown 输出。这类场景有一个共同点流程主干是确定的只是中途存在若干需要判断的分岔口。Graph 正好把“确定的部分”和“判断的部分”分开管理。Graph Engineering 不适合的场景也要说清楚。如果任务是开放闲聊、开放式探索、完全不确定的自主行为图的节点边界反而会变成限制。比如“帮我在互联网上找资料并自主决定下一步”很难预先画出一张完整的图更适合用自由式 Agent Loop 或 Plan-and-Execute 配合动态规划。还有一类场景是长时记忆高度依赖型任务对话系统依赖长期历史记忆把状态全部塞进图 State 会让结构非常臃肿。使用边界方面需要特别注意图编排本身不解决数据权限问题。节点里可能调用订单接口、读取用户信息、访问内部资料库这些操作必须提前做好角色权限控制。涉及个人信息时要脱敏涉及外部版权内容时只能检索和引用不能随意生成和分发。上线前至少确认三件事谁有权触发这个流程、流程中间能不能插人工审批、输出内容是否需要复核。3. 环境准备与前置条件Graph Engineering 依赖不复杂核心是 Python 环境和一个图编排框架。这里以 LangGraph 生态为例因为它和 LangChain 生态衔接好社区资料也多。前置环境清单项目建议要求操作系统Windows 10/11、macOS、Linux 均可Python3.9 以上建议 3.10 或 3.11包管理工具pip建议配合 virtualenv 或 condaLLM 服务需要可访问的 OpenAI 兼容 API或本地部署模型图编排框架langgraph按实际项目版本安装磁盘空间纯代码编排很小程序依赖通常 2-5GB 以内显存如果只调远程 LLM不强制独显如果用本地模型按模型要求准备先创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install langgraph langchain-core openai fastapi uvicorn说明一下LangGraph 的 API 在不同版本之间会有调整安装之后如果遇到函数名不一致以当前版本的官方文档为准。下面的代码按常见写法演示核心逻辑不变。还需要准备 LLM 服务的访问凭证。建议放到环境变量而不是直接写死在代码里export OPENAI_API_KEY你的密钥 export OPENAI_BASE_URL你的接口地址如果不准备用任何第三方图编排框架也可以自研一个极简图引擎这个方案放在第 5 节单独说明。自研的好处是零依赖缺点是可视化、断点恢复、并发调度都要自己写适合学习不建议直接上生产。4. 把 SOP 拆成 Graph 节点一套可直接套用的方法不要一上来就写代码先把 SOP 画出来。拆解过程建议按下面六步走。第一步列出完整步骤清单。把现有 SOP 从开始到结束的每一步写出来不要省略人觉得“理所当然”的环节。判断、记录、重试都属于步骤。第二步找出判断点。每一步里只要出现“如果”“否则”“根据 XX 决定”这就是一个条件路由。条件路由可以交给规则也可以交给 LLM。第三步找出可并行步骤。如果两个节点互不依赖且执行顺序不影响结果就设计成并行分支。例如“检索知识库”和“查询订单状态”互不依赖可以并行。第四步定义状态字段。把每个节点的输入输出字段列出来。状态字段越收敛越好不要把所有临时变量都塞进去。第五步为每个节点写函数签名。每个节点函数接收 State 类型返回一个 dict更新 State 中需要变化的字段。第六步组装图。先用顺序边连接主干再用条件路由把分支接上最后编译运行。用一个实际的客服工单 SOP 举例。假设原始流程是这样的用户发起咨询 → 系统识别用户意图 → 如果用户要查订单调用订单系统查询如果用户是普通咨询检索知识库 → 根据查询结果生成回复 → 高风险内容进入人工复核否则直接回复。拆成图节点后是这样原流程步骤图节点节点类型主要输出字段识别用户意图classify_input普通节点或 LLM 节点intent检索知识库search_kb普通节点kb_result查询订单系统query_order工具节点order_result生成回复generate_answerLLM 节点final_answer是否进入人工复核check_risk条件路由risk_level这个例子会在下一节直接用代码实现。注意一个细节订单查询和知识库检索不是必然全部执行它们由 classify_input 的条件路由决定走哪条边这才是图和顺序代码的核心区别。5. 实操最小可运行的 Agent 图这一节写一个完整示例演示怎么把上面那张图变成可执行代码。示例使用 LangGraph 的常见写法。5.1 定义工作流状态先定义一个 TypedDict也就是所有节点共享的状态结构from typing import Literal, TypedDict from langgraph.graph import StateGraph, START, END class WorkflowState(TypedDict): user_input: str intent: str kb_result: str order_result: str final_answer: str字段含义很清楚user_input 是用户原始输入intent 是分类结果kb_result 和 order_result 分别是两个分支产生的中间结果final_answer 是最终回复。5.2 写节点函数每个节点函数只做一件事读取需要的字段返回需要更新的字段。这里先用规则代替 LLM保证示例能脱离外部 API 运行def classify_input(state: WorkflowState) - dict: text state[user_input] if 订单 in text or 发货 in text or 物流 in text: return {intent: order} return {intent: consult} def search_kb(state: WorkflowState) - dict: # 实际项目里这里可以调用检索服务或向量数据库 return {kb_result: 知识库检索结果用户咨询的是物流标准的常见问题解答。} def query_order(state: WorkflowState) - dict: # 实际项目里这里可以调用内部订单接口 return {order_result: 订单 OD-2024-001 已发货预计 2 天后到达。} def generate_answer(state: WorkflowState) - dict: if state[intent] order: answer f关于「{state[user_input]}」的查询结果{state[order_result]} else: answer f关于「{state[user_input]}」的回复{state[kb_result]} return {final_answer: answer}节点函数返回值会覆盖 State 中对应字段。这里故意把每个节点保持成纯函数形态后续测试非常省事。5.3 建图、加条件路由、编译接下来构建图结构graph StateGraph(WorkflowState) graph.add_node(classify, classify_input) graph.add_node(kb, search_kb) graph.add_node(order, query_order) graph.add_node(answer, generate_answer) graph.add_edge(START, classify) def route_by_intent(state: WorkflowState) - Literal[consult, order]: return state[intent] graph.add_conditional_edges( classify, route_by_intent, { consult: kb, order: order, }, ) graph.add_edge(kb, answer) graph.add_edge(order, answer) graph.add_edge(answer, END) app graph.compile()条件路由这一段是核心。classify 节点执行完会调用 route_by_intent拿到返回字符串再根据映射字典决定边走向。如果返回的字符串不在映射字典里LangGraph 会报错所以路由函数必须保证返回值受控。现在跑一次result app.invoke({user_input: 我的订单为什么还没发货}) print(result[final_answer])输出符合预期关于「我的订单为什么还没发货」的查询结果订单 OD-2024-001 已发货预计 2 天后到达。再跑一个咨询类输入result app.invoke({user_input: 一般快递多久能送到}) print(result[final_answer])会走知识库分支。到这里一张最小 Agent 图已经跑通了。5.4 把节点换成真实 LLM把 classify_input 换成 LLM 判断核心思路是用 LLM 返回结构化结果然后写回 State。示例使用 OpenAI 兼容接口import json import os from openai import OpenAI client OpenAI( api_keyos.environ[OPENAI_API_KEY], base_urlos.environ.get(OPENAI_BASE_URL, https://api.openai.com/v1), ) def classify_input_with_llm(state: WorkflowState) - dict: response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: ( 你是一个意图识别模块。只从下面两个选项中选择一个不要输出其他内容。 选项consult、order ), }, { role: user, content: state[user_input], }, ], temperature0, ) intent response.choices[0].message.content.strip() return {intent: intent}然后建图时把节点替换成这个函数即可graph StateGraph(WorkflowState) graph.add_node(classify, classify_input_with_llm) # 其余边和上面保持一致这里要注意如果 LLM 返回了预期之外的内容条件路由会直接报错。更稳妥的做法是在路由函数里加归一化处理def route_by_intent(state: WorkflowState) - Literal[consult, order]: if state[intent].startswith(订单) or order in state[intent]: return order return consult在这个阶段把“模型判断”和“流程控制”解耦开后续换模型、换提示词都不用动图结构。5.5 不引入框架的自研方案如果只是想先验证思路也可以不装 LangGraph用普通 Python 函数和 if/else 模拟同样的流程def run_workflow(user_input: str) - dict: state {user_input: user_input} state.update(classify_input(state)) if state[intent] order: state.update(query_order(state)) else: state.update(search_kb(state)) state.update(generate_answer(state)) return state这个方案更容易理解但可扩展性差很多新增节点要改函数体并行和循环都要手写断点恢复也没有现成机制。所以生产环境建议直接用图编排框架自研版本只适合作为教学验证。6. 封装 API 与批量任务图跑通之后第一步是把编译好的 app 对象封装成 HTTP 接口方便前端、定时任务和其他后端系统调用。6.1 用 FastAPI 暴露接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): user_input: str class TaskResponse(BaseModel): final_answer: str app.post(/run, response_modelTaskResponse) def run_task(req: TaskRequest) - TaskResponse: result app.invoke({user_input: req.user_input}) return TaskResponse(final_answerresult[final_answer])启动服务uvicorn server:app --host 127.0.0.1 --port 8000调用测试curl -X POST http://127.0.0.1:8000/run \ -H Content-Type: application/json \ -d {user_input: 我的订单为什么还没发货}返回结果也是 JSON直接把图执行能力暴露成标准 API。需要注意这个示例没有加鉴权生产环境要放在内网或加 Token 校验不能裸奔到公网。6.2 批量任务与并发控制批量任务的核心是把每条输入当成独立的一次 graph.invoke然后控制并发。并发过高会打满上游接口所以建议用限流器from concurrent.futures import ThreadPoolExecutor, as_completed inputs [ {user_input: 订单怎么还没发货}, {user_input: 一般快递要多久}, ] with ThreadPoolExecutor(max_workers4) as pool: futures [pool.submit(app.invoke, item) for item in inputs] for future in as_completed(futures): result future.result() print(result[final_answer])批量任务要做好三件事事项建议失败重试使用带退避的重试机制避免瞬时故障直接丢任务日志记录记录输入 ID、节点名、开始时间、结束时间、输出摘要任务去重批量任务要有唯一任务 ID避免重复提交造成重复处理如果单条任务运行时间很长还要考虑把任务放进队列用 Worker 异步消费而不是在请求线程里同步等待。7. 资源占用与运行观察先明确一个认知图编排引擎本身占用的算力极低。它主要做节点调度、状态合并、边判断不承担模型推理。真正消耗资源的是三个地方LLM 调用次数、外部接口响应时间、State 体积。LLM 调用次数是第一个要盯的指标。一条任务如果经过 5 个节点每个节点都调一次 LLM那就是 5 次调用。批量 1000 条任务就是 5000 次。调整路由策略把不需要 LLM 的节点改成规则判断能直接砍掉大半成本。外部接口响应时间是第二个指标。图流程里如果查询订单接口耗时 3 秒那整条链路就被卡住 3 秒。建议优先做并行分支把两个不相干的接口调用放到同一批执行。State 体积是第三个指标。每次节点返回的 dict 都会被合并进 State。如果节点塞入大段文本、完整日志、全量 JSONState 会越来越臃肿传输和调试都会变慢。合理做法是只保留后续节点需要的字段中间大对象建议落到外部存储只把引用 ID 放进 State。如果要系统化观察建议在每条任务里记录import time start time.time() result app.invoke(input_data) print(ftask complete, cost{time.time() - start:.2f}s)更细的节点级耗时可以给每个节点函数外层包一层带日志的装饰器把节点名、入参摘要、耗时和返回字段名打出来。这样在线上一眼能看到瓶颈在哪个节点。显存占用方面如果 LLM 走远程 API本机基本不需要显存。如果节点里跑本地模型那就要按本地模型的要求准备显存。这部分数字因模型而异实际占用需要以本机测试为准。不要只看总量要看并发数4 条并发任务同时跑显存占用可能接近单任务的 4 倍。8. 常见问题与排查方法图编排看起来简单实际踩坑点不少。下面按常见程度列一份排查清单。问题现象可能原因排查方式解决方案图编译后不执行任何节点START 节点没有连接打印 graph.get_graph() 看拓扑结构检查 START 到首节点的边条件路由报错 KeyError返回字符串不在映射字典里打印路由函数的返回值在路由函数里做归一化限制返回值集合输出缺少某个字段节点函数返回 dict 拼写错误检查节点返回的字段名给 State 所有字段补默认值节点重复执行给同一个节点加了多条进入边查看图结构明确关键节点只有一条入口多余分支被跳过条件边和目标节点没连接检查 add_conditional_edges 的映射补全分支映射并行分支结果互相覆盖多个节点写入同一字段检查各节点返回字段使用不同的状态字段API 调用超时外部接口响应慢检查节点日志耗时拆成异步节点或增加重试批量任务跑到一半崩溃单条任务异常导致整个进程退出逐个任务捕获异常用 future.exception 收集错误单独重试失败任务LLM 返回内容混乱没有约束输出格式打印原始返回值提示词要求 JSON 输出并做解析容错还有一个隐蔽问题反复出现批量并发时状态字段互相污染。虽然每次 invoke 都是独立状态但如果节点函数内部用了普通模块级变量或类属性保存临时数据就可能串数据。节点函数要保持无状态只从 state 读取只往 state 写入。9. 最佳实践与使用建议把 SOP 工程化成 Graph 之后维护成本才是长期问题。下面这几条建议来自工程经验建议直接落到团队规范里。9.1 状态字段要克制State 不是业务数据库只保存节点之间需要传递的数据。字段越少越好宁可多拆几个节点也不要把“顺便可能用到的信息”全塞进去。9.2 路由函数必须收敛所有条件路由的返回值最好集中在有限集合里。不要用自由文本作为跳转标记不要直接拿 LLM 的完整输出做路由值。先解析、再映射、再跳转三步拆开。9.3 每个节点都要有日志和失败策略线上排查时最怕某个节点静默失败。给每个节点配上统一的运行日志输出节点名、输入摘要、耗时和返回结果摘要。失败时区分三种策略重试一次、跳到兜底节点、直接终止人工处理。9.4 把人工审批做成节点而不是事后补合规要求高的流程建议在图中显式加入 human_review 节点。这个节点不调用 LLM而是发送消息给审批人收到人工确认后才进入下一步。看到图的时候谁在哪个环节审批、如果不通过走哪条边一目了然。9.5 首次上线跑小流量不管本地测试多顺利第一次上线都应限制并发、控制流量。先让少量真实任务跑一段时间观察条件路由命中率、LLM 结果质量和外部接口稳定性再逐步放开。批量任务建议从几十条开始跑完检查输出再放量。9.6 素材和数据合规必须前置如果流程涉及用户个人数据、内部业务数据、版权素材交付前必须确认数据权限。生成类 Agent 的输出在面向用户前最好再过一个内容合规检查节点。10. 总结与下一步Graph Engineering 解决的核心问题不是“让模型更聪明”而是“让流程更可控”。Graph 让 SOP 变成数据结构Agent 按图执行每一步都知道自己在哪儿、要去哪儿、失败了怎么处理。这个思路在客服工单、内容审核、报告生成、批量数据处理这类强流程场景里落地速度非常快。最值得先做的事很简单找一个现有 SOP挑出 3 到 5 个关键节点把步骤画成图再按第 5 节的代码模板跑通一次。先不要追求复杂并行先把顺序流程和条件路由跑起来这是所有复杂编排的地基。最大的坑已经强调过几次条件路由的返回值必须受控State 字段必须精简节点函数必须无状态。这三个问题如果还踩说明图还没真正工程化。下一步可以尝试的方向加 Checkpoint 实现故障恢复把并行分支改成真正的异步执行把路由判断从规则升级成 LLM 决策再接一个任务队列管理批量任务。SOP 画成图不是结束画完之后有一种稳稳的掌控感。 Graph Engineering：把SOP变成可执行图，让Agent流程更可控
很多 Agent 项目跑不好不是模型不够强而是流程没有边界。SOP 写在文档里模型执行的时候经常漏步骤、跳判断、把不该并行的事并行掉把 SOP 画成一张有向图之后每一步是什么、下一步由谁决定、异常走哪条边全部落成数据结构。Agent 不再靠“临场发挥”决定顺序而是在图上按规则前进这就是 Graph Engineering 的核心思路。这篇文章不聊概念空转直接拆解一套可以落地的做法把现有 SOP 拆成节点和条件边用代码构建可执行图跑通之后封装成 API 服务再接入批量任务。如果已经在用 LangGraph、自研编排引擎或者准备改造一个多步骤 Agent这篇文章可以直接收藏。1. Graph Engineering 是什么先解决流程失控问题Graph Engineering 用图数据结构来编排 Agent 的执行流程。图里常见的几个要素是Node一个具体工作步骤比如“识别意图”“检索知识库”“调用订单接口”“生成回复”。Edge节点之间的连接关系表达“上一步完成后进入下一步”。State所有节点共享的状态对象保存用户输入、中间结果、最终输出。Conditional Edge条件路由由规则或 LLM 判断后选择走哪条边。Loop循环边用来实现多轮修正、重试、继续追问。Checkpoint检查点把执行到一半的状态落盘异常后可以从断点恢复。之所以说“把 SOP 画成 Graph它就能自己跑”是因为图结构把流程控制权从模型手里收回来了。模型只负责节点内部的计算比如文本分类、内容生成、信息抽取下一步去哪个节点由预先定义好的边和条件路由决定。这样既保留了 LLM 的语义判断能力又不至于让它把关键步骤带偏。核心能力速览能力项说明流程表达用节点和边表达多步骤 SOP支持顺序、分支、循环、并行状态管理节点之间通过全局 State 共享数据避免口头传参条件路由支持规则路由和 LLM 路由模型判断后进入不同分支可观测性每个节点执行前、执行后都能打日志定位问题到具体步骤可恢复性结合 Checkpoint 可在失败后恢复不用整条流程重跑接口化编译后的图对象可以封装成 HTTP API供外部系统调用批量支持同一张图可并发处理多条任务配合限流和队列使用从这里开始后面所有内容都围绕这个目标展开把一个普通的客服处理流程变成一张可运行、可测试、可上线的 Agent 图。2. 适用场景与使用边界Graph Engineering 最适合“步骤清晰但中间有判断”的业务场景。比如客服工单处理先判断用户意图再决定走订单查询分支、知识库分支还是投诉升级分支。内容审核先做基础过滤再由 LLM 判断风险等级高风险进入人工复核。数据分析报告先取数再清洗再让 LLM 生成结论最后交给审查节点。内部工具编排把多个内部系统的操作串成一条固定流程每一步都有权限校验。文档处理先解析文件再分块再检索再生成摘要最后转成 Markdown 输出。这类场景有一个共同点流程主干是确定的只是中途存在若干需要判断的分岔口。Graph 正好把“确定的部分”和“判断的部分”分开管理。Graph Engineering 不适合的场景也要说清楚。如果任务是开放闲聊、开放式探索、完全不确定的自主行为图的节点边界反而会变成限制。比如“帮我在互联网上找资料并自主决定下一步”很难预先画出一张完整的图更适合用自由式 Agent Loop 或 Plan-and-Execute 配合动态规划。还有一类场景是长时记忆高度依赖型任务对话系统依赖长期历史记忆把状态全部塞进图 State 会让结构非常臃肿。使用边界方面需要特别注意图编排本身不解决数据权限问题。节点里可能调用订单接口、读取用户信息、访问内部资料库这些操作必须提前做好角色权限控制。涉及个人信息时要脱敏涉及外部版权内容时只能检索和引用不能随意生成和分发。上线前至少确认三件事谁有权触发这个流程、流程中间能不能插人工审批、输出内容是否需要复核。3. 环境准备与前置条件Graph Engineering 依赖不复杂核心是 Python 环境和一个图编排框架。这里以 LangGraph 生态为例因为它和 LangChain 生态衔接好社区资料也多。前置环境清单项目建议要求操作系统Windows 10/11、macOS、Linux 均可Python3.9 以上建议 3.10 或 3.11包管理工具pip建议配合 virtualenv 或 condaLLM 服务需要可访问的 OpenAI 兼容 API或本地部署模型图编排框架langgraph按实际项目版本安装磁盘空间纯代码编排很小程序依赖通常 2-5GB 以内显存如果只调远程 LLM不强制独显如果用本地模型按模型要求准备先创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install langgraph langchain-core openai fastapi uvicorn说明一下LangGraph 的 API 在不同版本之间会有调整安装之后如果遇到函数名不一致以当前版本的官方文档为准。下面的代码按常见写法演示核心逻辑不变。还需要准备 LLM 服务的访问凭证。建议放到环境变量而不是直接写死在代码里export OPENAI_API_KEY你的密钥 export OPENAI_BASE_URL你的接口地址如果不准备用任何第三方图编排框架也可以自研一个极简图引擎这个方案放在第 5 节单独说明。自研的好处是零依赖缺点是可视化、断点恢复、并发调度都要自己写适合学习不建议直接上生产。4. 把 SOP 拆成 Graph 节点一套可直接套用的方法不要一上来就写代码先把 SOP 画出来。拆解过程建议按下面六步走。第一步列出完整步骤清单。把现有 SOP 从开始到结束的每一步写出来不要省略人觉得“理所当然”的环节。判断、记录、重试都属于步骤。第二步找出判断点。每一步里只要出现“如果”“否则”“根据 XX 决定”这就是一个条件路由。条件路由可以交给规则也可以交给 LLM。第三步找出可并行步骤。如果两个节点互不依赖且执行顺序不影响结果就设计成并行分支。例如“检索知识库”和“查询订单状态”互不依赖可以并行。第四步定义状态字段。把每个节点的输入输出字段列出来。状态字段越收敛越好不要把所有临时变量都塞进去。第五步为每个节点写函数签名。每个节点函数接收 State 类型返回一个 dict更新 State 中需要变化的字段。第六步组装图。先用顺序边连接主干再用条件路由把分支接上最后编译运行。用一个实际的客服工单 SOP 举例。假设原始流程是这样的用户发起咨询 → 系统识别用户意图 → 如果用户要查订单调用订单系统查询如果用户是普通咨询检索知识库 → 根据查询结果生成回复 → 高风险内容进入人工复核否则直接回复。拆成图节点后是这样原流程步骤图节点节点类型主要输出字段识别用户意图classify_input普通节点或 LLM 节点intent检索知识库search_kb普通节点kb_result查询订单系统query_order工具节点order_result生成回复generate_answerLLM 节点final_answer是否进入人工复核check_risk条件路由risk_level这个例子会在下一节直接用代码实现。注意一个细节订单查询和知识库检索不是必然全部执行它们由 classify_input 的条件路由决定走哪条边这才是图和顺序代码的核心区别。5. 实操最小可运行的 Agent 图这一节写一个完整示例演示怎么把上面那张图变成可执行代码。示例使用 LangGraph 的常见写法。5.1 定义工作流状态先定义一个 TypedDict也就是所有节点共享的状态结构from typing import Literal, TypedDict from langgraph.graph import StateGraph, START, END class WorkflowState(TypedDict): user_input: str intent: str kb_result: str order_result: str final_answer: str字段含义很清楚user_input 是用户原始输入intent 是分类结果kb_result 和 order_result 分别是两个分支产生的中间结果final_answer 是最终回复。5.2 写节点函数每个节点函数只做一件事读取需要的字段返回需要更新的字段。这里先用规则代替 LLM保证示例能脱离外部 API 运行def classify_input(state: WorkflowState) - dict: text state[user_input] if 订单 in text or 发货 in text or 物流 in text: return {intent: order} return {intent: consult} def search_kb(state: WorkflowState) - dict: # 实际项目里这里可以调用检索服务或向量数据库 return {kb_result: 知识库检索结果用户咨询的是物流标准的常见问题解答。} def query_order(state: WorkflowState) - dict: # 实际项目里这里可以调用内部订单接口 return {order_result: 订单 OD-2024-001 已发货预计 2 天后到达。} def generate_answer(state: WorkflowState) - dict: if state[intent] order: answer f关于「{state[user_input]}」的查询结果{state[order_result]} else: answer f关于「{state[user_input]}」的回复{state[kb_result]} return {final_answer: answer}节点函数返回值会覆盖 State 中对应字段。这里故意把每个节点保持成纯函数形态后续测试非常省事。5.3 建图、加条件路由、编译接下来构建图结构graph StateGraph(WorkflowState) graph.add_node(classify, classify_input) graph.add_node(kb, search_kb) graph.add_node(order, query_order) graph.add_node(answer, generate_answer) graph.add_edge(START, classify) def route_by_intent(state: WorkflowState) - Literal[consult, order]: return state[intent] graph.add_conditional_edges( classify, route_by_intent, { consult: kb, order: order, }, ) graph.add_edge(kb, answer) graph.add_edge(order, answer) graph.add_edge(answer, END) app graph.compile()条件路由这一段是核心。classify 节点执行完会调用 route_by_intent拿到返回字符串再根据映射字典决定边走向。如果返回的字符串不在映射字典里LangGraph 会报错所以路由函数必须保证返回值受控。现在跑一次result app.invoke({user_input: 我的订单为什么还没发货}) print(result[final_answer])输出符合预期关于「我的订单为什么还没发货」的查询结果订单 OD-2024-001 已发货预计 2 天后到达。再跑一个咨询类输入result app.invoke({user_input: 一般快递多久能送到}) print(result[final_answer])会走知识库分支。到这里一张最小 Agent 图已经跑通了。5.4 把节点换成真实 LLM把 classify_input 换成 LLM 判断核心思路是用 LLM 返回结构化结果然后写回 State。示例使用 OpenAI 兼容接口import json import os from openai import OpenAI client OpenAI( api_keyos.environ[OPENAI_API_KEY], base_urlos.environ.get(OPENAI_BASE_URL, https://api.openai.com/v1), ) def classify_input_with_llm(state: WorkflowState) - dict: response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: ( 你是一个意图识别模块。只从下面两个选项中选择一个不要输出其他内容。 选项consult、order ), }, { role: user, content: state[user_input], }, ], temperature0, ) intent response.choices[0].message.content.strip() return {intent: intent}然后建图时把节点替换成这个函数即可graph StateGraph(WorkflowState) graph.add_node(classify, classify_input_with_llm) # 其余边和上面保持一致这里要注意如果 LLM 返回了预期之外的内容条件路由会直接报错。更稳妥的做法是在路由函数里加归一化处理def route_by_intent(state: WorkflowState) - Literal[consult, order]: if state[intent].startswith(订单) or order in state[intent]: return order return consult在这个阶段把“模型判断”和“流程控制”解耦开后续换模型、换提示词都不用动图结构。5.5 不引入框架的自研方案如果只是想先验证思路也可以不装 LangGraph用普通 Python 函数和 if/else 模拟同样的流程def run_workflow(user_input: str) - dict: state {user_input: user_input} state.update(classify_input(state)) if state[intent] order: state.update(query_order(state)) else: state.update(search_kb(state)) state.update(generate_answer(state)) return state这个方案更容易理解但可扩展性差很多新增节点要改函数体并行和循环都要手写断点恢复也没有现成机制。所以生产环境建议直接用图编排框架自研版本只适合作为教学验证。6. 封装 API 与批量任务图跑通之后第一步是把编译好的 app 对象封装成 HTTP 接口方便前端、定时任务和其他后端系统调用。6.1 用 FastAPI 暴露接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): user_input: str class TaskResponse(BaseModel): final_answer: str app.post(/run, response_modelTaskResponse) def run_task(req: TaskRequest) - TaskResponse: result app.invoke({user_input: req.user_input}) return TaskResponse(final_answerresult[final_answer])启动服务uvicorn server:app --host 127.0.0.1 --port 8000调用测试curl -X POST http://127.0.0.1:8000/run \ -H Content-Type: application/json \ -d {user_input: 我的订单为什么还没发货}返回结果也是 JSON直接把图执行能力暴露成标准 API。需要注意这个示例没有加鉴权生产环境要放在内网或加 Token 校验不能裸奔到公网。6.2 批量任务与并发控制批量任务的核心是把每条输入当成独立的一次 graph.invoke然后控制并发。并发过高会打满上游接口所以建议用限流器from concurrent.futures import ThreadPoolExecutor, as_completed inputs [ {user_input: 订单怎么还没发货}, {user_input: 一般快递要多久}, ] with ThreadPoolExecutor(max_workers4) as pool: futures [pool.submit(app.invoke, item) for item in inputs] for future in as_completed(futures): result future.result() print(result[final_answer])批量任务要做好三件事事项建议失败重试使用带退避的重试机制避免瞬时故障直接丢任务日志记录记录输入 ID、节点名、开始时间、结束时间、输出摘要任务去重批量任务要有唯一任务 ID避免重复提交造成重复处理如果单条任务运行时间很长还要考虑把任务放进队列用 Worker 异步消费而不是在请求线程里同步等待。7. 资源占用与运行观察先明确一个认知图编排引擎本身占用的算力极低。它主要做节点调度、状态合并、边判断不承担模型推理。真正消耗资源的是三个地方LLM 调用次数、外部接口响应时间、State 体积。LLM 调用次数是第一个要盯的指标。一条任务如果经过 5 个节点每个节点都调一次 LLM那就是 5 次调用。批量 1000 条任务就是 5000 次。调整路由策略把不需要 LLM 的节点改成规则判断能直接砍掉大半成本。外部接口响应时间是第二个指标。图流程里如果查询订单接口耗时 3 秒那整条链路就被卡住 3 秒。建议优先做并行分支把两个不相干的接口调用放到同一批执行。State 体积是第三个指标。每次节点返回的 dict 都会被合并进 State。如果节点塞入大段文本、完整日志、全量 JSONState 会越来越臃肿传输和调试都会变慢。合理做法是只保留后续节点需要的字段中间大对象建议落到外部存储只把引用 ID 放进 State。如果要系统化观察建议在每条任务里记录import time start time.time() result app.invoke(input_data) print(ftask complete, cost{time.time() - start:.2f}s)更细的节点级耗时可以给每个节点函数外层包一层带日志的装饰器把节点名、入参摘要、耗时和返回字段名打出来。这样在线上一眼能看到瓶颈在哪个节点。显存占用方面如果 LLM 走远程 API本机基本不需要显存。如果节点里跑本地模型那就要按本地模型的要求准备显存。这部分数字因模型而异实际占用需要以本机测试为准。不要只看总量要看并发数4 条并发任务同时跑显存占用可能接近单任务的 4 倍。8. 常见问题与排查方法图编排看起来简单实际踩坑点不少。下面按常见程度列一份排查清单。问题现象可能原因排查方式解决方案图编译后不执行任何节点START 节点没有连接打印 graph.get_graph() 看拓扑结构检查 START 到首节点的边条件路由报错 KeyError返回字符串不在映射字典里打印路由函数的返回值在路由函数里做归一化限制返回值集合输出缺少某个字段节点函数返回 dict 拼写错误检查节点返回的字段名给 State 所有字段补默认值节点重复执行给同一个节点加了多条进入边查看图结构明确关键节点只有一条入口多余分支被跳过条件边和目标节点没连接检查 add_conditional_edges 的映射补全分支映射并行分支结果互相覆盖多个节点写入同一字段检查各节点返回字段使用不同的状态字段API 调用超时外部接口响应慢检查节点日志耗时拆成异步节点或增加重试批量任务跑到一半崩溃单条任务异常导致整个进程退出逐个任务捕获异常用 future.exception 收集错误单独重试失败任务LLM 返回内容混乱没有约束输出格式打印原始返回值提示词要求 JSON 输出并做解析容错还有一个隐蔽问题反复出现批量并发时状态字段互相污染。虽然每次 invoke 都是独立状态但如果节点函数内部用了普通模块级变量或类属性保存临时数据就可能串数据。节点函数要保持无状态只从 state 读取只往 state 写入。9. 最佳实践与使用建议把 SOP 工程化成 Graph 之后维护成本才是长期问题。下面这几条建议来自工程经验建议直接落到团队规范里。9.1 状态字段要克制State 不是业务数据库只保存节点之间需要传递的数据。字段越少越好宁可多拆几个节点也不要把“顺便可能用到的信息”全塞进去。9.2 路由函数必须收敛所有条件路由的返回值最好集中在有限集合里。不要用自由文本作为跳转标记不要直接拿 LLM 的完整输出做路由值。先解析、再映射、再跳转三步拆开。9.3 每个节点都要有日志和失败策略线上排查时最怕某个节点静默失败。给每个节点配上统一的运行日志输出节点名、输入摘要、耗时和返回结果摘要。失败时区分三种策略重试一次、跳到兜底节点、直接终止人工处理。9.4 把人工审批做成节点而不是事后补合规要求高的流程建议在图中显式加入 human_review 节点。这个节点不调用 LLM而是发送消息给审批人收到人工确认后才进入下一步。看到图的时候谁在哪个环节审批、如果不通过走哪条边一目了然。9.5 首次上线跑小流量不管本地测试多顺利第一次上线都应限制并发、控制流量。先让少量真实任务跑一段时间观察条件路由命中率、LLM 结果质量和外部接口稳定性再逐步放开。批量任务建议从几十条开始跑完检查输出再放量。9.6 素材和数据合规必须前置如果流程涉及用户个人数据、内部业务数据、版权素材交付前必须确认数据权限。生成类 Agent 的输出在面向用户前最好再过一个内容合规检查节点。10. 总结与下一步Graph Engineering 解决的核心问题不是“让模型更聪明”而是“让流程更可控”。Graph 让 SOP 变成数据结构Agent 按图执行每一步都知道自己在哪儿、要去哪儿、失败了怎么处理。这个思路在客服工单、内容审核、报告生成、批量数据处理这类强流程场景里落地速度非常快。最值得先做的事很简单找一个现有 SOP挑出 3 到 5 个关键节点把步骤画成图再按第 5 节的代码模板跑通一次。先不要追求复杂并行先把顺序流程和条件路由跑起来这是所有复杂编排的地基。最大的坑已经强调过几次条件路由的返回值必须受控State 字段必须精简节点函数必须无状态。这三个问题如果还踩说明图还没真正工程化。下一步可以尝试的方向加 Checkpoint 实现故障恢复把并行分支改成真正的异步执行把路由判断从规则升级成 LLM 决策再接一个任务队列管理批量任务。SOP 画成图不是结束画完之后有一种稳稳的掌控感。