Python + Agent SDK 实战：从零搭建自动化工作流与多 Agent 协作

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 SDK	LangGraph
上手难度	低，几行代码就能跑	中，需要理解图结构
状态管理	简单，靠 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 点，从公司内部系统拉取前一天的销售数据，按区域汇总，生成日报，推送到企业微信。

拆解步骤：

1. 调用内部 API 获取原始数据
2. 数据清洗和格式转换
3. 按区域分组汇总
4. 计算同比环比
5. 生成自然语言描述
6. 推送到企业微信

其中步骤 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}, timeout=30 ) 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}, timeout=10 ) 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 判断条件不满足跳到 B，B 又跳回 A，永远出不来。

排查方法：在编译图的时候加上recursion_limit：

app = 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_type=DailyReport )

这样拿到的结果直接是结构化对象，后续处理不用再解析文本。

5.2 加缓存减少重复调用

有些 Tool 的调用成本高（比如调外部 API 有费用），可以加缓存：

from functools import lru_cache @lru_cache(maxsize=128) 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(f"process_refund called: order_id={order_id}, amount={amount}") try: result = do_refund(order_id, amount) logger.info(f"process_refund success: {result}") return {"success": True, "result": result} except Exception as e: logger.error(f"process_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(checkpointer=checkpointer, 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 转给客服 Agent）

OpenAI 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 条真实案例的测试集，每次上线前都跑一遍，帮我挡掉了不少回归问题。