这两年我跟不少团队聊 LLM 应用落地,发现最容易卡住的不是模型效果,而是流程设计。模型本身已经挺能打,但一旦要处理“多步骤、带判断、会调用工具”的真实业务,单靠一次 Prompt 根本撑不起来,这时候就需要用到 Agentic Workflow。说白了,Agentic Workflow 就是一套把大模型放进自动化流程里的工程方法:流程里每个节点的走向不是写死的,而是由 Agent 根据上下文动态决策。它跟传统工作流最大的区别,就是“决策”本身变成了流程的一部分。这篇文章我不讲空理论,直接从设计模式讲到代码实现,再讲到我实际跑项目时踩过的坑,尽量让你看完能直接动手搭一个属于自己的智能体工作流。
1. 先弄清楚:Agentic Workflow 到底改变了什么
1.1 从“固定流程”到“决策驱动流程”
我们以前做了很多年的自动化流程,本质上是把业务规则固化下来:如果 A 条件成立就走分支 B,否则走分支 C。这种方式在规则明确、输入结构固定的场景下非常稳定,比如订单状态流转、审批流、定时任务调度。但一旦遇到输入内容不可控、判断标准模糊的活儿,写死在代码里的 if-else 就会迅速失控。
举个例子:一个客服工单系统,用户的描述可能是“我上周买的东西到现在还没到,帮我查下”,也可能是“你们这什么破服务,我要投诉”,还可能是夹杂着截图、语音转文字、商品链接的混合信息。传统工作流拿到这种输入,根本不知道往哪里走。但在 Agentic Workflow 里,大模型首先理解意图,判断这是物流查询、退款申请还是投诉升级,然后决定后续调用哪些工具、需要走哪个分支、最终如何生成回复。整个流程的走向是在运行时动态决定的。
这个转变非常关键。它把原来“人先分析、再设计规则、最后编码”的模式,变成了“人在设计阶段定义好决策节点和可选路径,模型在运行时做选择”。你要做的不是把所有可能性写死,而是给 Agent 提供清晰的选择空间和判断依据。
1.2 它和普通 Workflow、Agent 之间的边界
很多同学会把 Agentic Workflow、Agent、Workflow 这三个概念混在一起。我用一个比较容易理解的方式拆解它们:
- Workflow(工作流):预先定义好的步骤序列,执行顺序固定,适合“怎么做已经明确”的任务。
- Agent(智能体):拥有大模型大脑,可以自主规划、调用工具、记忆上下文,适合“怎么做还没确定”的任务。
- Agentic Workflow(智能体工作流):将 Agent 作为流程中的决策节点,结合可编排的步骤和状态流转,既有 Workflow 的稳定性,又有 Agent 的灵活性。
换句话说,Agentic Workflow 不是把整个任务丢给 Agent 当黑盒跑,而是把 Agent 包装在流程的节点里。每个节点是“思考 + 行动”的最小闭环,节点与节点之间通过状态和数据传递来连接。这样既避免了纯 Agent 的自由发挥带来的不可控,也避免了纯 Workflow 的僵化。
从我自己的实践经验来看,设计一套好的 Agentic Workflow,前期的核心精力应该花在“哪些节点需要 Agent 决策、哪些节点用固定代码”这个问题上。把大模型用在刀刃上,让代码处理确定性逻辑,才能做到成本、稳定性、效果三者的平衡。
2. 五种主流工作流模式与选型思路
2.1 顺序链路与反射循环,最基础的骨架
第一种模式是顺序执行。流程按步骤走:A 节点做完传给 B,B 做完传给 C。这是最简单也最稳妥的形态,适合任务步骤明确、前后依赖清晰的场景,比如“生成文章大纲 -> 逐节撰写 -> 统一润色”。
但顺序链路有个天然的短板:一旦前面的节点输出质量不行,后面再努力也白搭。所以我在实践里经常会加一层反射循环,也就是让 Agent 对自己的输出先做一轮批判再决定是否修改。这个思路来自 ReAct 和 Reflexion 这两篇论文,但在工程上实现起来并不复杂,本质就是“生成 -> 评估 -> 再生成”的闭环。
def reflection_node(state): draft = state["response"] evaluator_prompt = f""" 请以严格审核员的身份审查以下回复,找出事实性、逻辑性、语气不当之处。 如果发现需要修正的地方,输出修正后的完整版本;如果没有问题,输出原样内容。 回复:{draft} """ revised = llm.invoke(evaluator_prompt) return {"response": revised}这里的要点是:评估和目标生成最好职责分离,不要用同一个 Prompt 既生成又自我评价。实际测试下来,让同一个 Prompt 自我修正容易“越改越偏”,而拆分出独立的评审角色,效果会稳定不少。
2.2 路由分派模式,给流程装上“转向器”
第二种模式是路由。Agent 先做一次意图分类,决定后续进入哪个处理分支。这在客服工单、内容审核、语音导航等场景里非常常见,也是我接触过的项目里落地最频繁的一类。
设计路由模式时,最关键的是“路由标签”必须业务可解释。比如我把工单分成“物流咨询”、“退换货”、“投诉建议”和“其他”,那每个标签后面的处理链路就是明确的。大模型在路由节点需要输出的不是长篇解释,而是一个固定的枚举值,这样后续节点读取判断时不会解析失败。
ROUTER_PROMPT = """ 你是工单分类助手,请判断用户问题属于以下哪个类别: - logistics: 物流、配送、快递相关问题 - refund: 退款、退货、换货问题 - complaint: 投诉、强烈不满、重复反馈 - other: 其他问题 用户输入:{user_input} 只输出一个类别标签,不要输出解释。 """我在工程化时还会多做一个“置信度”字段,让模型同时给出 0 到 1 的置信度。如果置信度低于阈值,就强制转人工,而不是硬着头皮走自动分支。这一步虽然简单,但能显著降低错误路由带来的连锁反应。
2.3 并行扇出与汇聚,提升吞吐量的关键手段
很多任务天然可以拆成多个独立子任务并行处理,比如同时查库存、查物流、查订单详情,或者批量生成不同渠道的营销文案。如果顺序执行,总耗时等于各个子任务耗时之和;改成并行扇出,总耗时约等于最慢子任务的耗时。
在 Agentic Workflow 里实现并行,通常需要定义好“扇出”和“汇聚”两个阶段。扇出阶段把一个大的输入拆成多个小的子任务,汇聚阶段把所有子任务的输出汇总、校验、排序。这两个阶段里的节点可以是固定代码,也可以是 Agent。
一个必须注意的细节:并行子任务之间要尽量避免依赖关系,否则你等半天只会换来一个死锁。还有,汇聚阶段要设计“容错容缺”,个别子任务失败不应该拖垮整个流程。我一般会让失败的子节点返回带 error 标记的结构化结果,汇聚节点发现错误标记就做降级处理,而不是直接抛异常。
2.4 人工在环与多智能体协作模式
第四种模式是人工在环。它适用于高风险场景,比如财务审批、医疗建议、法律文书生成。在这些场景里,即使 Agent 的判断再准,你也不敢完全不设防。人工在环的核心设计不是“简单弹窗让用户确认”,而是把人工节点作为工作流中的一个常规节点:任务走到这里会暂停,等待人在界面里审阅、修改、通过或驳回,然后流程继续往下走。
多智能体协作则是另一种扩展方向。多个 Agent 分别扮演不同角色,互相补充或互相挑战,比如“策划 Agent” 负责出方案,“批评 Agent” 负责找漏洞,“执行 Agent” 负责最终输出。但这里我要泼一盆冷水:多智能体模式看着炫酷,实际复杂度是成倍上升的,通信协议、上下文同步、状态一致性都会成为新的痛点。如果你刚接触 Agentic Workflow,建议先从单一 Agent 加清晰流程开始,等真有必要再上多智能体,否则很容易陷入“三个臭皮匠还是一群诸葛亮”的尴尬。
| 模式 | 核心特征 | 适用场景 | 主要风险 |
|---|---|---|---|
| 顺序链路 + 反射 | 步骤明确,迭代优化 | 内容生成、报告撰写 | 耗时长,成本偏高 |
| 路由分派 | 先分类后处理 | 工单、内容分类 | 标签不准时影响链路 |
| 并行扇出汇聚 | 子任务并行处理 | 批量取数、多路生成 | 结果汇总结算复杂 |
| 人工在环 | 人审关键节点 | 财务、法务、医疗 | 效率受人的响应限制 |
| 多智能体协作 | 多角色协同 | 复杂项目、模拟对抗 | 通信与状态复杂度高 |
3. 核心模块设计,这五个关键技术点决定了效果
3.1 状态管理:工作流的“唯一真相”
Agentic Workflow 里,状态是贯穿全局的核心变量。它就像一个“共享黑板”,每个节点都可以往上面写数据,也可以读取前面的节点留下的数据。如果没有清晰的状态管理,流程一旦变长,你根本不知道当前 Agent 有没有拿到上一步的关键信息。
我习惯用 TypedDict 或 Pydantic BaseModel 定义状态结构,这样每个节点都能知道自己要读什么、写什么,类型检查也能避免低级错误。实际实现里,状态一般分为两类:一类是临时运行状态,比如当前步骤、中间结果;另一类是累积上下文,比如对话历史、已检索的文档段落。前者适合用完就丢,后者需要在流程里持续传递。
3.2 工具层设计:Agent 的能力边界必须清晰
工具层是整个 Agentic Workflow 最容易出问题的地方,同时也决定了 Agent 到底能做什么。设计工具时,一个重要原则是“描述要为模型服务”。大模型没有读过你的源码,它只能靠函数的 name、description 和参数 schema 来理解工具怎么用。如果你的描述写得含糊,模型就会频繁传错参数,或者干脆不会主动调用。
{ "name": "query_order_status", "description": "根据订单号查询订单当前物流状态和预计送达时间。仅用于查询已存在的订单,若订单不存在会返回错误信息。", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "订单号,用户消息中通常以 OD 开头" } }, "required": ["order_id"] } }权限边界也必须在工具层统一管控。我的经验是:只给 Agent 最小必要权限,能只读就不要给写权限,能查单个订单就不要给批量导出接口。尤其是涉及生产环境数据时,一旦 Agent 被提示词注入成功,工具权限过大就会变成真正的安全风险。
3.3 记忆与上下文管理:控制 Token 消耗的命门
Agentic Workflow 里的记忆分为两类:短期记忆和长期记忆。短期记忆就是当前流程里产生的对话历史和中间结果;长期记忆则可以落到向量数据库、KV 存储甚至普通文件里,供后续任务复用。
上下文窗口是有限的,Token 成本是真实的,所以上下文管理几乎是每个项目都要优化的点。我常用的策略有三种:
- 滑动窗口:只保留最近 N 轮对话,更早的内容截断掉。
- 摘要压缩:把早期对话用大模型总结成几句话,需要时再展开。
- 证据抽取:从检索结果只摘取最相关的片段,而不是把整篇文档塞给模型。
这三种方式不是互斥的,我经常会组合使用。比如先做滑动窗口,当窗口快满时触发摘要节点,把当前上下文压到更小的体量再继续。这样既保留了关键信息,又能把 Token 费用控制在合理范围内。
3.4 任务规划与回退机制:给 Agent 上“保险丝”
如果你的流程里允许 Agent 自己拆解子任务,那一定要对规划能力作约束。我见过不少案例,Agent 在复杂任务面前规划出一个十几步的长链路,结果每一步都有小概率出错,最后整个流程的成功率被乘成了一个很低的数字。
比较稳妥的做法是限制最大子任务数,同时设置超时时间。一旦超过阈值,就走回退分支,比如“简化方案”或“转人工”。另一个我很推荐的做法是让 Agent 在规划阶段输出“步骤列表 + 每步预期结果”,然后在执行过程中定期校验“实际结果是否和预期一致”。如果不一致,就触发纠偏逻辑,而不是傻傻地把后面的步骤跑完。这套机制本质上是给 Agent 的自由度加了一个保险丝,让它在失控前有一个强制刹车点。
4. 实战:从零实现一个客服工单自动处理工作流
4.1 需求拆解与流程设计
我挑一个我实际做过的场景来拆解:客服工单自动处理。需求是用户提交一条工单消息,系统要自动理解意图、检索知识库、生成回复建议,并且判断这个回复能否直接发给用户。如果风险高或者置信度低,就转人工处理。
我设计了这样一条流程:
- 意图识别节点:判断工单属于物流、退款、投诉、其他中的哪一类。
- 信息抽取节点:从工单里抽取订单号、商品名、用户诉求等结构化字段。
- 知识库检索节点:根据意图和抽取出的信息,检索内部 FAQ 或业务知识库。
- 回复生成节点:结合检索结果生成一段拟回复草稿。
- 审核判断节点:评估回复是否准确完整,是否可以直接发送,还是需要转人工。
从第 5 步可以看出,审核判断节点是一个典型的 Agent 决策节点,它的输出不是“写好的文案”,而是“放行 / 转人工”这种流程控制信号。这样设计的好处是,高风险场景被兜住,低风险场景实现自动化,整体人工介入量大幅下降。
4.2 基于 LangGraph 的实现骨架
我用 LangGraph 实现这套工作流,因为它的 StateGraph 把状态流转表达得很清楚。下面是核心代码骨架,我加了详细注释,方便你直接对照改。
from typing import TypedDict, Literal from langgraph.graph import StateGraph, END class AgentState(TypedDict): user_input: str intent: str extracted_info: dict retrieved_docs: list draft_reply: str review_result: str final_output: str # 节点1:意图识别 def intent_node(state: AgentState) -> dict: intent = llm_classify(state["user_input"]) # 返回 logistics/refund/complaint/other return {"intent": intent} # 节点2:信息抽取 def extract_node(state: AgentState) -> dict: info = llm_extract(state["user_input"], state["intent"]) return {"extracted_info": info} # 节点3:知识库检索(这里用固定代码更稳) def retrieve_node(state: AgentState) -> dict: docs = search_knowledge_base( intent=state["intent"], info=state["extracted_info"] ) return {"retrieved_docs": docs} # 节点4:生成回复草稿 def generate_reply_node(state: AgentState) -> dict: draft = llm_generate_reply( user_input=state["user_input"], info=state["extracted_info"], docs=state["retrieved_docs"] ) return {"draft_reply": draft} # 节点5:审核判断,决定进入终稿还是转人工 def review_node(state: AgentState) -> dict: result = llm_review(state["draft_reply"], state["user_input"]) # result 示例:{"decision": "direct_send"} 或 {"decision": "human_review"} return {"review_result": result["decision"], "final_output": state["draft_reply"]} def route_after_review(state: AgentState) -> Literal["direct_end", "human_review_end"]: if state["review_result"] == "direct_send": return "direct_end" return "human_review_end" graph = StateGraph(AgentState) graph.add_node("intent", intent_node) graph.add_node("extract", extract_node) graph.add_node("retrieve", retrieve_node) graph.add_node("generate", generate_reply_node) graph.add_node("review", review_node) graph.set_entry_point("intent") graph.add_edge("intent", "extract") graph.add_edge("extract", "retrieve") graph.add_edge("retrieve", "generate") graph.add_edge("generate", "review") graph.add_conditional_edges( "review", route_after_review, { "direct_end": "direct_end", "human_review_end": "human_review_end" } ) app = graph.compile() result = app.invoke({"user_input": "我的订单OD12345为什么三天没更新物流?"}) print(result["final_output"])这套实现的优点在于节点函数是确定性代码和大模型的混合体。意图识别、信息抽取、回复生成、审核判断这四步是大模型决策;知识库检索则属于固定逻辑,用关键词或向量检索实现都行。你只要替换这几个节点的实现,就能把这个骨架套到其他业务场景中。
4.3 评估、压测与迭代
工作流搭好之后,不能直接上线,先用历史工单数据做回归评估。我的做法是准备 200 到 500 条已标注的工单,跑一遍工作流,把每一条的意图识别结果、回复质量、转人工决策都记录下来,和人工标注结果做对比。
关键指标我会重点关注这么四个:
- 意图识别准确率:分类错得越多,后续流程全部在错误链路上跑,影响最大。
- 直接回复率:自动放行的比例越高,说明自动化带来的成本节省越明显。
- 转人工准确率:该转人工的有没有转人工,这是安全底线。
- 平均处理耗时:衡量工作流的性能优化空间。
迭代方式也很朴素:把分错样本挑出来,分析是哪一步出了问题。如果是意图识别错了,就补意图分类的 few-shot 示例;如果是知识库检索没召回正确内容,就去修检索逻辑;如果是生成回复质量差,就优化生成 Prompt。每改一版,把同一批样本重新跑一遍,看有没有引入回归问题。
5. 常见问题与排查技巧实录
5.1 Agent 陷入死循环,流程迟迟不结束
这是我遇到最多的问题,尤其是在允许 Agent 自己决定下一步走向的流程里。表面现象是日志疯狂刷某一个节点的调用记录,底层原因五花八门:条件边没有覆盖所有可能值、Agent 反复重试同一个失败动作、反思循环的跳出条件写错。
排查思路是从日志和状态快照入手。我给 LangGraph 的每个节点都打了结构化日志,记录输入状态和输出状态,出问题时能直接看到是哪一步的状态没变化,从而判断是不是死循环。修复手段通常有三种:
- 在循环节点上加最大步数限制,超过就强制跳出。
- 在条件边函数里增加默认分支,杜绝“没有匹配项”的情况。
- 给 Agent 的决策 Prompt 里明确写“如果发现无法解决,请立即返回当前最佳结果”。
死循环这件事,防比治更重要。设计流程时就要问自己:如果 Agent 连续三次给出同一个动作,我是否允许它继续重复?如果答案是否定的,那就应该尽早加规则拦截。
5.2 工具调用成功率低,参数错乱频发
用大模型调函数,最烦的就是模型“自由发挥”参数。明明工具 schema 里写清楚了订单号是 OD 开头的字符串,模型还是会偶尔给你传个别的格式。这背后可能是指令理解不到位,也可能是用户原文本就含混。
我的处理办法是四管齐下:
- 优化工具描述,把参数格式、枚举值、示例都写进 description 里。
- 在抽取节点先做信息清洗,把用户原文里的关键字段标准化后再传入工具。
- 增加参数校验层,不合法就返回带提示的错误信息,让 Agent 根据提示重新抽取。
- 对高频调用工具增加 one-shot 示例,让模型照着示例格式输出。
特别提一下第二点,信息抽取和工具调用要解耦。不要在工具调用节点里让模型从原始文本里猜参数,而是先让抽取节点把结构化数据提取出来,工具节点只负责校验和执行业务逻辑。这样链路清晰,修起来也容易定位。
5.3 上下文越积越长,Token 成本失控
工作流跑得越久,累积的对话历史和中间结果就越长。尤其是有反射循环和多节点链路的场景,几分钟就能把上下文窗口撑爆,后面每调用一次模型都在烧钱。
我处理这个问题的核心原则是“该忘的果断忘,该留的精准留”。运行状态里的临时数据用完就清,不在状态里保留重复的文档原文。对于需要长期引用的信息,用摘要替换原文,或者只保留检索结果中与当前问题高度相关的段落。此外,还可以在流程中加入 Token 统计节点,每跑一步就估算一下当前上下文的 Token 数量,超过阈值自动触发压缩逻辑。
5.4 结果不稳定,同一输入每次输出都不一样
大模型天生有随机性,配置里还经常默认开启一定的温度,导致同样的输入在不同时刻可能得到不同结果。如果业务对稳定性有要求,比如生成正式文档或审核结论,那就要从两个维度控制:
一个维度是参数层面,把 temperature 调低,必要时直接设为 0,减少随机采样带来的波动。另一个维度是流程层面,用“多次采样 + 投票”或者“结构化输出 + 校验”的方式来收敛结果。前者适合开放性任务,后者适合有标准答案的任务。
我实际使用时会建一个小的回归测试集,每次改动 Prompt 或流程后都跑一遍,确认结果没有出现“这轮改好了那个,那轮改坏了这个”的情况。没有回归测试的 Agentic Workflow 迭代,就是在钢丝上跳舞。
6. 个人积累的几条实战经验
框架不要贪多。LangGraph、AutoGen、CrewAI 这些我都试过,最后还是回到“最小框架 + 自定义节点”的组合。框架只是帮你管理状态和流程,核心价值始终在节点设计里,框架换不换影响没那么大。
流程优先于模型。遇到效果不达预期,先不要急着换更大参数的模型,而是审视流程是不是少了某个判断节点、工具层是不是设计得不清晰、状态是不是传递丢了信息。很多问题用更小的模型加更好的流程就能解决,成本和延迟反而更优。
可观测性是一开始就要搭好的东西。每一步的输入输出、Token 消耗、耗时、工具调用记录,都要能看到。等你上线后遇到线上问题,才发现没有日志可查,那才是真的痛苦。
这套方法的应用范围比想象中广。我做过的场景里,除了客服工单,还有报告自动撰写、代码评审、销售线索跟进、供应链异常排查,核心思路都一样:拆节点、定义状态、设计工具、加决策点、跑回归。你只要把一个场景吃透,后面迁移起来会非常快。