☰
Agent SDK与LangGraph实战:业务自动化工作流编排指南
2026/10/2 15:58:28 网站建设 项目流程

1. 业务自动化的真实困境与 Agent SDK 的切入点

1.1 为什么传统脚本越写越像"一次性筷子"

做过业务自动化的朋友大概都有这种体会:一开始只是想写个脚本,把某个系统里的数据拉下来,清洗一下,再推到另一个系统里去。用 Python 写,几十行代码,跑得挺好。可过了两个月,业务方说"能不能加个判断,如果金额超过十万就发个提醒",你加了个 if。再过一个月,又说"这个提醒得看对方是不是重点客户,重点客户走另一条路",你又加了个分支。半年之后,这个脚本变成了八百行,里面嵌套了六七层条件判断,谁都不敢动,一动就出问题。

这就是传统脚本式自动化的通病——它把"业务规则"和"执行逻辑"死死焊在了一起。业务规则一变,代码就得改;代码一改,测试就得重来;测试一重来,上线就得排期。到最后,维护成本高到大家宁愿手动操作,也不愿意碰那个脚本。

我见过太多团队卡在这个阶段。他们不是不想自动化,而是自动化的边际成本太高了。每增加一个场景,就要重新写一遍流程,复用率极低。这时候就需要换一个思路:把业务场景本身抽象成可编排的工作流,让执行引擎去跑,而不是让开发者去写死。

1.2 Agent SDK 到底解决了什么问题

Agent SDK 这类工具的核心价值,说白了就一句话:让"决策"和"执行"分家。传统脚本里,什么时候该做什么,是写死在代码里的;而在 Agent 架构里,什么时候该做什么,是由模型根据当前上下文动态判断的,代码只负责提供"工具"和"约束"。

打个比方。传统脚本像是一份写死的菜谱:第一步切菜,第二步下锅,第三步放盐。而 Agent 更像是一个厨师,你告诉他"我要吃清淡的",他自己决定切什么菜、什么时候下锅、放多少盐。菜谱是死的,厨师是活的。

具体到技术层面,Agent SDK 通常提供这么几样东西:

  • 工具注册机制:你把一个个原子能力(查数据库、调接口、发消息、生成文档)注册成"工具",模型可以按需调用。
  • 循环控制:模型调用工具、拿到结果、再决定下一步,这个循环由 SDK 管理,不用你手写 while。
  • 状态管理:多轮对话、多步骤任务中的上下文,SDK 帮你维护。
  • 结构化输出:让模型按你定义的格式返回结果,方便后续程序处理。

OpenAI Agents SDK 和 LangGraph 是目前两条比较主流的路子。前者更偏向"轻量、开箱即用",适合快速把单点场景跑通;后者更偏向"图编排、强控制",适合复杂流程、多分支、需要人工介入的场景。选哪个,取决于你的业务复杂度,后面我会详细拆。

1.3 这篇文章适合谁看

如果你符合下面任意一条,这篇内容应该对你有用:

  • 手上有一些重复性的业务操作,想用 Python 自动化,但发现越写越乱;
  • 听说过 Agent、LangGraph 这些词,但不知道从哪下手,网上的教程要么太浅要么太学术;
  • 已经在用 LangChain 做了一些东西,想进一步了解怎么把流程编排起来;
  • 团队里要落地一个"智能助手"类的内部工具,需要一套可维护的架构方案。

我会尽量少讲空概念,多讲"这一步为什么这么做""这个参数为什么这么设""踩过什么坑"。代码会给关键片段,但不会贴一大堆让你自己猜。目标是:你看完之后,能照着把自己的一个业务场景跑通。

2. 方案选型:OpenAI Agents SDK 还是 LangGraph

2.1 两条路线的本质区别

很多人一上来就问"哪个更好",这个问题本身就不太对。它们不是替代关系,而是适用场景不同。

OpenAI Agents SDK的设计哲学是"最小可用"。它把 Agent 的核心要素——指令、工具、循环、交接——用很少的抽象封装起来。你定义一个 Agent,给它一组工具,然后 run 一下,它就会自己循环调用工具直到完成任务。代码量少,上手快,适合"一个 Agent 干一件事"的场景。

LangGraph的设计哲学是"显式编排"。它把整个流程画成一张图,节点是执行单元,边是流转条件。你可以精确控制每一步走哪条路、什么时候暂停等人工确认、什么时候回退重试。代码量相对多,但可控性强,适合"多步骤、有分支、要审计"的场景。

我个人的经验是:先用 Agents SDK 把单点跑通,验证价值;当流程开始出现分支和人工介入需求时,再迁移到 LangGraph。不要一上来就上重武器,容易把自己绕进去。

2.2 一张表看清选型依据

