企业官网建设流程全解析

做 Agent 落地的同学，最近应该都有感受：单轮 Prompt 玩得再花，也架不住真实业务要拆成好几个步骤来跑。我把设计重心从"把提示词写到极致"挪到了"面向 Agent 的轻量 Workflow 设计"上之后，很多原本又乱又脆的流程，突然变得可以维护、可以排查、可以复现了。

这篇文章不是讲怎么搭一个重量级工作流引擎，而是分享我实际用的这一套轻量设计思路。适合正在做工具调用、多步任务编排、自动化流程的开发者，或者想摆脱"一个 Prompt 干所有事"这种思路的团队成员。我会尽量用大白话拆解，把数据结构、调度逻辑、踩坑记录都摆出来，你照着改就能用。

1. 为什么 Agent 圈开始认真聊 Workflow

1.1 Agent 不是万能串珠，而是需要剧本的演员

最早做 Agent 的时候，大家习惯把所有逻辑塞进一个巨大的 System Prompt：你是个助手，请根据用户需求一步步思考，需要调用工具时先分析，再调用……听起来很聪明，实际一跑就露馅。模型输出不稳定，今天会按三步走，明天非要跳步；工具调用的参数格式偶发错误；用户中途改需求，你都不知道该从哪一步接着执行。

后来我意识到一个朴素的道理：大模型是聪明的执行者，但不是可靠的状态管理器。它擅长的是在给定上下文里做判断、生成内容，而不是记住"现在已经执行到第几步、上一步拿到了什么结果"。把这些状态和步骤管理交给程序，让模型只负责它擅长的局部任务，这就是 Workflow 存在的意义。

所谓"面向 Agent 的轻量 Workflow 设计"，我最朴素的理解是：用代码把任务拆成一个个节点，节点之间按依赖关系执行，每个节点可以是调用大模型、执行函数、做条件判断，甚至停下来等人工确认。Agent 还是那个 Agent，但它的行动路径从"自由发挥"变成了"在剧本框架内发挥"。

1.2 轻量 Workflow 和传统工作流引擎的边界

很多企业里已经有成熟的工作流引擎，比如基于 BPMN 的流程引擎，或者各种云厂商的 workflow 服务。那些东西非常强大，支持复杂的并发分叉、会签、历史版本、权限模型。但拿到 Agent 场景里你会发现问题：太重了。

传统工作流是为"确定性业务流程"设计的，每个节点的行为几乎完全固定：申请、审批、通知。Agent 工作流的节点行为却是概率性的，大模型可能产生意外输出。你需要的是轻量、灵活、能够容忍模型不确定性的编排框架，而不是一上来就是一堆 XML 定义、流程版本库、集群部署。

我在某个模拟项目里甚至见过同事用商用工作流引擎编排 Agent，结果光是把一个三十行的 JSON 流程定义导进去，就在权限配置上折腾了半天。后来我们拆掉重来，自己写了一个只有三个核心类、一百多行的调度器，反而足够支撑全部业务。轻量的好处在于，你真的能看懂每一步发生了什么，出了问题可以直接改代码，而不是去研究引擎的配置文档。

1.3 什么时候该上 Workflow，什么时候不需要

不是所有 Agent 都需要 Workflow。我的判断标准很简单：如果任务固定只有两三个步骤，而且顺序永远不变，直接在主程序里写死就行，没必要抽象。如果任务超过五个步骤、有分支条件、需要人工介入、或者多个 Agent 要配合，那就强烈建议上 Workflow。

还有一种"隐形需求"：复现性。单纯靠 Prompt 驱动的 Agent，跑同一个任务十次可能有十个不同的过程，这对测试和稳定性是灾难。引入轻量 Workflow 之后，你可以把"过程"变成固定的骨架，每次只是在关键节点上让模型做决策，整个运行轨迹是可控的。很多团队最后选择改造，不是觉得功能不够，而是觉得"复盘成本太高了"。

2. 设计核心：先把执行模型定下来

2.1 图模型选型：DAG 还是状态机还是两者混合

动笔写代码前，先想清楚用哪种执行模型。最常见的三种：

