ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Agent SDK与LangGraph实战:从零搭建业务自动化工作流

Agent SDK与LangGraph实战:从零搭建业务自动化工作流 1. 业务自动化的核心痛点与 Agent SDK 的破局思路1.1 为什么传统脚本自动化越来越不够用做过业务自动化的朋友应该都有体会早几年写个 Python 脚本用requests拉数据、pandas做清洗、openpyxl写报表一套流程跑下来确实能省不少人力。但这两年情况变了——业务方提的需求越来越“活”今天要你从邮件里提取客户意向并自动分类明天要你根据聊天记录判断工单优先级后天又要你把散落在五六个系统里的信息汇总成一份决策摘要。这些任务的共同点是输入是非结构化的判断逻辑是模糊的执行路径是需要动态决定的。传统脚本处理这类问题非常吃力。你当然可以写一堆if-else和正则表达式但维护成本会随着规则数量指数级上升。我见过一个团队为了处理客服工单分类写了将近两千行规则代码结果业务规则一变整个文件推倒重来。这就是典型的“用确定性代码解决不确定性问题”方向本身就错了。Agent SDK 的出现本质上是把“判断”这件事交给了大模型把“执行”这件事留给了代码。你不再需要穷举所有分支而是定义一个 Agent告诉它目标是什么、有哪些工具可以用、边界在哪里剩下的让它自己规划。这听起来有点玄但落到工程上其实很实在——Agent 就是一个带工具调用能力的循环控制器它反复执行“思考→选择工具→执行→观察结果”这个循环直到任务完成或触发终止条件。1.2 Agent SDK 到底解决了什么问题拿 OpenAI Agents SDK 和 LangGraph 这两个目前最主流的方案来说它们解决的核心问题可以归纳为三层。第一层是工具调用的标准化。以前你要让模型调用外部函数得自己解析模型输出的 JSON、校验参数、处理异常、把结果塞回对话历史。这些脏活累活每个项目都要重写一遍。Agent SDK 把这套流程封装成了声明式的工具注册机制你只需要用装饰器标注一个函数SDK 自动生成工具描述、处理调用协议、管理返回值。第二层是多步骤任务的编排。一个真实的业务场景往往不是“问一句答一句”而是“先查订单→再判断是否符合退款条件→符合就发起退款→不符合就生成说明→最后通知客户”。LangGraph 用图结构来表达这种流程节点是处理步骤边是流转条件状态在节点间传递。相比链式调用图结构能表达循环、分支、并行更贴近真实业务逻辑。第三层是状态管理与可观测性。Agent 执行过程中会产生大量中间状态——对话历史、工具调用记录、临时变量。这些状态如果管理不好调试就是噩梦。LangGraph 的检查点机制允许你在任意节点保存和恢复状态OpenAI Agents SDK 则提供了追踪功能能回放整个执行链路。对于生产环境来说这比“能跑通”重要得多。1.3 什么样的业务场景适合用 Agent 工作流不是所有自动化都值得上 Agent。我的判断标准很简单如果这个任务的规则可以用一张决策表穷举完那就别用 Agent。比如“每天凌晨把 A 系统的数据同步到 B 系统”这种确定性任务用 Airflow 或者简单的定时脚本就够了上 Agent 反而是杀鸡用牛刀。真正适合 Agent 的场景通常具备三个特征。一是输入非结构化比如邮件正文、聊天记录、PDF 合同、网页内容。二是判断需要语义理解比如“这封投诉邮件是否紧急”“这个需求描述属于哪个产品线”。三是执行路径依赖中间结果比如“先查库存有货才报价没货就推荐替代品”。这三个特征叠加就是 Agent 工作流的最佳射程。举个我实际做过的例子一个电商团队需要处理售后退款申请。流程是读取申请邮件→提取订单号和退款原因→查询订单状态→判断是否符合退款政策→符合则调用退款接口→生成回复邮件。这里面邮件解析需要语义理解政策判断需要结合订单状态和退款原因做推理回复邮件需要根据处理结果动态生成。用传统脚本写光政策判断那块就得维护上百条规则用 Agent 工作流核心逻辑就是三个工具加一段提示词。2. 从零搭建 Agent 工作流的核心组件拆解2.1 环境准备与依赖选型先把环境搭起来。Python 版本建议 3.10 以上因为 Agent SDK 和 LangGraph 都用到了较新的类型注解特性。虚拟环境用venv或者conda都行我个人习惯venv轻量且够用。python -m venv agent-env source agent-env/bin/activate # Windows 用 agent-env\Scripts\activate pip install openai-agents langgraph langchain-openai python-dotenv这里解释一下几个核心依赖的分工。openai-agents是 OpenAI 官方出的 Agent SDK提供了 Agent 定义、工具注册、执行循环这些基础能力。langgraph负责流程编排把多个 Agent 或者处理步骤组织成图。langchain-openai提供模型接口的封装方便切换不同模型。python-dotenv用来管理 API 密钥这类敏感配置。注意不要把 API 密钥硬编码在代码里。用.env文件管理并且把.env加入.gitignore。我见过太多因为密钥泄露导致账单爆炸的案例。如果你用的是 LangGraph 的较新版本可能还需要装langgraph-checkpoint-sqlite来做状态持久化。这个后面讲到检查点的时候再说。2.2 工具函数的定义规范与注册方式工具是 Agent 的手和脚。定义工具函数有几个硬性要求踩过坑的人都知道。第一函数签名必须带类型注解。Agent SDK 靠类型注解生成工具的 JSON Schema没有注解它不知道怎么传参。第二文档字符串必须写清楚。模型是根据文档字符串来判断什么时候该调用这个工具的写得含糊模型就会乱调。第三返回值尽量结构化。返回一个字典比返回一个长字符串好模型更容易从中提取关键信息。from agents import function_tool function_tool def query_order_status(order_id: str) - dict: 根据订单号查询订单状态。 Args: order_id: 订单编号格式为 ORD 开头的字符串 Returns: 包含 status状态、amount金额、create_time创建时间的字典 # 实际项目中这里调用订单系统接口 return { status: 已发货, amount: 299.00, create_time: 2024-01-15 }这个function_tool装饰器做的事情比看起来多。它读取函数的类型注解和文档字符串生成符合 OpenAI 工具调用规范的 JSON Schema注册到 Agent 的工具列表中。当模型决定调用这个工具时SDK 自动解析模型返回的参数、调用函数、把结果格式化后塞回对话。2.3 Agent 的配置参数与行为控制定义一个 Agent 本质上是在配置它的“人格”和“能力边界”。核心参数有这么几个。name是标识符调试的时候用。instructions是系统提示词决定了 Agent 的角色、行为准则和输出格式。tools是可用工具列表。model指定用哪个模型。还有一个容易被忽略但很重要的参数是model_settings可以控制温度、最大 token 数这些。from agents import Agent refund_agent Agent( name退款处理助手, instructions你是一个电商售后处理助手。你的任务是 1. 从用户消息中提取订单号和退款原因 2. 调用 query_order_status 查询订单状态 3. 根据退款政策判断是否批准退款 4. 如果批准调用 process_refund 发起退款 5. 生成一段简洁的回复说明处理结果 退款政策 - 订单状态为已发货且金额小于 500 元可直接批准 - 订单状态为已签收且退款原因包含质量问题需转人工审核 - 其他情况一律拒绝并说明原因 , tools[query_order_status, process_refund], modelgpt-4o )instructions的写法很讲究。我的经验是把判断逻辑写成明确的规则把输出格式写成模板。不要指望模型自己“理解”业务规则你得把规则翻译成它能执行的指令。上面这个例子里退款政策用列表形式写清楚模型执行起来就稳定得多。2.4 LangGraph 的图结构设计入门单个 Agent 能处理的任务有限复杂业务需要多个 Agent 协作或者多步骤流转。LangGraph 用图来表达这种流转关系。图的基本元素是节点和边。节点是一个处理函数接收当前状态、返回状态更新。边定义了节点之间的流转关系可以是固定的也可以是条件性的。状态是一个共享的字典在所有节点之间传递。from langgraph.graph import StateGraph, END from typing import TypedDict class WorkflowState(TypedDict): user_message: str order_id: str order_status: str refund_decision: str reply: str def extract_info(state: WorkflowState) - WorkflowState: # 从用户消息中提取订单号 state[order_id] ORD12345 # 实际用模型提取 return state def check_order(state: WorkflowState) - WorkflowState: result query_order_status(state[order_id]) state[order_status] result[status] return state def make_decision(state: WorkflowState) - WorkflowState: if state[order_status] 已发货: state[refund_decision] 批准 else: state[refund_decision] 拒绝 return state graph StateGraph(WorkflowState) graph.add_node(extract, extract_info) graph.add_node(check, check_order) graph.add_node(decide, make_decision) graph.set_entry_point(extract) graph.add_edge(extract, check) graph.add_edge(check, decide) graph.add_edge(decide, END) app graph.compile()这个例子里每个节点做一件明确的事状态在节点间传递。实际项目中节点内部可以调用 Agent这样就把“确定性流程”和“智能判断”结合起来了——流程骨架是确定的每个环节的判断交给 Agent。3. 完整实操把退款处理场景转化为 Agent 工作流3.1 场景拆解与流程设计假设我们有一个电商售后场景用户发来一段消息可能是邮件也可能是聊天记录内容五花八门。我们需要从中提取订单号、判断退款是否合理、执行退款、生成回复。这个场景的难点在于用户消息格式不固定退款政策有多条分支回复内容需要根据处理结果动态生成。我的设计思路是分四步走。第一步用 Agent 做信息提取把非结构化的用户消息转成结构化的订单号和退款原因。第二步用代码做订单查询这是确定性操作不需要 Agent。第三步用 Agent 做政策判断因为政策条款需要语义理解。第四步用 Agent 生成回复因为回复要自然且贴合上下文。这个设计的关键决策是把确定性操作和智能判断分开。订单查询是纯 API 调用用代码做又快又稳信息提取和政策判断涉及语义理解交给 Agent。很多新手容易犯的错误是把所有环节都塞给 Agent结果又慢又不稳定。3.2 信息提取 Agent 的实现细节信息提取看起来简单实际上是最容易出问题的环节。用户可能写“我上周买的那个东西要退”也可能写“订单 ORD12345 申请退款”还可能写一大段抱怨最后才提订单号。提取 Agent 的提示词需要覆盖这些情况。from agents import Agent, Runner from pydantic import BaseModel class ExtractedInfo(BaseModel): order_id: str | None refund_reason: str confidence: float extract_agent Agent( name信息提取, instructions从用户消息中提取以下信息 - order_id: 订单号格式为 ORD 开头。如果找不到返回 null - refund_reason: 退款原因用一句话概括 - confidence: 你对提取结果的置信度0 到 1 之间 只返回 JSON不要有其他内容。, modelgpt-4o-mini, output_typeExtractedInfo ) async def extract(message: str) - ExtractedInfo: result await Runner.run(extract_agent, message) return result.final_output这里用了output_type参数让 SDK 自动把模型输出解析成 Pydantic 模型。这比手动解析 JSON 靠谱得多模型输出格式不对时 SDK 会自动重试。gpt-4o-mini做信息提取足够了没必要用大模型成本和延迟都更优。实操心得信息提取 Agent 的提示词里一定要加“如果找不到返回 null”这类兜底指令。否则模型会编造一个订单号出来后面流程全乱套。3.3 政策判断 Agent 与工具调用的配合政策判断环节需要结合订单状态和退款原因做推理。这里的关键是把政策条款写清楚同时给 Agent 提供必要的工具。function_tool def get_refund_policy(category: str) - str: 获取指定类别的退款政策。 Args: category: 商品类别如电子产品、服装、食品 policies { 电子产品: 7天内无理由退款超过7天需质量问题证明, 服装: 15天内无理由退款吊牌需完整, 食品: 不支持无理由退款质量问题可全额退 } return policies.get(category, 通用政策7天内可申请退款) policy_agent Agent( name政策判断, instructions根据订单信息和退款原因判断是否批准退款。 判断步骤 1. 调用 get_refund_policy 获取对应类别的政策 2. 对比订单状态和退款原因是否符合政策 3. 返回判断结果批准 / 拒绝 / 转人工 输出格式{decision: 批准/拒绝/转人工, reason: 判断依据}, tools[get_refund_policy], modelgpt-4o )这里有个设计技巧把政策数据做成工具而不是塞进提示词。原因是政策可能会变做成工具后更新政策只需要改数据不用动提示词。而且工具调用会让模型更认真地“查阅”政策而不是凭记忆瞎编。3.4 用 LangGraph 串联全流程现在把各个环节用 LangGraph 串起来。图的结构是提取信息→查询订单→判断政策→执行退款→生成回复。其中查询订单和判断政策之间有个条件分支如果订单不存在直接跳到生成回复说明情况。from langgraph.graph import StateGraph, END from typing import TypedDict, Literal class RefundState(TypedDict): user_message: str order_id: str | None order_info: dict | None decision: str reply: str async def extract_node(state: RefundState) - RefundState: info await extract(state[user_message]) state[order_id] info.order_id return state async def query_node(state: RefundState) - RefundState: if not state[order_id]: state[order_info] None return state state[order_info] query_order_status(state[order_id]) return state def route_after_query(state: RefundState) - Literal[judge, reply]: if state[order_info] is None: return reply return judge async def judge_node(state: RefundState) - RefundState: result await Runner.run(policy_agent, str(state[order_info])) state[decision] result.final_output[decision] return state async def reply_node(state: RefundState) - RefundState: if state[order_info] is None: state[reply] 未找到对应订单请确认订单号是否正确。 else: state[reply] f您的退款申请已{state[decision]}。 return state graph StateGraph(RefundState) graph.add_node(extract, extract_node) graph.add_node(query, query_node) graph.add_node(judge, judge_node) graph.add_node(reply, reply_node) graph.set_entry_point(extract) graph.add_edge(extract, query) graph.add_conditional_edges(query, route_after_query, { judge: judge, reply: reply }) graph.add_edge(judge, reply) graph.add_edge(reply, END) workflow graph.compile()这个图结构清晰表达了业务逻辑提取→查询→有订单则判断无订单则直接回复→回复。条件边add_conditional_edges是 LangGraph 处理分支的标准方式路由函数的返回值决定走哪条边。3.5 状态持久化与断点续跑生产环境里工作流可能跑一半挂了或者需要人工介入审核。这时候状态持久化就很重要。LangGraph 支持检查点机制可以在每个节点执行后保存状态。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.db) workflow graph.compile(checkpointermemory) # 执行时指定 thread_id config {configurable: {thread_id: refund-001}} result await workflow.ainvoke( {user_message: 我要退款订单 ORD12345}, configconfig ) # 如果中途需要人工审核可以暂停后恢复 # 恢复时用同样的 thread_id 即可thread_id是会话标识同一个 thread_id 的状态会被保存和恢复。这个机制对于需要人工审核的场景特别有用——Agent 判断“转人工”后暂停人工审核完再恢复执行状态不会丢。4. 常见问题排查与生产环境避坑指南4.1 工具调用失败的典型原因与修复工具调用失败是最高频的问题。我整理了一个排查表按出现频率排序。问题现象根本原因修复方法模型不调用工具工具描述太模糊在 docstring 里写清楚“什么时候用这个工具”参数格式错误类型注解缺失或不明确所有参数加类型注解复杂参数用 Pydantic 模型工具调用死循环没有终止条件设置最大迭代次数或在提示词里加“最多调用 N 次”返回值解析失败返回了非 JSON 可序列化对象返回值统一转成 dict/list/str工具调用超时外部接口响应慢加超时参数超时后返回错误信息让模型处理其中“模型不调用工具”最常见。很多人以为写了工具模型就会用实际上模型是根据 docstring 判断的。如果 docstring 只写“查询订单”模型可能觉得“我现在不需要查订单”。改成“当用户提到订单号或需要查询订单状态时调用此工具”调用率会明显提升。4.2 提示词工程的实战技巧Agent 的提示词和普通对话的提示词写法不一样。普通对话追求自然Agent 提示词追求可执行。我的经验是遵循三个原则。第一用编号列表写步骤。模型对编号列表的执行准确率明显高于段落描述。第二给出输出格式示例。不要只说“返回 JSON”要给出具体的 JSON 结构示例。第三明确边界情况。比如“如果信息不足返回 need_more_info 而不是猜测”。instructions 你是退款处理助手。按以下步骤执行 1. 检查用户消息中是否包含订单号ORD 开头 2. 如果没有订单号返回 {status: need_order_id, message: 请提供订单号} 3. 如果有订单号调用 query_order_status 查询 4. 根据查询结果判断 - 订单不存在 → {status: not_found} - 订单存在且符合政策 → {status: approved} - 订单存在但不符合政策 → {status: rejected, reason: 具体原因} 输出必须是 JSON不要有其他内容。这种写法看起来啰嗦但实测下来稳定性比“优雅”的提示词高很多。Agent 不是人它需要明确的指令而不是暗示。4.3 成本控制与性能优化Agent 工作流的成本主要来自模型调用。一个四步流程如果每步都用 GPT-4o单次执行成本可能到几毛钱。量大了就是一笔不小的开支。优化手段有这么几个。模型分级。信息提取、格式转换这类简单任务用gpt-4o-mini复杂推理才用gpt-4o。实测下来信息提取用 mini 的准确率和 4o 差距很小但成本差了一个数量级。缓存重复调用。同样的输入如果频繁出现可以加一层缓存。LangChain 提供了set_llm_cache接口支持内存缓存和 Redis 缓存。并行化。LangGraph 支持并行节点没有依赖关系的步骤可以同时执行。比如“查询订单”和“查询用户历史”可以并行能省不少时间。设置 token 上限。在model_settings里设置max_tokens防止模型输出过长。Agent 场景下输出通常不需要太长设个 500 到 1000 就够了。4.4 生产部署的注意事项从 demo 到生产有几个坑必须提前填。错误处理。Agent 执行过程中任何一步都可能失败——模型超时、工具报错、网络抖动。每个节点都要有 try-except失败后要么重试要么走降级路径。不要让一个节点的失败导致整个工作流崩溃。日志与追踪。OpenAI Agents SDK 自带追踪功能可以记录每次模型调用和工具调用的详细信息。生产环境一定要开启出问题时能快速定位。LangGraph 的检查点也能当日志用每个节点的输入输出都有记录。并发控制。如果工作流会被多个用户同时触发要注意并发问题。数据库连接、外部 API 调用都要考虑限流。LangGraph 的检查点机制在并发场景下需要配合数据库的事务隔离级别使用。版本管理。提示词和工具定义都是代码的一部分要纳入版本管理。我见过提示词改了一行导致线上效果大幅波动的案例没有版本管理根本回滚不了。最后分享一个我踩过的坑Agent 的提示词里千万不要写“尽可能帮助用户”这种模糊指令。模型会过度解读该拒绝的也批准了。提示词要具体、可执行、有明确边界。宁可写得“死板”一点也不要给模型太多自由发挥的空间。业务场景要的是稳定不是创意。
返回列表