维度OpenAI Agents SDKLangGraph
上手难度低,几十行能跑中,需要理解图概念
流程控制隐式,模型自主循环显式,开发者定义节点和边
分支处理靠模型判断,弱控制条件边,强控制
人工介入需要自己实现内置 interrupt 机制
状态持久化基础支持完善,可接数据库
适合场景单点任务、快速验证复杂流程、生产级编排
调试体验简单直接需要可视化工具辅助

选型的时候,我一般会问三个问题:这个流程有没有明确的分支?需不需要人工确认环节?出错了要不要能回退到某一步重来?三个都是"否",用 Agents SDK;有一个是"是",考虑 LangGraph。

2.3 环境准备:别在第一步浪费时间

不管选哪条路,Python 环境是基础。这里说几个实际会踩的坑。

Python 版本建议 3.10 以上,因为很多 Agent 相关的库用到了较新的类型语法。安装的时候,Windows 用户记得勾选"Add Python to PATH",不然后面命令行里敲 python 会提示找不到。macOS 用户如果系统自带的是 2.x 版本,别去动它,用 pyenv 或者直接装 3.11 的独立版本。

虚拟环境一定要用。我见过太多人把所有库装在全局环境里,结果两个项目依赖冲突,排查半天。用 venv 就行:

python -m venv agent-env source agent-env/bin/activate # Windows 用 agent-env\Scripts\activate

装库的时候,OpenAI Agents SDK 的包名是openai-agents,LangGraph 是langgraph。注意 LangGraph 通常还要配langchain-openai来对接模型。别装错了,网上有些教程写的包名是旧的。

提示:如果你在国内网络环境下装包慢,可以配置镜像源,这是常规操作,具体源地址自己搜一下就有,我不在这里展开。

VS Code 里记得选对解释器,右下角点一下,选到刚才创建的虚拟环境。不然你明明装了库,代码里还是报 ModuleNotFoundError,白白浪费半小时。

3. 把业务场景拆成 Agent 能理解的原子能力

3.1 先别写代码,拿张纸把流程画出来

这是我最想强调的一步。很多人一拿到需求就打开编辑器开始写,写到一半发现逻辑不对,推倒重来。正确的做法是:先用自然语言把业务流程完整描述一遍,然后标出哪些是"判断",哪些是"动作"。

举个例子。假设业务场景是"每天从邮件里提取客户询价,查一下库存,能供就回复报价,不能供就转给采购"。拆解下来:

  • 动作:读取邮件
  • 判断:这封邮件是不是询价
  • 动作:提取产品名和数量
  • 动作:查库存
  • 判断:库存是否充足
  • 动作:生成报价并回复
  • 动作:转给采购

这里面,"动作"就是将来要注册成工具的东西,"判断"就是模型要做的决策。你会发现,判断其实不多,大部分是动作。这意味着工具的数量决定了 Agent 的能力边界。

3.2 工具设计的三个原则

工具不是越多越好,也不是越细越好。我总结下来三个原则:

第一,一个工具只做一件事,但要做完整。比如"查库存"这个工具,它应该接收产品名,返回库存数量、仓库位置、预计补货时间。不要设计成"查库存数量"和"查仓库位置"两个工具,那样模型要调两次,还容易漏。

第二,工具的输入输出要结构化。用 Pydantic 定义参数模型,让模型知道每个字段是什么类型、什么含义。这比在描述里写一大段自然语言管用得多。模型看到结构化的 schema,调用准确率会明显提升。

第三,工具描述要写"什么时候用",而不只是"是什么"。比如一个"发送邮件"的工具,描述里要写清楚"当需要向外部人员发送信息时使用,不要用于内部通知"。这样模型在多个相似工具之间选择时,才有依据。

3.3 用 Pydantic 定义工具参数的实际写法

下面是一个查库存工具的示例,用 OpenAI Agents SDK 的风格:

from pydantic import BaseModel, Field from agents import function_tool class StockQuery(BaseModel): product_name: str = Field(description="产品名称,尽量使用标准名称") quantity: int = Field(description="需要的数量", gt=0) @function_tool def check_stock(query: StockQuery) -> dict: """查询指定产品的库存情况。 当客户询问某产品是否有货、能否供货时使用此工具。 返回库存数量、仓库位置和预计补货时间。 """ # 实际业务里这里查数据库或调接口 result = db.query(query.product_name) return { "available": result.stock, "warehouse": result.location, "restock_days": result.restock_days }

注意Field里的description,这不是写给人看的,是写给模型看的。写得越清楚,模型调用越准。gt=0这种约束也要加上,能挡掉一部分无效调用。

LangGraph 里定义工具的方式类似,用@tool装饰器,参数模型一样用 Pydantic。区别在于 LangGraph 更强调工具和节点的绑定关系,后面讲编排的时候会说。