• 有向无环图（DAG）：节点是任务，边是依赖关系，一个节点所有依赖完成后才能执行。优点是适合并行、结构清晰；缺点是不好表达循环，不过 Agent 场景里大部分流程天然是 DAG。
• 状态机：把整个流程看作状态的迁移，每个事件触发一次迁移。优点是异常控制强，可以精确表达"什么时候允许重试、什么时候回退到哪个状态"；缺点是流程复杂时状态爆炸，维护累。
• 两者混合：用 DAG 表达整体结构，在部分节点内部用状态机管理重试和人工确认。这是我最推荐的。

实际写的时候，我会把 Workflow 定义为一组节点和一组边，边决定拓扑关系。至于循环，比如"生成内容质量不合格则重做"，我不会在图里画环，而是把它建模成一个特殊节点，内部自己循环，完成后再把出口交给下游。这样既保住了 DAG 的简洁性，又不丢失常见的"循环重试"语义。

2.2 节点怎么描述：从"做什么"到"怎么做"

节点是整个设计的地基，描述方式会直接影响扩展性。我习惯把每个节点拆成两部分：声明（declaration）和实现（implementation）。声明是配置，比如节点 ID、类型、依赖的输入字段、输出字段、参数。实现是真正的执行逻辑，比如一段 Python 函数、一个 LLM 调用的封装、一个写入外部系统的操作。

为什么这样拆？因为想在日志、编排、测试的时候只看声明，不关心实现。我通常用字典来定义节点，大概长这样：

{ "id": "parse_requirement", "type": "llm", "model": "gpt-4o-mini", "prompt_template": "from_template_xxx", "input": ["user_request"], "output": "structured_requirement", "max_retries": 3 }

这种描述方式的好处是，节点之间通过字段名显式依赖，上游输出对应下游输入，完全是数据驱动。后面你要加一个新节点，不需要改动调度器，只要加一份声明和对应的执行函数就行。

如果某个流程特别复杂，声明式配置反而别扭，我会允许节点直接用代码块包裹逻辑，但外部接口仍然统一。原则是：调度器永远只认节点接口，不认内部实现。

2.3 上下文传递：避免把状态塞进模型提示词

新手最爱犯的错，就是把上下文全部塞进下一轮的 Prompt。比如在内容生成流程里，先让 Agent 生成大纲，再把大纲、历史对话、一堆中间结果全拼进提示词里让它写全文。这样做有三个问题：token 成本指数级上升；模型容易被冗余信息带偏；你无法精确控制它到底用了哪些中间结果。

我的做法是设计一个统一的上下文对象，理解为整条流程的"工作台"。每个节点从工作台读取自己声明的输入字段，执行完把结果写回工作台。模型节点的 Prompt 模板里只引用它真正需要的字段，而不是把整个工作台丢进去。

class WorkflowContext: def __init__(self, init_data): self.data = dict(init_data) def get(self, field): return self.data.get(field) def set(self, field, value): self.data[field] = value

这样你审计流程的时候，只要看每个节点的 input / output 声明，就能确认数据链路。有人问，如果某个节点需要访问全局信息怎么办？我一般把全局信息跟流程级变量分开，放进context.meta，让模型节点只读取显式指定的入参，减少意外。

3. 轻量实现的关键细节

3.1 调度内核：用队列还是用递归

调度器是执行模型落地的地方。对于轻量实现，我不建议用递归，因为节点多、嵌套深的时候调用栈不清晰，异常也容易堆叠。推荐的做法是：先对 DAG 做一次拓扑排序，然后维护一个"就绪队列"——每次找出所有依赖都已完成且自身未执行的节点，放进队列依次执行。

如果节点没有依赖，一开始就进队列；每完成一个节点，就检查它的下游节点是否需要被激活。周期重复，直到所有节点完成。稀疏标记方法我写了个 schema 示例，不过核心逻辑都是用队列简单可靠：

from collections import deque def run_workflow(workflow, init_data): ctx = WorkflowContext(init_data) node_map = {n["id"]: n for n in workflow["nodes"]} next_map = build_next_map(workflow["edges"]) dep_map = {n["id"]: set(dep) for n, dep in collect_dependencies(workflow["edges"])} ready = deque([nid for nid, deps in dep_map.items() if not deps]) completed = set() while ready: nid = ready.popleft() node = node_map[nid] result = execute_node(node, ctx) ctx.set(nid, result) completed.add(nid) for downstream in next_map.get(nid, []): dep_map[downstream].discard(nid) if not dep_map[downstream]: ready.append(downstream) return ctx

