1. 从业务痛点到自动化工作流为什么选择 Python Agent SDK业务系统里最不缺的就是重复劳动。每天早上打开邮箱导出报表、把 CRM 里的客户信息同步到表格、根据工单内容自动分类再派发给对应负责人——这些活儿单看都不难但架不住量大、频次高、还容易出错。我见过太多团队用“人肉 Excel”硬扛也见过有人写了几百行 if-else 脚本结果业务规则一变就全废。这两年 Agent SDK 这类工具逐渐成熟思路就变了不再把业务逻辑写死成代码而是把“意图”和“工具”交给模型去编排。Python 作为胶水语言天然适合做这件事——生态全、上手快、和各类 API 打交道方便。OpenAI Agents SDK、LangGraph 这些框架把“多步骤推理 工具调用 状态管理”封装得越来越顺手你只需要把业务场景拆成一个个可执行的原子操作剩下的交给 Agent 去串联。这篇文章面向的是有一定 Python 基础、想把日常业务流程自动化的开发者。不管你是做运营支撑、数据分析还是内部工具开发只要手里有重复性任务这套思路都能直接套用。我会从整体设计讲到具体实现把踩过的坑和实测有效的参数都摊开说争取让你看完就能动手搭一个属于自己的自动化工作流。2. 整体设计思路把业务场景拆成 Agent 能理解的积木2.1 核心思路意图识别 工具编排 状态流转传统脚本的写法是“步骤 1 做 A步骤 2 做 B”线性且脆弱。Agent 工作流的写法是“告诉它目标是什么它自己决定先做哪步、用哪个工具”。这中间的差别本质上是从“过程式编程”转向“声明式编排”。具体来说一个业务场景要转化成 Agent 工作流需要拆成三层意图层用户输入一句话比如“把上周的销售数据整理成周报发给张总”Agent 需要理解这里面包含“查数据”“做汇总”“发邮件”三个意图。工具层每个意图对应一个可调用的函数或 API比如query_sales_data(start_date, end_date)、generate_report(data)、send_email(to, subject, body)。状态层多步骤之间需要传递数据比如查出来的销售数据要传给报告生成器报告内容要传给邮件发送器。LangGraph 的 StateGraph 就是干这个的。为什么不用简单的链式调用因为真实业务里经常有分支和循环。比如“如果数据缺失就先去补数据补完再继续”这种逻辑用链式写会非常别扭而用图结构表达就自然得多。2.2 框架选型OpenAI Agents SDK 还是 LangGraph这两个框架我都实际用过说下感受。OpenAI Agents SDK 的优势是轻量、上手快适合“单 Agent 少量工具”的场景。它的Agent和Runner抽象很干净几行代码就能跑起来。但它的短板也明显多 Agent 协作、复杂状态管理、条件分支这些支持得比较薄。LangGraph 则是为“有状态的、多步骤的、可能带循环的”工作流设计的。它的核心概念是StateGraph每个节点是一个函数边定义流转逻辑状态在节点间传递。你可以加条件边、可以加循环、可以加人工审核节点。代价是学习曲线陡一些概念多。我的建议是如果业务逻辑简单、步骤少于 5 步、没有复杂分支用 OpenAI Agents SDK 就够了如果涉及多轮交互、条件跳转、人工介入、或者需要持久化状态直接上 LangGraph别后期再迁移。2.3 工具设计原则原子化、幂等、可观测工具函数是整个工作流的基石设计得好不好直接决定后期维护成本。我总结三条原则原子化一个工具只做一件事。不要写process_and_send()这种把处理和发送揉在一起的函数拆成process()和send()两个。这样 Agent 可以灵活组合也方便单独测试。幂等同一个工具用相同参数调用多次结果应该一致。比如send_email如果重复调用会发多封邮件那就需要加去重逻辑比如用 message_id 判断。Agent 在推理时可能会重试幂等性能避免很多麻烦。可观测每个工具调用都要有日志记录输入、输出、耗时、是否成功。后期排查问题时这些日志就是救命稻草。我一般会在工具函数里加装饰器统一记录。import functools import logging import time logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def observable(func): functools.wraps(func) def wrapper(*args, **kwargs): start time.time() logger.info(f调用工具: {func.__name__}, 参数: {args}, {kwargs}) try: result func(*args, **kwargs) elapsed time.time() - start logger.info(f工具 {func.__name__} 完成, 耗时 {elapsed:.2f}s) return result except Exception as e: logger.error(f工具 {func.__name__} 失败: {e}) raise return wrapper这个装饰器看起来简单但实际用起来能省掉大量排查时间。尤其是当 Agent 连续调用多个工具时你能清楚看到是哪一步卡住了。3. 核心细节解析从零搭建一个可用的 Agent 工作流3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.10 以上因为 LangGraph 和 OpenAI Agents SDK 都用到了较新的类型注解特性。python -m venv agent-env source agent-env/bin/activate # Windows 用 agent-env\Scripts\activate pip install openai-agents langgraph langchain-openai python-dotenv如果你用的是 LangGraph还需要额外装langgraph-checkpoint-sqlite来做状态持久化可选但推荐。环境变量里配好OPENAI_API_KEY这个不用多解释。注意不要把所有依赖装在全局环境里。Agent 项目经常需要试不同版本的框架虚拟环境能帮你隔离冲突。3.2 定义工具函数把业务操作变成 Agent 可调用的接口假设我们的业务场景是“自动处理客户工单”读取工单内容、分类、根据分类查询知识库、生成回复、发送回复。每个步骤对应一个工具函数。from agents import function_tool function_tool def fetch_ticket(ticket_id: str) - dict: 根据工单 ID 获取工单详情 # 实际项目中这里调用内部 API return { id: ticket_id, content: 客户反馈登录后页面白屏已尝试清除缓存无效, customer: 张三, priority: high } function_tool def classify_ticket(content: str) - str: 对工单内容进行分类返回分类标签 # 这里可以用模型做分类也可以用规则 keywords { 登录问题: [登录, 白屏, 无法进入], 支付问题: [支付, 扣款, 退款], 功能咨询: [怎么用, 如何, 在哪里] } for category, words in keywords.items(): if any(w in content for w in words): return category return 其他 function_tool def search_knowledge_base(category: str) - list: 根据分类查询知识库返回相关解决方案 kb { 登录问题: [清除浏览器缓存和 Cookie, 尝试无痕模式, 检查网络代理设置], 支付问题: [确认支付账户余额, 检查订单状态, 联系财务核对], } return kb.get(category, [暂无标准解决方案建议人工介入]) function_tool def send_reply(ticket_id: str, reply: str) - bool: 发送回复给客户 # 实际项目中调用邮件或工单系统 API print(f已回复工单 {ticket_id}: {reply}) return True这几个函数都很简单但组合起来就能完成一个完整的工单处理流程。关键在于function_tool装饰器会把函数签名和 docstring 自动转成模型能理解的工具描述所以 docstring 一定要写清楚这是模型判断“什么时候该调用这个工具”的依据。3.3 用 LangGraph 编排工作流节点、边与状态工具定义好了接下来用 LangGraph 把它们串起来。核心是定义一个 State 类型然后建图。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator class TicketState(TypedDict): ticket_id: str content: str category: str solutions: list reply: str status: str def fetch_node(state: TicketState) - TicketState: ticket fetch_ticket(state[ticket_id]) return {content: ticket[content], status: fetched} def classify_node(state: TicketState) - TicketState: category classify_ticket(state[content]) return {category: category, status: classified} def search_node(state: TicketState) - TicketState: solutions search_knowledge_base(state[category]) return {solutions: solutions, status: searched} def reply_node(state: TicketState) - TicketState: reply f您好关于您反馈的问题建议您尝试{.join(state[solutions])} send_reply(state[ticket_id], reply) return {reply: reply, status: replied} # 建图 graph StateGraph(TicketState) graph.add_node(fetch, fetch_node) graph.add_node(classify, classify_node) graph.add_node(search, search_node) graph.add_node(reply, reply_node) graph.set_entry_point(fetch) graph.add_edge(fetch, classify) graph.add_edge(classify, search) graph.add_edge(search, reply) graph.add_edge(reply, END) app graph.compile()跑起来就是result app.invoke({ticket_id: T-2024-001}) print(result[reply])这个例子是线性的但你可以很轻松地加条件边。比如分类为“其他”时跳过知识库查询直接转人工def should_search(state: TicketState) - str: return search if state[category] ! 其他 else human_review graph.add_conditional_edges(classify, should_search, { search: search, human_review: human_review })这就是 LangGraph 比链式调用强的地方——分支和循环都是图的一等公民。3.4 状态管理与持久化让工作流可恢复真实业务里工作流可能跑一半挂了或者需要人工审核后继续。这时候状态持久化就很重要。LangGraph 支持 checkpoint可以把每一步的状态存到 SQLite 或 Postgres。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.db) app graph.compile(checkpointermemory) config {configurable: {thread_id: ticket-001}} result app.invoke({ticket_id: T-2024-001}, config)这样即使中途程序崩溃下次用同一个thread_id调用就能从上次中断的地方继续。对于需要人工审核的节点这个特性尤其有用——审核人可以在几小时后回来继续状态不会丢。实操心得thread_id建议用业务 ID比如工单号不要用随机 UUID。这样排查问题时能直接对应到具体业务。4. 实操过程与核心环节实现一个完整的自动化工作流案例4.1 场景定义每日销售数据汇总与推送光说理论没意思我们拿一个具体场景走完整流程每天早上 9 点自动从数据库拉取前一天的销售数据按区域汇总生成 Markdown 报告推送到企业微信群里。这个场景涉及四个步骤查数据、汇总、生成报告、推送。每一步都是一个工具用 LangGraph 串起来再加一个定时触发。4.2 工具函数实现与参数计算先写数据查询工具。这里用 SQLite 模拟实际项目换成 MySQL 或 Postgres 都一样。import sqlite3 from datetime import datetime, timedelta function_tool def query_sales(date: str) - list: 查询指定日期的销售记录date 格式 YYYY-MM-DD conn sqlite3.connect(sales.db) cursor conn.cursor() cursor.execute( SELECT region, product, amount FROM sales WHERE date ?, (date,) ) rows cursor.fetchall() conn.close() return [{region: r[0], product: r[1], amount: r[2]} for r in rows] function_tool def aggregate_by_region(records: list) - dict: 按区域汇总销售额 result {} for r in records: result[r[region]] result.get(r[region], 0) r[amount] return result function_tool def generate_markdown_report(summary: dict, date: str) - str: 生成 Markdown 格式的销售报告 lines [f# {date} 销售日报\n] lines.append(| 区域 | 销售额 |) lines.append(|------|--------|) total 0 for region, amount in sorted(summary.items(), keylambda x: -x[1]): lines.append(f| {region} | {amount:.2f} |) total amount lines.append(f\n**总计{total:.2f}**) return \n.join(lines) function_tool def push_to_wecom(content: str) - bool: 推送消息到企业微信群 # 实际调用 webhook print(f推送内容\n{content}) return True参数计算这块日期处理是最容易出错的。datetime.now() - timedelta(days1)拿到的是昨天但如果脚本在凌晨跑可能拿到的是“昨天”但数据还没生成。我的做法是显式传入日期由调度器决定跑哪天的数据工具函数只负责查询。4.3 工作流编排与条件分支class SalesState(TypedDict): date: str records: list summary: dict report: str status: str def query_node(state: SalesState) - SalesState: records query_sales(state[date]) return {records: records, status: queried} def check_data_node(state: SalesState) - SalesState: if not state[records]: return {status: no_data} return {status: has_data} def aggregate_node(state: SalesState) - SalesState: summary aggregate_by_region(state[records]) return {summary: summary, status: aggregated} def report_node(state: SalesState) - SalesState: report generate_markdown_report(state[summary], state[date]) return {report: report, status: reported} def push_node(state: SalesState) - SalesState: push_to_wecom(state[report]) return {status: pushed} def no_data_node(state: SalesState) - SalesState: push_to_wecom(f{state[date]} 无销售数据请检查数据源) return {status: no_data_notified} graph StateGraph(SalesState) graph.add_node(query, query_node) graph.add_node(check, check_data_node) graph.add_node(aggregate, aggregate_node) graph.add_node(report, report_node) graph.add_node(push, push_node) graph.add_node(no_data, no_data_node) graph.set_entry_point(query) graph.add_edge(query, check) graph.add_conditional_edges(check, lambda s: s[status], { has_data: aggregate, no_data: no_data }) graph.add_edge(aggregate, report) graph.add_edge(report, push) graph.add_edge(push, END) graph.add_edge(no_data, END) app graph.compile()这个图里check节点后面加了条件边根据有没有数据走不同分支。没数据时不是直接报错而是推送一条提醒这样运维人员能及时知道数据源出了问题。4.4 定时触发与异常处理工作流写好了怎么让它每天自动跑最简单的是用 cron 或 Windows 任务计划调一个 Python 脚本。# run_daily.py from datetime import datetime, timedelta def main(): yesterday (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) try: result app.invoke({date: yesterday}) print(f工作流完成状态{result[status]}) except Exception as e: # 异常时推送告警 push_to_wecom(f销售日报工作流失败{e}) raise if __name__ __main__: main()cron 配置0 9 * * * /path/to/agent-env/bin/python /path/to/run_daily.py /var/log/sales_report.log 21注意异常处理一定要加。Agent 工作流涉及多个外部调用任何一个环节都可能失败。失败时推送告警比静默失败强一百倍。5. 常见问题与排查技巧实录5.1 工具调用不触发或触发错误这是最常见的问题。模型该调工具的时候不调或者调了错误的工具。排查思路现象可能原因解决方法模型直接回答不调工具工具描述不清晰完善 docstring写清楚什么时候用调用了错误的工具工具之间功能重叠合并或重命名让职责更单一参数传错参数类型不明确用类型注解加参数说明反复调用同一工具缺少终止条件在 prompt 里加“最多调用一次”约束我踩过最坑的一次是工具函数名太相似get_data和fetch_data两个函数模型分不清结果随机调。后来改成get_sales_data和get_user_data问题就没了。工具命名一定要有区分度。5.2 状态丢失或传递错误LangGraph 的状态是 TypedDict节点返回的字典会合并到全局状态。但有个坑如果你返回的 key 不在 State 定义里会被静默忽略。我调试了半天才发现是拼写错误categoy少了个r。另一个坑是状态覆盖。如果两个节点都返回status字段后执行的会覆盖前面的。解决办法是用Annotated加 reducerfrom typing import Annotated import operator class State(TypedDict): logs: Annotated[list, operator.add] # 追加而不是覆盖5.3 循环与死锁LangGraph 支持循环但如果不加终止条件就会无限循环。我见过一个案例Agent 判断“数据不完整就重新查询”但查询工具永远返回不完整数据结果死循环。解决办法有两个一是加最大迭代次数二是加条件边判断。LangGraph 的recursion_limit可以设上限result app.invoke(input, {recursion_limit: 25})超过限制会抛异常至少不会把资源耗光。5.4 性能优化减少不必要的模型调用Agent 工作流慢多半是模型调用太多。优化思路能规则化的不用模型比如分类如果关键词匹配能覆盖 80% 的情况就用规则剩下 20% 再走模型。批量处理多个相似任务合并成一次调用。缓存相同输入的结果缓存起来避免重复推理。小模型做简单任务分类、提取用便宜的小模型复杂推理再用大模型。我实测下来一个原本需要 8 次模型调用的工作流优化后降到 3 次整体耗时从 40 秒降到 12 秒。5.5 调试技巧把中间状态打出来Agent 工作流是黑盒出问题时很难定位。我的做法是在每个节点加日志把输入输出都打出来def debug_node(name): def decorator(func): def wrapper(state): print(f[{name}] 输入: {state}) result func(state) print(f[{name}] 输出: {result}) return result return wrapper return decorator配合 LangSmith 或 LangFuse 这类追踪工具能看到完整的调用链。不过这些工具需要额外配置前期用 print 就够了。6. 从单 Agent 到多 Agent工作流的扩展思路6.1 什么时候需要多 Agent单 Agent 能搞定的事不要上多 Agent。多 Agent 的复杂度是指数级上升的。但以下几种情况多 Agent 确实更合适职责差异大一个负责数据查询一个负责内容生成一个负责审核。每个 Agent 有自己的工具集和 prompt。需要并行多个子任务可以同时跑最后汇总。需要对抗一个生成一个审核互相制衡。6.2 用 LangGraph 实现 Supervisor 模式Supervisor 模式是最常用的多 Agent 架构一个主管 Agent 负责分派任务多个工人 Agent 负责执行。class SupervisorState(TypedDict): task: str next_agent: str results: list def supervisor_node(state: SupervisorState) - SupervisorState: # 根据任务决定下一个 Agent if 数据 in state[task]: return {next_agent: data_agent} elif 报告 in state[task]: return {next_agent: report_agent} return {next_agent: end} def data_agent_node(state: SupervisorState) - SupervisorState: # 数据 Agent 的逻辑 return {results: state[results] [数据已查询]} def report_agent_node(state: SupervisorState) - SupervisorState: return {results: state[results] [报告已生成]}这种模式的好处是职责清晰每个 Agent 的 prompt 可以针对性优化。代价是需要设计好路由逻辑否则容易互相踢皮球。6.3 多 Agent 的坑通信开销与状态同步多 Agent 最大的问题是通信。每个 Agent 之间传递信息都要经过状态状态越大传递越慢。我的经验是状态里只放必要信息大对象存外部比如文件、数据库状态里放引用。Agent 之间尽量少来回能一次传完就一次传完。加超时机制防止某个 Agent 卡住拖垮整个流程。7. 上线前的检查清单与个人经验工作流跑通只是第一步上线前还有几件事必须做。检查清单所有工具函数都有异常处理失败时不会静默关键节点有日志能追溯每一步的输入输出有超时和重试机制外部调用不会无限等待状态持久化已配置崩溃后能恢复有告警机制失败时能通知到人敏感信息API Key、数据库密码不在代码里硬编码个人经验我最早做自动化工作流时总想一步到位把所有业务逻辑都塞进去。结果就是调试困难、维护成本高。后来学乖了先跑通最小闭环再逐步加功能。一个能稳定跑的单步骤工作流比一个经常挂的复杂工作流有价值得多。另外不要迷信模型。能用规则的地方就用规则模型只用在真正需要理解语义的地方。我见过有人用模型做日期格式转换纯属浪费。工具函数里该写死的逻辑就写死Agent 负责的是编排不是替代所有代码。最后监控比开发更重要。工作流上线后你不可能天天盯着。加个简单的健康检查每天跑完发个状态通知出问题时能第一时间知道。这个习惯能帮你省掉很多半夜被叫起来排查的麻烦。