3.4 状态设计:别让上下文无限膨胀

多步骤任务里,状态管理是个容易被忽视的坑。如果你把所有中间结果都塞进对话历史,几轮之后 token 就爆了,而且模型容易被无关信息干扰。

我的做法是:只把"下一步决策需要的信息"放进状态,其余的存在外部。比如查库存的结果,如果下一步只需要知道"够不够",那就存一个布尔值,不要把整个库存记录塞进去。

LangGraph 里用 TypedDict 定义状态,可以精确控制每个节点读写哪些字段:

from typing import TypedDict, Annotated from operator import add class WorkflowState(TypedDict): email_content: str is_inquiry: bool product_name: str quantity: int stock_enough: bool reply_draft: str messages: Annotated[list, add]

Annotated[list, add]这个写法表示 messages 字段是累加的,新消息会追加而不是覆盖。这是 LangGraph 里处理对话历史的常见模式。

4. 完整实操:从零搭一个询价处理工作流

4.1 整体架构与数据流

我们把这个工作流拆成四个阶段:接收与识别、信息提取、决策与执行、结果归档。每个阶段对应图里的一个或几个节点。

数据流是这样的:邮件进来,先过"识别节点"判断是不是询价;是的话进"提取节点"拿到产品名和数量;然后进"决策节点"查库存并判断;根据判断结果走不同的边,要么"报价节点",要么"转采购节点";最后统一进"归档节点"记录日志。

这个结构的好处是,每个节点职责单一,测试的时候可以单独测。哪个环节出问题,一眼就能定位。

4.2 节点实现:识别与提取

识别节点其实就是一个分类任务。用模型判断邮件是不是询价,返回布尔值。这里有个技巧:不要让模型直接返回 True/False,让它返回一个结构化的判断结果,包含理由。这样出错了你能知道它为什么判断错。

def classify_node(state: WorkflowState) -> dict: prompt = f"""判断以下邮件是否为产品询价邮件。 询价邮件的特征:询问产品价格、数量、交期。 非询价邮件:投诉、闲聊、广告、内部通知。 邮件内容: {state['email_content']} 返回 JSON:{{"is_inquiry": true/false, "reason": "判断理由"}} """ result = llm.invoke(prompt) parsed = json.loads(result.content) return {"is_inquiry": parsed["is_inquiry"]}

提取节点类似,但要注意:提取失败是常态。客户可能写"要一批那个红色的",没有明确产品名。这时候不要让流程崩掉,而是返回一个标记,让后续节点决定是转人工还是追问。

def extract_node(state: WorkflowState) -> dict: prompt = f"""从以下邮件中提取产品名称和数量。 如果无法确定,对应字段返回 null。 邮件:{state['email_content']} 返回 JSON:{{"product_name": "...", "quantity": 数字或null}} """ result = llm.invoke(prompt) parsed = json.loads(result.content) return { "product_name": parsed.get("product_name"), "quantity": parsed.get("quantity") }

4.3 条件边:让流程真正"活"起来

LangGraph 最核心的能力就是条件边。它让你能根据状态决定下一步走哪。上面说的"库存够不够"就是一个典型的分支点。

def route_after_check(state: WorkflowState) -> str: if state["product_name"] is None: return "manual_review" if state["stock_enough"]: return "send_quote" return "forward_to_purchase" graph.add_conditional_edges( "check_stock", route_after_check, { "send_quote": "send_quote", "forward_to_purchase": "forward_to_purchase", "manual_review": "manual_review" } )

这个route_after_check函数就是决策逻辑的显式表达。它不依赖模型,是纯代码判断,所以稳定、可测试。能用代码判断的,就不要交给模型,这是我一直坚持的原则。模型适合处理模糊的、需要理解语义的环节,明确的规则判断交给代码。

4.4 人工介入:interrupt 的正确用法

有些环节必须人工确认,比如报价金额。LangGraph 的 interrupt 机制可以让图在某个节点暂停,等人工输入后再继续。

from langgraph.types import interrupt def send_quote(state: WorkflowState) -> dict: draft = generate_quote(state["product_name"], state["quantity"]) # 暂停,等待人工确认 approval = interrupt({"draft": draft, "action": "confirm_quote"}) if approval["approved"]: send_email(draft) return {"reply_draft": draft} else: return {"reply_draft": approval.get("modified", draft)}

这里的关键是:interrupt 的返回值就是人工输入的内容。你可以在前端做一个确认界面,把 draft 展示出来,让人改完再提交。这样既保留了自动化的效率,又守住了关键环节的风险。

4.5 状态持久化:别让流程一崩就全丢