这是我经验里最稳的调度器骨架。它没有处理优先级、并发、超时，因为轻量设计不追求一次性把复杂度做满，先把主流程跑通再说。优先级问题，在队列里按类型排序即可；并发问题，后面单开章节聊。

3.2 失败重试与分支回退

Agent 工作流里，模型调用失败太常见了。网络超时、JSON 解析失败、内容安全过滤拦截，都可能让一个节点失败。失败不能直接终止整个流程，要有重试；重试也不是简单重跑一遍就完事，还涉及退避、幂等和上下文回退。

我的重试策略分两层：

• 第一层，节点内部重试。比如模型调用连续失败三次，第三次失败前每次等待时间按指数退避，1s、2s、4s。
• 第二层，流程级处理。如果节点重试仍失败，我会判断这个节点是否定义了对失败的"回落方式"。比如：LLM 提取信息失败，可以回落到规则正则；函数调用失败，可以跳到人工节点。

def execute_node(node, ctx): last_exc = None for attempt in range(node.get("max_retries", 1)): try: return invoke_node_implementation(node, ctx) except Exception as e: last_exc = e time.sleep(2 ** attempt) return handle_node_failure(node, ctx, last_exc)

分支回退更有意思。我以前接到一个需求：先让 Agent 做意图判断，如果判断为"咨询"就走问答分支，判断为"下单"就走电商分支，但半路发现信息不足，要退回意图判断节点重新识别。这就需要在上下文里记录当前节点链。轻量方式是在 Workflow 里增加一个特殊边，表示"回退目标"，遇到失败时顺着这条边跳回去。状态机的优势在这里就体现出来了，所以我前面强调混合模型，你在图结构之外保留少量状态信息，处理回退会舒服很多。

3.3 人工审批节点的设计

Agent 越聪明，人在关键环节的确认越重要，尤其涉及发消息、下单、删除数据这种不可逆操作。人工节点我通常设计成三类状态：等待中、通过、拒绝。本质上它把流程挂起，等待外部回调来唤醒。

实现上不需要复杂。我会把节点执行状态持久化到一个存储里，同时生成一个 token 作为审批凭据，推送给人工处理端。处理端调用confirm(token, decision, comment)接口，调度器再重新激活对应的节点。注意，流程挂起期间，工作流占用的资源需要释放，不要让线程阻塞在那里干等，而是用异步回调唤醒。

在设计审批节点时，还要考虑超时：用户一直不审批怎么办。我给每个审批节点配一个timeout_policy，可以是自动通过、自动拒绝、或者转交给替代人。这在大厂工作流引擎里属于常规能力，但轻量实现里很多人会漏掉，导致流程卡死。

3.4 可观测性：日志、追踪、运行快照

最后一块核心设计是可观测性。Agent Workflow 比起传统接口，不确定性很高，没有日志你根本不知道模型为什么做出这个决定。

我会在每个节点执行时记录四样信息：

1. 节点开始时间、结束时间、耗时；
2. 读取的输入字段和值；
3. 节点输出（如果是模型节点，同时记录原始响应和解析后的结构化结果）；
4. 重试次数和失败原因。

运行快照（snapshot）也很有用：每隔一段时间或者每个节点完成时，把当前WorkflowContext的完整内容保存一份。这样一旦流程出错，你可以精准回复现场，而不是靠猜测中间变量是什么。轻量实现里，快照直接写成 JSON 存本地磁盘或者对象存储都行，成本很低。

提示：不要偷懒只打日志不存快照。模型节点的中间输出有时候会被安全过滤或者被下游节点覆盖，没有快照你将无从排查。

4. 一套可直接借鉴的最小实现

4.1 节点类型定义

我实际项目里常用的节点类型不算多，下面这张表是我在多个项目里沉淀下来的一套最小集合：

节点类型不要贪多。每增加一种类型，调试成本都跟着翻倍。先用最少的类型跑通，等确实出现了新需求，再加一种对应类型，并重新审视是否能用现有类型组合实现。

4.2 完整示例：内容生产 Agent

拿一个沾边的内容生产 Agent 举个例子。任务：根据用户给的主题，生成一篇带标题、摘要、结构和正文的文章，正文写完需人工确认，确认后再发布到指定系统。整个流程拆成六个节点：

