ARTICLE DETAIL

资讯详情

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

Agent SDK与LangGraph实战:业务自动化工作流编排指南

Agent SDK与LangGraph实战:业务自动化工作流编排指南 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 还是 LangGraph2.1 两条路线的本质区别很多人一上来就问哪个更好这个问题本身就不太对。它们不是替代关系而是适用场景不同。OpenAI Agents SDK的设计哲学是最小可用。它把 Agent 的核心要素——指令、工具、循环、交接——用很少的抽象封装起来。你定义一个 Agent给它一组工具然后 run 一下它就会自己循环调用工具直到完成任务。代码量少上手快适合一个 Agent 干一件事的场景。LangGraph的设计哲学是显式编排。它把整个流程画成一张图节点是执行单元边是流转条件。你可以精确控制每一步走哪条路、什么时候暂停等人工确认、什么时候回退重试。代码量相对多但可控性强适合多步骤、有分支、要审计的场景。我个人的经验是先用 Agents SDK 把单点跑通验证价值当流程开始出现分支和人工介入需求时再迁移到 LangGraph。不要一上来就上重武器容易把自己绕进去。2.2 一张表看清选型依据维度OpenAI Agents SDKLangGraph上手难度低几十行能跑中需要理解图概念流程控制隐式模型自主循环显式开发者定义节点和边分支处理靠模型判断弱控制条件边强控制人工介入需要自己实现内置 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-agentsLangGraph 是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需要的数量, gt0) 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这不是写给人看的是写给模型看的。写得越清楚模型调用越准。gt0这种约束也要加上能挡掉一部分无效调用。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(checkpointermemory)调用的时候要传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_turns10)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 不是写完就完事的它更像一个需要持续调教的员工。你给它清晰的职责、好用的工具、明确的边界它就能干得不错你放任不管它就会在各种边缘情况上翻车。把业务场景转成工作流技术只是一半另一半是对业务本身的理解和持续打磨。
返回列表