生产环境里,图跑到一半服务重启了,状态不能丢。LangGraph 支持 checkpointer,把状态存到数据库。用 SQLite 做开发测试,上生产换 Postgres。

from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string("checkpoints.db") graph = builder.compile(checkpointer=memory)

调用的时候要传thread_id,同一个 thread 的状态会被关联起来:

config = {"configurable": {"thread_id": "email-001"}} result = graph.invoke(initial_state, config)

这个 thread_id 你可以用邮件 ID 或者业务单号,方便追溯。出问题的时候,拿着 thread_id 就能把整个执行历史调出来看。

5. 常见问题与排查技巧实录

5.1 模型不调用工具,或者调错工具

这是最高频的问题。原因通常有三个:工具描述不清楚、工具太多导致选择困难、提示词里没有引导。

排查顺序:先看工具描述,是不是只写了"是什么"没写"什么时候用";再看工具数量,如果超过十个,考虑分组或者用子 Agent;最后看系统提示词,有没有明确告诉模型"你有这些工具,遇到 X 情况用 Y 工具"。

我自己的经验是,工具描述里加一句反例特别管用。比如"发送邮件工具:用于对外沟通,不要用于内部通知,内部通知请用 send_internal_message"。模型看到这个对比,选择准确率会高很多。

5.2 流程陷入死循环

Agent 循环调用同一个工具,停不下来。这通常是因为工具返回的结果让模型觉得"任务没完成"。解决办法是加一个最大迭代次数,以及让工具返回明确的状态。

result = Runner.run(agent, input, max_turns=10)

max_turns是兜底。更重要的是,工具返回里要带一个明确的"完成"信号。比如查库存返回{"found": true, "stock": 100},模型看到 found 为 true,就知道不用再查了。

5.3 结构化输出解析失败

模型返回的 JSON 格式不对,json.loads报错。这个太常见了。两个办法:一是用 SDK 自带的结构化输出功能,让它强制按 schema 返回;二是加容错,解析失败时重试或者用正则提取。

try: parsed = json.loads(result.content) except json.JSONDecodeError: # 尝试提取 JSON 片段 match = re.search(r'\{.*\}', result.content, re.DOTALL) parsed = json.loads(match.group()) if match else {}

但更好的做法是从源头解决:在提示词里明确"只返回 JSON,不要有其他文字",并且用 SDK 的 response_format 参数约束。

5.4 排查速查表

现象可能原因排查动作
模型不调工具描述不清/工具太多检查 description,加使用场景说明
调错工具工具职责重叠合并或明确区分工具边界
死循环缺完成信号加 max_turns,工具返回明确状态
JSON 解析失败输出格式不稳用结构化输出,加容错重试
状态丢失没配 checkpointer加持久化,传 thread_id
人工介入卡住interrupt 没接前端检查 resume 逻辑和输入传递

5.5 几个我踩过的坑

第一个坑:在工具里做耗时操作。有个工具要调外部接口,响应要十几秒,结果整个流程卡住。后来改成异步,或者把耗时操作拆出去单独跑,流程里只查状态。

第二个坑:状态字段命名随意。一开始用data1、data2这种名字,过了两周自己都忘了是什么。后来统一用业务语义命名,product_name、stock_enough,一看就懂。

第三个坑:忽略 token 消耗。多轮循环加上长上下文,一次任务跑下来 token 用量惊人。后来在状态里只保留必要信息,历史消息做摘要压缩,成本降了一大半。

6. 从能跑到好用:几个提升稳定性的细节

6.1 给模型加"护栏"

模型再聪明也会犯错。关键操作前加校验,比如报价金额超过阈值必须人工确认,发送对象不在白名单里就拦截。这些护栏用代码写,不依赖模型判断。

6.2 日志要记全

每个节点的输入输出、模型的原始返回、工具的调用参数和结果,都要记下来。出问题的时候,这些日志就是你的救命稻草。我一般用结构化日志,方便后续检索和分析。

6.3 灰度上线

别一上来就全量跑。先拿一部分邮件试,人工盯着,看它处理得对不对。跑顺了再逐步放开。这个过程可能要一两周,但比出事之后再回滚划算得多。

6.4 定期回顾失败案例

每周把处理失败的案例捞出来看看,是提取错了、判断错了还是工具挂了。针对性地改提示词、加工具、调流程。这个习惯坚持下来,系统的准确率会稳步上升。

我在实际项目里最大的体会是:Agent 不是写完就完事的,它更像一个需要持续调教的员工。你给它清晰的职责、好用的工具、明确的边界,它就能干得不错;你放任不管,它就会在各种边缘情况上翻车。把业务场景转成工作流,技术只是一半,另一半是对业务本身的理解和持续打磨。

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

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

立即咨询