• 主题解析：llm 节点，把用户主题解析成结构化需求：目标读者、字数、风格标签；
• 资料整理：function 节点，根据结构化需求调用搜索接口，拉回若干参考资料；
• 生成大纲：llm 节点，基于需求+资料，输出大纲数组；
• 大纲确认：human 节点，人工审阅大纲，可以批准或者要求重写；
• 生成正文：llm 节点，输入大纲，输出 Vue 风格的正文内容（当然只是正文文体，不是前端）；
• 发布：function 节点，调用发布接口写入内容库。

这几个节点用 DAG 串起来，非常直观。代码就不全贴了，数据流大概是：

workflow = { "nodes": [ {"id": "topic_parse", "type": "llm", "input": ["topic"], "output": "requirement"}, {"id": "material_collect", "type": "function", "input": ["requirement"], "output": "materials"}, {"id": "outline_generate", "type": "llm", "input": ["requirement", "materials"], "output": "outline"}, {"id": "outline_confirm", "type": "human", "input": ["outline"], "output": "confirm_result"}, {"id": "article_generate", "type": "llm", "input": ["outline", "confirm_result"], "output": "article"}, {"id": "publish", "type": "function", "input": ["article"], "output": "publish_url"} ], "edges": [ ("topic", "topic_parse"), ("topic_parse", "material_collect"), ("material_collect", "outline_generate"), ("outline_generate", "outline_confirm"), ("outline_confirm", "article_generate"), ("article_generate", "publish") ] }

注意，edges的第一条("topic", "topic_parse")我是直接拿外部输入当源节点处理的，这只是个记号，真正实现时我会在调度器里把输入当作已完成的虚拟节点。

4.3 接入真实 Agent 时的上下文约定

接入真实模型时，最容易遇到的问题就是上下文对象与 Prompt 模板如何拼接。我的约定如下：

• 每个 LLM 节点从上下文中读取声明好的字段，把它格式化成 Prompt 模板的变量；
• 模板里只允许引用该节点声明过的输入字段；
• 模型返回结果如果是 JSON，必须走一次结构校验器，校验失败就重试或者走回落逻辑。

def invoke_llm_node(node, ctx): variables = {k: ctx.get(k) for k in node["input"]} prompt = render_template(node["prompt_template"], variables) raw = call_llm(node["model"], prompt, response_format=node.get("response_format")) if node.get("response_format") == "json": parsed = safe_parse_json(raw) if parsed is None: raise ValueError("model returned invalid json") return parsed return raw

很多人说模型返回 JSON 不稳定的问题，我的经验是：先在 Prompt 里要求"只输出 JSON，不要用 markdown 代码块包裹"，然后用safe_parse_json兜底，解析失败时截取大括号部分重试。这条建议能解决大部分问题。

4.4 并发执行的轻量方案

如果某个流程里有两个节点互相没有依赖，比如"同时搜集 A 渠道和 B 渠道的资料"，轻量调度器怎么处理？答案是用线程池。

我一般会在架构里给调度器传入一个executor：

from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=4) def run_workflow_parallel(workflow, init_data): ctx = WorkflowContext(init_data) # ready 队列同一时刻可能包含多个节点 futures = {executor.submit(execute_node, node, ctx): nid for nid, node in get_ready_nodes()} while futures: done, _ = wait(futures, return_when=FIRST_COMPLETED) for f in done: nid = futures.pop(f) try: result = f.result() ctx.set(nid, result) activate_downstreams(nid, ready_queue) except Exception: handle_failure(nid) # 持续提交新的 ready 节点

并发引入后，问题会成倍增加：数据竞态、结果写回顺序、共享上下文污染。我的建议是：并发节点必须在声明里显式标注"parallel": true，并且彼此读写字段不能重叠，在提交时校验一遍。宁可在最开始的时候限制并发度小一点，也不要一上来开 32 线程，后面排查起来头大。

5. 常见问题与排查技巧实录

5.1 循环依赖检测

即使 DAG 是设计目标，人总有手滑的时候。我遇到过一个案例，调试流程时发现节点 A 等 B、B 等 C、C 又等 A，调度器直接死循环了，队列永远处理不完。后来我在流程加载时强制做一次拓扑排序，排序失败直接报错列出环路成员。

