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 SDK | LangGraph |
|---|---|---|
| 上手难度 | 低,几十行能跑 | 中,需要理解图概念 |
| 流程控制 | 隐式,模型自主循环 | 显式,开发者定义节点和边 |
| 分支处理 | 靠模型判断,弱控制 | 条件边,强控制 |
| 人工介入 | 需要自己实现 | 内置 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 不是写完就完事的,它更像一个需要持续调教的员工。你给它清晰的职责、好用的工具、明确的边界,它就能干得不错;你放任不管,它就会在各种边缘情况上翻车。把业务场景转成工作流,技术只是一半,另一半是对业务本身的理解和持续打磨。