★ FEATURED ARTICLE 1. 从业务痛点到自动化工作流为什么选择 Python Agent SDK业务场景里最不缺的就是“重复但需要动脑子”的活。比如每天早上要从三个系统里拉数据、比对差异、生成日报、推送给相关负责人比如客服收到退款申请后要查订单状态、判断是否符合规则、计算退款金额、写回工单系统。这些流程有明确的步骤但每一步都涉及判断、查询、格式转换纯靠写死的脚本很难覆盖所有分支靠人工又太慢。我最早的做法是写 Python 脚本把每个步骤硬编码进去。订单状态变了、规则调整了、接口字段改了脚本就得跟着改维护成本极高。后来开始用 LangChain 做链式调用把 LLM 嵌进流程里做判断和文本生成灵活了不少但遇到需要多轮决策、动态选择工具、根据中间结果调整后续步骤的场景链式结构就显得僵硬。Agent SDK 解决的就是这个问题。它把“LLM 做决策 工具调用 状态管理”这套模式封装成了可编程的框架你只需要定义好工具和任务目标Agent 会自己决定调用哪个工具、按什么顺序调用、什么时候结束。OpenAI Agents SDK 和 LangGraph 是目前两个主流选择前者更轻量、上手快后者更偏向复杂状态机和多 Agent 协作。这篇文章面向的是有一定 Python 基础、想把业务场景落地成自动化工作流的开发者。我会从整体设计思路讲起拆解核心概念和实操要点然后给出一套完整的实现流程最后分享我在实际项目中踩过的坑和排查技巧。全文基于 OpenAI Agents SDK 和 LangGraph 的常见实践代码可以直接参考复现。2. 核心概念拆解Agent、Tool、Workflow 到底怎么理解2.1 Agent 不是聊天机器人是一个会自己做决定的执行器很多人第一次接触 Agent 会把它当成“更聪明的 ChatGPT”这个理解偏了。Agent 的核心能力不是聊天而是根据目标自主决策。你给它一个任务描述它自己分析需要哪些步骤、调用哪些工具、按什么顺序执行遇到不确定的情况还会主动追问或重试。用生活化的类比普通脚本像自动售货机你按哪个按钮它就出哪个货Agent 像你雇的一个助理你告诉它“帮我把这周的报销单整理了”它会自己去翻邮件、识别发票、分类汇总、填表中间遇到缺发票的情况还会来问你。在 OpenAI Agents SDK 里Agent 的定义非常简洁from agents import Agent, Runner, function_tool function_tool def query_order_status(order_id: str) - str: 根据订单号查询订单状态 # 实际项目中这里调用内部 API return f订单 {order_id} 状态已发货 agent Agent( name订单助手, instructions你是一个订单处理助手根据用户提供的订单号查询状态并给出处理建议。, tools[query_order_status], ) result Runner.run_sync(agent, 帮我查一下订单 ORD-2024-001 的状态) print(result.final_output)这段代码里Agent定义了角色和可用工具Runner负责驱动整个决策循环。Agent 收到任务后会判断需要调用query_order_status传入正确的参数拿到结果后再生成最终回复。整个过程不需要你写 if-else 来判断“什么时候该调哪个工具”。2.2 Tool 是 Agent 的手和脚定义质量决定落地效果Agent 再聪明没有工具也只能空谈。Tool 就是 Agent 能实际操作外部世界的手段——查数据库、调 API、读写文件、发邮件、执行计算都是 Tool。定义 Tool 有几个关键点容易被忽略第一描述要写清楚。LLM 是根据 Tool 的 docstring 和参数描述来决定是否调用的。如果你的描述含糊Agent 可能该调的时候不调不该调的时候乱调。比如query_order_status的 docstring 写“查询订单”就不如写“根据订单号查询订单的当前状态包括是否发货、物流信息、预计到达时间”来得明确。第二参数类型要严格。用 Python 的类型注解SDK 会自动生成 JSON Schema 给 LLM。参数类型不明确会导致 Agent 传错格式比如把字符串传成数字。第三错误处理要内置。Tool 执行失败时应该返回有意义的错误信息而不是直接抛异常。Agent 看到错误信息后可能会尝试其他方案直接抛异常则会导致整个流程中断。function_tool def calculate_refund(order_id: str, reason: str) - dict: 根据订单号和退款原因计算退款金额。 Args: order_id: 订单编号格式为 ORD-YYYY-NNN reason: 退款原因可选值质量问题、七天无理由、发错货 try: # 实际业务逻辑 order fetch_order(order_id) if reason 七天无理由: amount order.total * 0.95 # 扣除手续费 elif reason 质量问题: amount order.total else: amount order.total return {success: True, amount: round(amount, 2)} except Exception as e: return {success: False, error: str(e)}2.3 Workflow 是把多个 Agent 串起来的骨架单个 Agent 能处理的任务有限真实业务场景往往需要多个 Agent 协作。比如一个完整的退款流程可能涉及意图识别 Agent → 订单查询 Agent → 规则判断 Agent → 退款执行 Agent → 通知 Agent。这时候就需要 Workflow 来编排。LangGraph 在这方面做得更成熟它用图结构定义状态流转from langgraph.graph import StateGraph, END from typing import TypedDict class RefundState(TypedDict): order_id: str reason: str refund_amount: float status: str def identify_intent(state: RefundState): # 意图识别逻辑 return {status: intent_identified} def check_order(state: RefundState): # 订单查询逻辑 return {status: order_checked} def process_refund(state: RefundState): # 退款处理逻辑 return {status: refund_processed} workflow StateGraph(RefundState) workflow.add_node(identify, identify_intent) workflow.add_node(check, check_order) workflow.add_node(refund, process_refund) workflow.set_entry_point(identify) workflow.add_edge(identify, check) workflow.add_edge(check, refund) workflow.add_edge(refund, END) app workflow.compile()LangGraph 的优势在于状态管理清晰、支持条件分支和循环、可以持久化中间状态。如果你的业务场景步骤固定、分支不多OpenAI Agents SDK 的轻量模式就够了如果需要复杂的状态流转和人工介入节点LangGraph 更合适。2.4 两者怎么选一张表说清楚维度OpenAI Agents SDKLangGraph上手难度低几行代码就能跑中需要理解图结构状态管理简单靠 Agent 自身强显式状态定义多 Agent 协作支持 handoff支持更灵活条件分支靠 Agent 决策显式条件边持久化需自行实现内置 checkpointer适用场景单 Agent 多工具复杂工作流、多 Agent调试体验简单直接需要理解图执行路径我个人的经验是先用 OpenAI Agents SDK 快速验证可行性如果发现流程分支太多、状态太复杂再迁移到 LangGraph。不要一上来就上重框架容易陷入过度设计。3. 从零搭建把业务场景转化为 Agent 工作流的完整步骤3.1 环境准备与依赖安装先把基础环境搭好。Python 版本建议 3.10 以上因为 Agent SDK 和 LangGraph 都用到了较新的类型注解特性。# 创建虚拟环境 python -m venv agent-env source agent-env/bin/activate # Windows 用 agent-env\Scripts\activate # 安装核心依赖 pip install openai-agents pip install langgraph pip install langchain-openai pip install pydantic如果你用的是 VSCode记得在设置里把 Python 解释器指向虚拟环境里的那个否则会出现“模块找不到”的问题。这个坑我踩过好几次明明 pip install 成功了运行时报 ImportError就是因为解释器选错了。环境变量配置export OPENAI_API_KEY你的密钥注意不要把密钥硬编码在代码里用环境变量或 .env 文件管理。.env 文件记得加到 .gitignore 里。3.2 业务场景拆解以“自动拉取数据生成日报”为例我拿一个真实场景来演示每天早上 9 点从公司内部系统拉取前一天的销售数据按区域汇总生成日报推送到企业微信。拆解步骤调用内部 API 获取原始数据数据清洗和格式转换按区域分组汇总计算同比环比生成自然语言描述推送到企业微信其中步骤 1、2、3、4、6 是确定性操作适合写成 Tool步骤 5 需要 LLM 生成自然语言适合让 Agent 来做。整体流程可以用一个 Agent 加多个 Tool 实现。3.3 定义 Tool把每个业务操作封装成 Agent 能调用的函数from agents import function_tool import requests import pandas as pd from datetime import datetime, timedelta function_tool def fetch_sales_data(date: str) - dict: 从内部系统拉取指定日期的销售数据。 Args: date: 日期格式 YYYY-MM-DD try: resp requests.get( https://internal-api.example.com/sales, params{date: date}, timeout30 ) resp.raise_for_status() return {success: True, data: resp.json()} except Exception as e: return {success: False, error: str(e)} function_tool def aggregate_by_region(raw_data: list) - dict: 按区域汇总销售数据。 Args: raw_data: 原始销售数据列表每条包含 region, amount, order_count 字段 df pd.DataFrame(raw_data) result df.groupby(region).agg( total_amount(amount, sum), total_orders(order_count, sum) ).reset_index() return {success: True, summary: result.to_dict(records)} function_tool def calculate_growth(current: float, previous: float) - dict: 计算同比增长率。 Args: current: 当前值 previous: 同期值 if previous 0: return {success: True, growth: N/A同期为零} growth (current - previous) / previous * 100 return {success: True, growth: f{growth:.2f}%} function_tool def send_to_wechat(content: str) - dict: 推送消息到企业微信。 Args: content: 消息内容支持 Markdown 格式 try: resp requests.post( https://internal-api.example.com/wechat/send, json{content: content}, timeout10 ) resp.raise_for_status() return {success: True} except Exception as e: return {success: False, error: str(e)}每个 Tool 都返回 dict 格式包含success字段。这样 Agent 能根据成功与否决定下一步操作而不是被异常打断。3.4 组装 Agent定义角色、指令和工具集from agents import Agent, Runner daily_report_agent Agent( name日报生成助手, instructions你是一个销售日报生成助手。你的任务是 1. 调用 fetch_sales_data 获取指定日期的销售数据 2. 调用 aggregate_by_region 按区域汇总 3. 调用 calculate_growth 计算各区域的同比增长 4. 用自然语言生成一份简洁的日报包含各区域销售额、订单量、同比增长 5. 调用 send_to_wechat 推送日报 注意如果任何步骤失败先重试一次仍然失败则生成错误报告并推送。, tools[fetch_sales_data, aggregate_by_region, calculate_growth, send_to_wechat], ) result Runner.run_sync( daily_report_agent, f生成 {datetime.now().strftime(%Y-%m-%d)} 的销售日报 ) print(result.final_output)这段代码跑起来后Agent 会自己决定先调哪个 Tool、传什么参数、拿到结果后下一步做什么。你不需要写任何编排逻辑。3.5 用 LangGraph 编排多 Agent 协作流程如果场景更复杂比如需要人工审核节点、需要根据金额大小走不同审批流程LangGraph 更合适。下面是一个带条件分支的退款流程from langgraph.graph import StateGraph, END from typing import TypedDict, Literal class RefundState(TypedDict): order_id: str reason: str amount: float need_manual_review: bool status: str def check_order(state: RefundState) - RefundState: # 查询订单计算退款金额 amount 299.0 # 模拟 return {**state, amount: amount, status: order_checked} def auto_approve(state: RefundState) - RefundState: return {**state, status: auto_approved} def manual_review(state: RefundState) - RefundState: return {**state, status: pending_manual_review} def route_by_amount(state: RefundState) - Literal[auto, manual]: if state[amount] 500: return auto return manual workflow StateGraph(RefundState) workflow.add_node(check, check_order) workflow.add_node(auto, auto_approve) workflow.add_node(manual, manual_review) workflow.set_entry_point(check) workflow.add_conditional_edges(check, route_by_amount, { auto: auto, manual: manual }) workflow.add_edge(auto, END) workflow.add_edge(manual, END) app workflow.compile() result app.invoke({order_id: ORD-2024-001, reason: 质量问题}) print(result)LangGraph 的条件边让分支逻辑非常清晰而且每个节点的输入输出都是显式定义的调试起来比 Agent 的黑盒决策容易得多。4. 实操中的常见问题与排查技巧4.1 Agent 不调用 Tool 怎么办这是最常见的问题。Agent 收到任务后直接用自己的知识回答完全不调 Tool。原因通常有三个Tool 描述不够明确。LLM 判断是否需要调用 Tool主要看描述。如果描述写得太泛LLM 会觉得“这个问题我自己能回答”。解决办法是把描述写具体明确说明“什么时候必须调用这个 Tool”。指令里没有强制要求。在 Agent 的 instructions 里明确写“你必须先调用 XXX 获取数据不能凭记忆回答”。我实测下来加了这句话之后 Tool 调用率明显提升。模型能力不够。小模型在 Tool 调用上的表现确实差一些。如果条件允许用 GPT-4 级别的模型做 Agent 决策用便宜模型做文本生成。4.2 Tool 调用参数传错怎么排查Agent 传错参数通常是因为参数描述不清晰。比如一个date参数LLM 可能传2024-01-01也可能传2024/01/01还可能传昨天。解决办法是在 docstring 里明确格式要求并且在 Tool 内部做参数校验和归一化function_tool def fetch_data(date: str) - dict: 获取指定日期的数据。 Args: date: 日期必须为 YYYY-MM-DD 格式例如 2024-01-15 # 参数归一化 date date.replace(/, -).strip() try: datetime.strptime(date, %Y-%m-%d) except ValueError: return {success: False, error: f日期格式错误{date}请使用 YYYY-MM-DD 格式} # 继续处理4.3 工作流执行到一半卡住或死循环LangGraph 里如果条件边写错了可能导致节点之间无限循环。比如 A 判断条件不满足跳到 BB 又跳回 A永远出不来。排查方法在编译图的时候加上recursion_limitapp workflow.compile() result app.invoke( {order_id: ORD-2024-001}, config{recursion_limit: 25} )超过限制会抛异常你就能定位到是哪个条件边出了问题。另外每个节点函数里加日志输出记录输入状态和输出状态能快速定位卡在哪一步。4.4 常见问题速查表问题现象可能原因排查方法解决方案Agent 不调 Tool描述不明确/指令未强制打印 Agent 决策日志细化 docstring指令中强制要求参数格式错误类型注解不清晰在 Tool 里打印收到的参数加参数校验和归一化工作流死循环条件边逻辑错误加 recursion_limit检查条件分支的覆盖完整性Tool 执行超时外部 API 响应慢加超时日志设置 timeout加重试机制多 Agent 状态丢失状态未正确传递打印每步 state用 TypedDict 显式定义状态输出格式不稳定指令不够具体收集多次输出对比在指令中给出输出示例4.5 几个我踩过的坑坑一Tool 返回太长的文本。有一次一个 Tool 返回了完整的 HTML 页面Agent 处理不了直接报错。后来改成只返回关键字段问题解决。Tool 的返回值要精简只给 Agent 需要的信息。坑二在 Agent 指令里写太多规则。指令太长会导致 LLM 注意力分散反而容易出错。我的经验是指令控制在 200 字以内复杂规则放到 Tool 里用代码实现。坑三忽略并发问题。多个 Agent 同时调用同一个 Tool 时如果 Tool 里有共享状态比如写同一个文件会出现竞争条件。解决办法是 Tool 设计成无状态的或者加锁。坑四没有做幂等。工作流重试时同一个操作可能被执行两次。比如退款操作重试导致重复退款。每个 Tool 都要考虑幂等性用唯一 ID 做去重。5. 进阶技巧让工作流更稳、更快、更好维护5.1 用 Pydantic 做结构化输出Agent 生成的文本如果直接推送给用户格式可能不稳定。用 Pydantic 定义输出结构让 Agent 按固定格式返回from pydantic import BaseModel class DailyReport(BaseModel): date: str total_amount: float total_orders: int regions: list[dict] summary: str agent Agent( name日报助手, instructions生成日报输出必须符合 DailyReport 结构, tools[...], output_typeDailyReport )这样拿到的结果直接是结构化对象后续处理不用再解析文本。5.2 加缓存减少重复调用有些 Tool 的调用成本高比如调外部 API 有费用可以加缓存from functools import lru_cache lru_cache(maxsize128) def _fetch_cached(date: str): return requests.get(...).json() function_tool def fetch_sales_data(date: str) - dict: 获取销售数据 try: data _fetch_cached(date) return {success: True, data: data} except Exception as e: return {success: False, error: str(e)}注意缓存 key 要包含所有影响结果的参数否则会拿到错误数据。5.3 日志和可观测性Agent 的决策过程是黑盒出问题时很难排查。建议在每个 Tool 里加日志import logging logger logging.getLogger(__name__) function_tool def process_refund(order_id: str, amount: float) - dict: 处理退款 logger.info(fprocess_refund called: order_id{order_id}, amount{amount}) try: result do_refund(order_id, amount) logger.info(fprocess_refund success: {result}) return {success: True, result: result} except Exception as e: logger.error(fprocess_refund failed: {e}) return {success: False, error: str(e)}LangGraph 可以用 callback 机制记录每个节点的执行情况OpenAI Agents SDK 可以自定义 Runner 的 hook。5.4 人工介入节点的设计有些场景需要人工审核比如大额退款。在 LangGraph 里可以用interrupt实现from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer, interrupt_before[manual]) # 执行到 manual 节点前暂停 result app.invoke(inputs, config{configurable: {thread_id: 1}}) # 人工审核后继续 app.update_state(config, {approved: True}) result app.invoke(None, config)这个模式在实际项目中非常实用既保留了自动化的效率又给了人工兜底的能力。5.5 性能优化并行执行独立步骤如果工作流里有多个互不依赖的步骤可以并行执行。LangGraph 支持并行节点workflow.add_node(fetch_sales, fetch_sales) workflow.add_node(fetch_inventory, fetch_inventory) workflow.add_node(fetch_customer, fetch_customer) # 三个节点并行执行 workflow.add_edge(start, fetch_sales) workflow.add_edge(start, fetch_inventory) workflow.add_edge(start, fetch_customer) # 都完成后汇总 workflow.add_edge(fetch_sales, aggregate) workflow.add_edge(fetch_inventory, aggregate) workflow.add_edge(fetch_customer, aggregate)并行执行能把总耗时从“各步骤之和”降到“最慢步骤的耗时”在数据拉取场景下提升非常明显。6. 从单 Agent 到多 Agent协作模式与选型建议6.1 什么时候需要多 Agent单 Agent 能处理的任务有上限。当出现以下情况时考虑拆成多 Agent任务涉及多个专业领域一个 Agent 的指令难以覆盖不同步骤需要不同的工具集全塞给一个 Agent 会导致决策混乱需要不同角色之间的交接比如销售 Agent 转给客服 AgentOpenAI Agents SDK 支持 handoff 机制from agents import Agent refund_agent Agent( name退款专员, instructions处理退款申请, tools[calculate_refund, process_refund] ) order_agent Agent( name订单助手, instructions处理订单查询遇到退款需求转给退款专员, tools[query_order_status], handoffs[refund_agent] )Agent 在运行过程中可以主动把任务转给另一个 Agent这个机制在客服场景下特别有用。6.2 多 Agent 协作的三种模式模式一流水线。Agent A 的输出作为 Agent B 的输入依次传递。适合步骤明确的场景。模式二路由分发。一个入口 Agent 判断任务类型分发给对应的专业 Agent。适合客服、工单分类场景。模式三辩论协作。多个 Agent 对同一问题给出方案再由一个汇总 Agent 综合。适合需要多角度分析的场景。选哪种模式取决于业务复杂度。我的建议是能用单 Agent 就不用多 Agent多 Agent 的调试成本是单 Agent 的好几倍。6.3 多 Agent 状态传递的注意事项多 Agent 协作时状态传递是最容易出问题的地方。每个 Agent 的输出格式要统一否则下游 Agent 解析不了。建议用 Pydantic 定义统一的 Agent 间通信协议class AgentMessage(BaseModel): source: str target: str task_type: str payload: dict timestamp: str所有 Agent 的输入输出都用这个结构能大幅减少格式不匹配的问题。7. 上线前的检查清单与运维要点7.1 上线前必须验证的几件事Tool 的边界情况。空数据、超大数据、特殊字符、超时这些都要测。我见过一个 Tool 在数据量为零时直接除零报错导致整个工作流崩溃。Agent 的异常处理。模拟 Tool 失败、API 超时、返回格式错误看 Agent 是否能优雅降级。理想情况下Agent 应该能识别失败并给出有意义的错误信息而不是直接崩溃。成本估算。Agent 每次决策都会消耗 token多轮决策的 token 消耗可能远超预期。上线前跑一批真实数据估算单次执行成本。并发压力。如果工作流会被多个用户同时触发要测试并发下的表现。Tool 里的共享资源数据库连接、文件句柄要处理好。7.2 运维监控的关键指标指标说明告警阈值建议单次执行耗时从触发到完成的时间超过 5 分钟告警Tool 调用失败率失败次数/总调用次数超过 10% 告警Token 消耗每次执行的 token 用量超过预算 2 倍告警工作流完成率成功完成/总触发次数低于 95% 告警人工介入率需要人工处理的占比超过 20% 需优化7.3 版本迭代与回滚Agent 的指令和 Tool 定义变更后行为可能发生很大变化。建议每次变更都做 A/B 测试用一批固定输入对比新旧版本输出。如果新版本表现下降能快速回滚。把 Agent 的指令、Tool 定义、工作流图都纳入版本管理每次上线打 tag。出问题时能快速定位到是哪个版本引入的。8. 我个人的实操体会这套东西我从去年开始在生产环境跑最大的感受是Agent 不是银弹它适合“步骤明确但分支多”的场景不适合“步骤本身就不清楚”的场景。如果你自己都说不清楚这个业务该怎么处理别指望 Agent 能帮你理清楚。另一个体会是Tool 的质量决定一切。Agent 再聪明Tool 写得烂结果就是灾难。我花在打磨 Tool 上的时间远多于调 Agent 指令的时间。每个 Tool 的输入输出、错误处理、边界情况都要像写生产级 API 一样认真对待。还有一点不要追求全自动。实际业务里总有一些边缘情况需要人工判断。设计工作流时预留人工介入节点比追求 100% 自动化更务实。我现在的做法是常规情况自动处理异常情况自动转人工人工处理完的结果再反馈给系统做学习。最后分享一个小技巧用真实数据做回归测试。每次改完 Agent 或 Tool拿一批历史真实数据跑一遍对比输出结果。这比写单元测试更能发现实际问题因为 LLM 的行为很难用断言来覆盖。我建了一个包含 200 条真实案例的测试集每次上线前都跑一遍帮我挡掉了不少回归问题。
阅读完成 · 觉得有帮助？