def topo_sort(nodes, edges): indeg = {nid: 0 for nid in nodes} adj = {nid: [] for nid in nodes} for src, dst in edges: indeg[dst] += 1 adj[src].append(dst) queue = deque([nid for nid, d in indeg.items() if d == 0]) result = [] while queue: cur = queue.popleft() result.append(cur) for nxt in adj[cur]: indeg[nxt] -= 1 if indeg[nxt] == 0: queue.append(nxt) if len(result) != len(nodes): raise Exception("cycle detected in workflow") return result

这个检查放在工作流注册阶段，启动即报错，不要等到运行时才暴露。

5.2 模型输出不稳定，JSON 解析经常挂

这个问题几乎每个做 Agent 的都碰到过。我的处理分三层：

第一层，Prompt 里限定输出格式，并给 example，而不是只给 schema 描述。第二层，解析器做容错，比如剥离 markdown 代码块标记、自动修复缺失尾部括号，这种能力可以自己写几十行，也可以用第三方解析库。第三层，解析失败按重试机制跑一次，重试时在 Prompt 末尾追加一句提示："你上一次返回的 JSON 格式无法解析，请严格按要求的格式输出 JSON"。

实测下来，三层叠加能把 JSON 解析成功率从不到八成拉到接近百分百。剩下一小部分，就落到人工兜底或者规则兜底。

5.3 上下文越跑越大，token 成本飙高

很多 Agent 工作流做成链式之后，上下文对象会不断累积所有节点的输出。到最后一个节点，它明明只需要一个字段，却要把前面十步的中间结果全带上，白白烧 token。

我一般约定：每个节点执行完后，可以声明"这个节点输出是否长期存活"。一些临时字段可以在流程结束后清理，或者只在工作台里保留最近 N 个字段。LLM 节点的输入必须显式字段引用，不允许"全部注入"这种偷懒写法。

5.4 并发节点的幂等与重复执行

并发 + 失败重试会导致一个节点被重复执行。如果你的 node 实现里有副作用，比如发送通知、写数据库，重复执行就是事故。轻量设计里，我会给有副作用的节点加一个执行 ID（run_id），并且让外部系统支持幂等键。每次执行前检查幂等键，如果已经处理过就直接返回旧结果。

这一步很多初学者会漏掉，觉得调接口而已没什么大不了的。等你真的因为一次网络超时重试给客户发了两遍通知，就会乖乖补上幂等设计了。

6. 几个容易踩的坑，提前帮你排掉

最后一个部分，把我在实际项目里踩过的坑一次性列出来。

• 不要把所有逻辑一股脑塞进 Workflow。有些简单的顺序调用硬套框架反而增加理解成本，只有节点数量、分支复杂度到达临界点再上。
• 不要在 Workflow 里写太重的前置逻辑。解析、校验这类操作尽量在流程入口前完成，让每个节点职责单一。
• Prompt 模板版本也要纳入配置管理。模型节点的输出质量跟模板强相关，我在模板文件里固定了version字段，改动模板时同步更新，方便回溯。
• 人工节点挂起时的时间处理，要注意跨天、夏令时等情况。
• 监控看板一定要画，不然轻量系统跑起来完全黑盒，出问题根本找不到罪魁祸首。

我自己的习惯是，给每个工作流加一个简单的统计页：各节点执行次数、平均耗时、失败次数、重试次数。别看数据简单，很多时候模型某个输出 schema 频繁解析失败，就是靠这些统计先暴露出来的。轻量不代表不要运维，核心是"最小必要"的运维。

最后分享一个小细节：调度器启动前，我会把要跑的流程定义整体做一次 dry-run，也就是用假数据跑一遍，检查节点之间的字段引用是否都存在、模型节点的响应 schema 是否合法。这个步骤跑完之后，真正的流程上线心里会踏实很多。平时迭代新节点，只要 dry-run 过，基本不会出现到了生产环境才发现字段拼错的尴尬。

面向 Agent 的轻量 Workflow 设计，本质不是一件大工程，它更像是给 Agent 画了一条条清晰的路径，然后再给这条路装上护栏。你完全可以从今天开始，用几个类、一个队列、几个字典，搭出第一个可用的版本，然后慢慢加分支、加人工节点、加并发。先把骨架立起来，剩下来的填肉活儿就都简单了。