从去年开始，我陆续做了好几个跟 Agent 相关的项目，从最初“给大模型套个 Prompt 就让它自由发挥”，到后来不得不面对多步骤、有条件分支、需要调用多个工具的真实业务场景。很快发现一个尴尬的事实：纯靠 Prompt 驱动的 Agent，在步骤一多之后就变得不可控——它可能反复做同一件事，可能跳过关键步骤，也可能在某个分支里越走越远。于是我开始研究“面向 Agent 的轻量 Workflow 设计”，也就是今天这篇博文想聊的东西：不搞重型引擎，不用复杂的 DSL，只用一个轻量的控制骨架，把 Agent 的决策能力和 Workflow 的确定性结合起来，让 AI 干活儿既有灵活性，又不至于跑偏。这篇文章适合正在做 Agent 应用落地的开发者、技术负责人，以及想从“Demo 级 Agent”走向“可交付产品”的朋友们参考。
1. 先想清楚：Agent 到底需不需要 Workflow
我见过不少团队在这个问题上有分歧。一派认为 Agent 就应该完全自主，你给它目标，它自己拆解任务、自己决定步骤，加 Workflow 反而是束缚。另一派觉得凡是上了 Workflow，Agent 就退化成了传统自动化脚本。
我的看法比较折中：Agent 需要 Workflow，但它需要的是面向 Agent 设计的 Workflow，而不是传统意义上的流程引擎。
1.1 纯 Prompt 驱动的 Agent 在哪些场景会失控
先说我踩过的坑。之前做一个自动生成行业分析报告的项目，最初设计方案是：给 Agent 一个总目标“生成一份完整的行业分析报告”，然后让它自己规划步骤。单看几个简单任务，它表现确实惊艳，能自己列出“收集资料、分析数据、撰写大纲、输出全文”这些步骤。但一旦任务链变长，问题就出来了。
最典型的问题有三个。
第一是重复执行。Agent 会反复调用同一个搜索工具，搜同样的关键词，拿回几乎一样的内容，白白消耗 token 和时间。有一次我盯日志，发现它连续四次搜索了同一个页面。
第二是关键步骤丢失。它是目标驱动的，如果 sub-agent 反馈说“当前信息足够”，主 Agent 可能就直接跳到写报告阶段，把“数据验证”“竞品对比”这些该做的环节跳过去了。你很难在纯 Prompt 层面强制它不跳步。
第三是上下文无限膨胀。每一轮工具调用结果都塞进上下文，几百轮之后 Prompt 越来越长，模型开始忽略早期信息，甚至出现“遗忘”早期约束的情况。这就是为什么后来我下定决心，在设计上引入 Workflow 控制层。
1.2 轻量 Workflow 和传统工作流引擎的本质区别
传统工作流引擎（比如那些面向企业流程审批的）解决的是“确定性流程的自动化”，每个节点做什么、下一步去哪、什么条件下走哪个分支，全是人肉预先定义好的。它追求的是零意外、强约束、全记录。
面向 Agent 的轻量 Workflow 不一样。它不追求把每一步都定义死，而是把“流程的骨架”和“节点的决策”分开：骨架负责控制方向、定义状态、约束边界，节点内部的具体执行策略交给 Agent 自己判断。
我用一个比喻来解释：传统 Workflow 像铁路，轨道铺到哪车就开到哪；面向 Agent 的 Workflow 更像是给车配了一个导航 + 安全围栏。导航告诉你从北京到上海有几条路，每个路口怎么走；安全围栏保证你不偏离路线、不来回兜圈。但道路上的车况、临时切换路线、服务区休息这些事，导航不管，你自己决定。
所以轻量 Workflow 设计的核心，并不是“用什么框架”，而是回答几个问题：
- 流程有哪些必经的阶段？
- 每个阶段由谁来执行（是 Agent 还是确定性代码）？
- 状态和上下文如何流转？
- Agent 在什么范围内可以自由决策？
- 流程跑偏了如何纠正？
把这些想清楚，框架本身很简单——普通的编程语言 + 判断循环就足够了。
2. 轻量 Workflow 的整体设计与关键决策
这个项目的核心设计，我拆成四个维度来讲：执行模型、节点抽象、状态管理和上下文裁剪。这四个维度决定了你的 Workflow 是“能跑”还是“扛得住真实业务”。
2.1 执行模型：从“一个 Agent 干到底”拆成“多阶段接力”
我之前犯过一个设计错误：试图设计一个超级 Agent，让它掌管一切。做出来之后发现，单个 Agent 的上下文窗口根本扛不住，而且它内部的推理过程是一个黑盒，出了问题你只能重跑，没法定位。
后来我改成多阶段接力模型：整个任务拆成几个阶段，每个阶段是一个独立的 Agent 节点或一个确定性函数节点，阶段之间通过结构化的数据传递结果。
比如生成行业分析报告的这个项目，最终拆成五个阶段：
- 需求解析阶段：接收用户原始需求，用 Agent 解析出分析目标、行业范围、报告受众、报告长度。
- 资料收集阶段：一个带工具调用能力的 Agent，根据需求解析结果搜索和采集资料。
- 内容提炼阶段：对采集到的原始内容做结构化提炼，生成摘要、关键数据点、观点列表。
- 报告撰写阶段：基于提炼后的信息，按模板生成报告正文。
- 质量校验阶段：一个专门的“审查者” Agent 检查报告是否覆盖需求、是否有明显事实错误。
这个设计的好处在于：问题定位清晰。如果报告写得烂，问题大概率在第 4 阶段；如果资料不相关，看第 2 阶段的输出就知道。每个阶段的输入输出都有明确 schema，测试起来也很方便——我可以单独 mock 前面阶段的数据，去测后面阶段的 Agent 表现。
你可能会问：阶段拆这么细，是不是又变成传统 Workflow 了？不完全。区别在于，每个阶段内部仍然是 Agent 自主决策的。比如第 2 阶段，Agent 自己决定搜哪些关键词、访问哪些页面、抓取哪些字段。只是说“这个阶段的任务边界”被固定了，它不能越界去做第 4 阶段的事情。
2.2 节点类型：Agent 节点、函数节点、控制节点
轻量 Workflow 里的节点，我归纳出三种类型。
Agent 节点：内部包含 LLM 调用、工具调用、上下文管理。它接收结构化的输入，输出结构化的结果。Agent 节点是流程中最灵活的部分，适合“需要理解、判断、生成”的任务。
函数节点：纯粹的确定性代码，比如数据清洗、格式转换、HTTP 请求、文件读写。这类节点不需要 LLM 参与，直接写函数就行。我之前见过有人连“把字符串转成 JSON”这种操作都要调一次 LLM，完全是浪费。
控制节点：负责判断流程走向，比如条件分支、循环、并行分发。在轻量设计里，控制节点可以是简单的 if-else、for 循环，也可以用 LLM 做路由决策。
这里有个重要的经验：能用函数节点解决的问题，绝不用 Agent 节点。Agent 节点意味着延迟、成本、不确定性。函数节点意味着快、便宜、结果可预测。一个成熟的 Workflow 设计，Agent 节点应该只占 30% 左右，剩下 70% 都是函数节点和控制逻辑。
2.3 状态管理：统一上下文还是局部传递
状态管理是 Workflow 设计里最容易翻车的地方。我见过两种典型反模式：
反模式一：所有节点的输入输出全部塞进一个全局 context 对象，跑完几步之后，context 越来越大，每次传给 LLM 的 token 数量直线上升，成本爆炸。
反模式二：节点之间毫无共享状态，每个节点只接收上一个节点的输出，导致你需要设计特别复杂的参数传递关系，流程一改代码就崩。
我的做法是分层状态管理。全局层面维护一个轻量的持久化信息区，只放那些贯穿整个流程的关键信息（比如任务 ID、用户需求原文、最终输出文件的路径、关键决策记录）。节点之间的数据传递，走显式的 input/output schema，把上一节点的输出作为下一节点的输入。
这样做的好处：全局上下文保持精简，不会无限膨胀；节点之间的数据流清晰，方便单独调试和测试。我甚至可以在某个节点失败后，用之前保存的中间结果做断点重跑。
具体到实现，每个节点定义自己的input_schema和output_schema，引擎负责把上一个节点的输出映射到下一个节点的输入。这样有一个额外好处：每个节点都可以独立测试。我经常在开发期直接写一段测试数据喂给 Agent 节点，观察它的输出是否符合预期，不需要跑完整条流程。
2.4 上下文裁剪：不把历史包袱全丢给每个 Agent
聊到状态就绕不开上下文管理。LLM 的上下文窗口是有限的，就算你用的是大窗口模型，塞太多无关内容也会稀释注意力、降低输出质量、增加成本。
我在上下文裁剪上总结了一套非常实用的策略，核心原则是“给每个节点它恰好需要的那一段，而不是给它全部”。
具体来说，在每个 Agent 节点启动之前，引擎会做一个动作：从全局状态里筛选出“这个节点需要的字段”，组装成一份小型的上下文包。比如需求解析阶段只需要用户原始需求，就没必要把资料采集阶段的原始 HTML 也塞进去；报告撰写阶段只需要提炼后的要点，没必要把几十页的原始网页内容都带上。
操作上，每个 Agent 节点声明自己需要的输入字段，引擎负责从全局状态中摘取并组装。这本质上是按需取用，避免把上一节点的所有输出都无脑传给下一节点。
这还不够。工具调用的历史记录，也要定期精简。Agent 在资料收集阶段可能调用了 30 次搜索和抓取工具，如果 30 次调用历史全保留在上下文里，后续再调用工具时模型会被历史干扰。我的做法是：工具调用的轮次超过一个阈值（比如 15 轮）之后，把最旧的记录折叠成摘要，只保留最近的详细记录。
3. 核心实现：从一个最小可跑的 Workflow 引擎说起
理论说多了容易飘，直接上实操。这一步我做一个极简的“面向 Agent 的轻量 Workflow 引擎骨架”，代码量不大，但把前面设计里说的内容都涵盖进去。
3.1 引擎的核心数据结构与执行逻辑
先定义三个基础数据结构：节点、边、执行上下文。
from dataclasses import dataclass, field from typing import Any, Callable, Optional import json @dataclass class Node: node_id: str node_type: str # "agent" | "function" | "condition" handler: Callable # 节点执行函数 input_schema: dict = field(default_factory=dict) output_schema: dict = field(default_factory=dict) config: dict = field(default_factory=dict) @dataclass class Edge: source: str target: str condition: Optional[Callable] = None # 条件函数，返回 bool @dataclass class WorkflowContext: global_state: dict = field(default_factory=dict) node_results: dict = field(default_factory=dict) current_node: str = ""核心执行逻辑就是一个循环：从入口节点开始，执行当前节点，根据边的条件找下一个节点，直到没有下一个节点为止。
class WorkflowEngine: def __init__(self, nodes: list[Node], edges: list[Edge]): self.nodes = {n.node_id: n for n in nodes} self.edges = edges self.entry_nodes = [n.node_id for n in nodes if n.config.get("is_entry")] def run(self, initial_state: dict) -> dict: ctx = WorkflowContext() ctx.global_state.update(initial_state) current = self.entry_nodes[0] max_steps = 100 # 保险丝，防止死循环 step = 0 while current and step < max_steps: step += 1 ctx.current_node = current node = self.nodes[current] input_data = self._prepare_input(node, ctx) output_data = node.handler(input_data, ctx) ctx.node_results[current] = output_data ctx.global_state.update(output_data) current = self._get_next_node(current, ctx) return ctx.global_state_prepare_input做的事情就是前面说的上下文裁剪，只从global_state里摘取input_schema声明的字段：
def _prepare_input(self, node: Node, ctx: WorkflowContext) -> dict: input_data = {} schema = node.input_schema for key, spec in schema.items(): source = spec.get("source", "global") if source == "global": if key in ctx.global_state: input_data[key] = ctx.global_state[key] elif source == "node": input_data[key] = ctx.node_results.get(spec["node"], {}).get(key) return input_data_get_next_node遍历所有以当前节点为起点的边，执行条件函数，返回第一个满足条件的边对应的下游节点：
def _get_next_node(self, current: str, ctx: WorkflowContext) -> Optional[str]: for edge in self.edges: if edge.source == current: if edge.condition is None or edge.condition(ctx): return edge.target return None这个骨架看起来很简单，但已经能表达链式流程、条件分支这些基础能力了。真正让它“面向 Agent”的地方，是在 handler 的编写上。
3.2 给 Agent 节点定义输入输出契约（Schema）
Agent 节点是整个 Workflow 设计里最需要花心思的地方。它背后是 LLM，模型输出天然是自由文本，不保证结构，所以你要在节点层面做好约束和解析。
我的做法是：每个 Agent 节点配置一个输入 schema 和一个输出 schema。输入 schema 定义它需要哪些字段和对应的类型、描述；输出 schema 定义它返回的 JSON 结构。在 prompt 里明确告诉模型“你必须输出满足以下 JSON Schema 的结果”，然后用代码校验和解析输出。
举个例子，假设我们的资料收集阶段节点，输入是“分析主题”和“报告需求”，输出是“一组带来源链接的资料列表”。
def research_agent_handler(input_data: dict, ctx: WorkflowContext) -> dict: topic = input_data["topic"] requirements = input_data["requirements"] prompt = f"""你是一名行业研究助理。你需要围绕目标主题收集高质量的资料线索。 研究主题: {topic} 研究要求: {requirements} 步骤: 1. 调用 search 工具搜索相关内容。 2. 对搜索结果中值得深入阅读的页面，调用 fetch 工具获取正文。 3. 提炼每份资料的核心观点，并记录来源链接。 完成后，输出 JSON，格式如下: {{ "materials": [ {{ "title": "资料标题", "url": "来源链接", "summary": "200字以内的摘要", "key_points": ["核心观点1", "核心观点2"] }} ], "search_queries": ["实际使用的搜索关键词"] }} 要求: - materials 至少包含 5 条资料。 - 所有 URL 必须来自真实搜索结果，不要编造。 - key_points 从资料正文中提炼，不要凭空发挥。 """ response = call_llm(prompt, tools=[search_tool, fetch_tool]) parsed = parse_and_validate_json(response, default_schema) return parsed这一段代码的核心思想是：把流程控制和模型决策分开。流程控制（比如至少需要 5 条资料、输出哪些字段）由 schema 和 prompt 约束；模型决策（搜什么关键词、点开哪些页面、提炼哪些观点）完全由 Agent 自主完成。
3.3 条件路由：让 Agent 自己决定下一步走哪个分支
这是轻量 Workflow 区别于传统流程引擎最有意思的地方。传统流程的分支条件，是人写死的判断逻辑，比如if age > 18。面向 Agent 的分支条件，可以让模型实时判断应该走哪个分支。
举一个具体场景：在报告生成流程里，做完质量校验之后，如果校验不通过要返工。此时走哪条边不再是简单的阈值判断（比如“错误数 > 3 就返工”），而是让一个审查 Agent 输出结构化结论：{"passed": false, "reasons": [...], "suggested_actions": ["补充某个数据章节"]}。然后控制节点根据这个结论决定是进入“返工修复”分支还是“完成输出”分支。
def quality_check_handler(input_data: dict, ctx: WorkflowContext) -> dict: draft_report = input_data["draft_report"] requirements = input_data["requirements"] prompt = f"""你是一名报告审校专家。请检查以下报告是否完整覆盖了需求。 需求原文: {requirements} 报告正文: {draft_report} 输出 JSON: {{ "passed": true/false, "issues": [ {{"severity": "high|medium|low", "description": "问题描述"}} ], "missing_sections": ["缺失的章节"] }} """ response = call_llm(prompt) return parse_and_validate_json(response, quality_schema) # 定义边和条件 def is_passed(ctx: WorkflowContext) -> bool: qc_result = ctx.node_results.get("quality_check", {}).get("passed", False) return qc_result is True def needs_rework(ctx: WorkflowContext) -> bool: qc_result = ctx.node_results.get("quality_check", {}).get("passed", True) return qc_result is False edges = [ Edge(source="quality_check", target="finalize", condition=is_passed), Edge(source="quality_check", target="rework", condition=needs_rework), ]这样做和传统条件分支最大的区别是：分支的依据不是程序预先定义的指标，而是模型对当前结果的理解和判断。它比硬编码规则更灵活，比完全自由更可控。
3.4 人工确认节点：关键操作不能全交给模型
Agent 应用落地的拦路虎之一就是“信任问题”。模型再强，你也不敢让它全自动执行某些高风险动作，比如发邮件、下单、删数据、对外发布内容。所以设计上一定要有人工确认节点。
轻量实现方式就是：Workflow 走到某个节点时暂停，把“待确认内容”发到指定的审批通道（钉钉/企微/邮件/控制台），等人点了通过或驳回，再根据结果继续流程。这个暂停和恢复机制，在代码里实现并不复杂，核心是一个持久化的状态存储加一个恢复入口。
我用最简单的文件存储做过一版：
def human_approval_handler(input_data: dict, ctx: WorkflowContext) -> dict: pending_item = input_data["pending_item"] # 将待确认事项写到 inbox 文件 / 数据库等待人工处理 approval_ticket = create_approval_ticket(pending_item) # 轮询等待人工结果（简单实现） result = wait_for_approval(approval_ticket.id, timeout=3600, poll_interval=30) return { "approved": result["approved"], "comment": result.get("comment", "") }人工确认节点是 Workflow 设计里“面向真实业务”的关键一环。它不增加多少技术复杂度，但对产品可信度的提升是巨大的。
4. 实操全过程：用 Workflow 搭一个“竞品分析报告生成器”
讲了设计思路和代码骨架，这一节我完整走一遍实操过程，从定义流程到跑通全链路，把前面所有环节串起来。
4.1 业务流程拆解
先说需求：用户输入一个行业和几个竞品名称，系统自动生成一份竞品分析报告，内容包括市场概况、竞品对比、优势劣势分析、行动建议。
按前面说的设计原则，我把它拆成六个节点：
parse_requirement（Agent）：从用户输入的原始文本中提取分析目标、竞品范围、关注的维度。collect_market_data（Agent）：搜索结果获取市场规模、增长趋势、行业动态。collect_competitor_data（Agent）：逐个竞品收集产品信息、定价、用户评价、融资情况。build_comparison_matrix（函数）：将收集到的竞品信息整理成对比表格，供后续分析使用。generate_report（Agent）：基于对比矩阵和市场数据，生成报告全文。quality_check（Agent）：审查报告质量，判断是否需要修改。
前三个节点虽然都是 Agent 节点，但输入输出 schema 完全不同，各自关注的内容边界也不重叠。这就是前面说的“接力模型”的实际落地。
4.2 节点装配与流程定义
用前面定义的 WorkflowEngine，装配这个流程：
nodes = [ Node( node_id="parse_requirement", node_type="agent", handler=parse_requirement_handler, input_schema={ "raw_input": {"source": "global", "type": "string", "required": True} }, output_schema={ "analysis_target": {"type": "string"}, "competitors": {"type": "list"}, "dimensions": {"type": "list"} }, config={"is_entry": True} ), Node( node_id="collect_market_data", node_type="agent", handler=collect_market_data_handler, input_schema={ "analysis_target": {"source": "global", "type": "string"}, "dimensions": {"source": "global", "type": "list"} }, output_schema={ "market_overview": {"type": "string"}, "market_stats": {"type": "list"}, "industry_trends": {"type": "list"} } ), Node( node_id="collect_competitor_data", node_type="agent", handler=collect_competitor_data_handler, input_schema={ "competitors": {"source": "global", "type": "list"}, "dimensions": {"source": "global", "type": "list"} }, output_schema={ "competitor_profiles": {"type": "list"} } ), Node( node_id="build_comparison_matrix", node_type="function", handler=build_comparison_matrix_handler, input_schema={ "competitor_profiles": {"source": "global", "type": "list"} }, output_schema={ "comparison_matrix": {"type": "list"} } ), Node( node_id="generate_report", node_type="agent", handler=generate_report_handler, input_schema={ "comparison_matrix": {"source": "global", "type": "list"}, "market_overview": {"source": "global", "type": "string"}, "analysis_target": {"source": "global", "type": "string"} }, output_schema={ "report": {"type": "string"} } ), Node( node_id="quality_check", node_type="agent", handler=quality_check_handler, input_schema={ "report": {"source": "global", "type": "string"}, "analysis_target": {"source": "global", "type": "string"} }, output_schema={ "passed": {"type": "boolean"}, "issues": {"type": "list"} } ), ] edges = [ Edge(source="parse_requirement", target="collect_market_data"), Edge(source="collect_market_data", target="collect_competitor_data"), Edge(source="collect_competitor_data", target="build_comparison_matrix"), Edge(source="build_comparison_matrix", target="generate_report"), Edge(source="generate_report", target="quality_check"), ]这里我故意简化了：没有加“质量审查不通过就返工”的循环分支。真实场景应该加上的，否则审查节点形同虚设。加一个rework节点，以及从quality_check回到generate_report的边，配合重试次数计数器。
4.3 关键参数设定与成本控制
Flow 能跑起来只是第一步，真实项目里要花心思调整的是参数。我给出几个关键参数的经验值：
单阶段 Agent 的最大工具调用轮次：我设置的是 15 轮。超过这个轮数强制结束本节点，输出当前已收集的结果。限制原因很简单：工具调用轮次太多就意味着 Agent 在低效打转，及时止损比让它继续耗下去划算。
LLM 温度：资料收集和提炼阶段设 0.2，报告撰写阶段设 0.7，质量审查阶段设 0.1。资料收集追求稳定和忠实；报告撰写需要一点生成多样性，避免读出“模板味”；审查环节要严格，温度设最低。
超时控制：单节点超时时间设 120 秒（不包括人工确认节点）。LLM API 偶尔会有慢响应，超过 120 秒直接判定失败并走重试逻辑。重试最多三次，使用指数退避：第一次等 5 秒，第二次 20 秒，第三次 60 秒。
Token 成本预估：以 1000 篇竞品分析报告的量级来算，单次完整流程消耗约 15 万到 25 万 token。其中大头在资料收集阶段（工具调用产生的输入 token）。如果上下文裁剪做得好，可以砍掉 30% 到 40% 的 token 消耗。这也是为什么我一直强调上下文裁剪不只是技术洁癖，它直接影响你的账单。
并发策略：资料收集阶段的竞品数据是相互独立的，我实现了一个简单的并发分发控制节点，把多个竞品的收集任务并行丢给多个 Handler。并行度别设太高，LLM API 会有速率限制，我一般控制在 3 到 5 并发。
4.4 实测效果与观察记录
把这套流程跑在 5 个真实行业场景上（消费电子、企业服务、在线教育、医疗健康、物流），我记录了一些值得分享的观察。
第一，质量审查节点的“挑刺”接受率：第一次跑，5 份报告里有 3 份被审查节点打回。打回的原因集中在“缺少数据支撑”“竞品对比维度不完整”这两类。返工一次之后，报告质量明显上了一个台阶。这说明审查节点不是摆设，它确实能发现生成阶段的问题。
第二，上下文裁剪效果显著：对比相同任务、裁剪前后两次完整运行，裁剪版本（按 input_schema 按需取数据）比未裁剪版本 token 消耗节省了 36%。同时，模型输出的相关性和准确性也更高了，因为喂给它的信息更聚焦。
第三，工具调用轮次限制在真实场景很重要：收集竞品信息的时候，Agent 经常无限深入某个竞品的新闻页面，查得越来越细，但这些都是边际收益很低的动作。加了 15 轮限制后，整个流程的平均耗时有明显下降，同时报告质量没有变差。
5. 常见问题与排障实录
这一节整理我在实际开发中遇到的高频问题，包括报错现象、排查思路和最终解决方式，按速查表的格式给你参考。
5.1 问题速查表
| 症状 | 可能原因 | 排查步骤 | 解决方案 |
|---|---|---|---|
| Agent 节点陷入工具调用死循环 | 节点职责边界不清晰，Prompt 里没有明确的“停止条件” | 查看该节点完整工具调用日志；统计每轮调用是否在重复相似查询 | Prompt 中显式声明“当收集到足够信息时立即停止调用工具，即使还有更多搜索结果”；增加工具调用轮次上限 |
| 节点输出不符合 JSON Schema | Prompt 约束力不足；模型返回了 Markdown 包裹的 JSON | 打开原始响应日志，确认是格式问题还是内容问题 | 在解析层做容错：先尝试直接 json.loads，失败后剥离 ```json 标记再解析；再失败则强制重试一次，重试时在 Prompt 中加入“上次输出格式非法，本次必须输出纯 JSON” |
| 某个阶段的结果明显偏离需求 | 上一节点的输出质量差，导致后续节点带病运行 | 分别查看每个节点的输入和输出，定位“第一个跑偏的节点” | 给关键节点配置 output validation，增加规则校验（比如必须包含不少于 N 条资料）；在 Prompt 里加“回看需求原文”的指令 |
| 全流程跑很久但结果很浅 | 工具调用结果大量是低质量内容，Agent 花了很多轮寻找到有用信息 | 统计搜索工具返回结果的相关性评分 | 优化搜索查询生成逻辑：让 Agent 在调用搜索前先生成一个“搜索意图”声明，帮助它更精准地构造关键词 |
| 流程到中间断了，前面积累的状态全丢 | 没有持久化状态，进程重启后上下文丢失 | 检查引擎是否有状态恢复机制 | 引入轻量持久化层，每个节点执行完成后把结果实时写入本地存储；再加一个断点重跑接口，从任意指定节点开始恢复 |
5.2 案例复盘：一个“返工循环”的排查过程
有一次测试“竞品分析报告生成器”，遇到一个诡异的现象：报告质量审查总是不过，返工一次之后还是不过，再返工还是不过，直到触发最大重试次数，流程失败。一开始我以为审查太严格了。
排查过程是这样的。先看审查节点的输出，发现返回的issues列表里有这么一条：“报告中缺少对竞品 C 的产品定价分析”。但我回头检查报告正文，发现报告里明确写了竞品 C 的定价信息，而且信息源是真实抓取的。
问题出在哪里？出在generate_report节点的 Prompt 里有一句话：“报告必须覆盖所有竞品。”但build_comparison_matrix阶段生成的对比矩阵是按维度组织的，不是按竞品组织的。Agent 在生成报告时，把矩阵中“定价”这一栏的部分内容遗漏了，导致报告里虽然提到了价格，但没有系统性串联所有竞品的定价对比。
根因是节点之间的数据格式约定没有对齐。对比矩阵的输出 schema 和报告生成阶段的输入期望之间存在语义鸿沟。解决方案也很简单：在build_comparison_matrix的输出 schema 里，增加一个coverage_check字段，明确列出这个矩阵覆盖了哪些竞品的哪些维度。报告生成阶段拿到这个检查清单，就知道自己要全覆盖哪些内容。
这个案例给我的教训是：跨节点的数据传递，不仅要定义字段格式，还要定义“完整性要求”。字段格式保证了解析正确，完整性要求保证了业务不漏项。这是轻量 Workflow 设计中的一个高频坑，值得你特别注意。
5.3 独家避坑技巧：三个我反复用的调试手段
第一个手段是节点级重放。给每个节点缓存输入和输出，出问题时可以只重跑某一个节点，而不是全流程重跑。开发效率提升非常多。比如报告写得不好，直接修改generate_report的 Prompt，然后用缓存的comparison_matrix数据重跑这一个节点，看改进效果，五分钟一轮实验，不用等全流程跑完。
第二个手段是输入变异测试。做 Agent 应用，最大的焦虑是“改了 A 节点的 Prompt，会不会导致 B 节点的输入质量变化？”我的做法是准备一组固定的测试输入，每次改了任何节点，就全流程重跑一遍这组测试输入，比较输出关键指标（比如审查通过率、平均 token 消耗、平均耗时）。不做回归测试的改 Prompt 行为，本质上是在裸奔。
第三个手段是日志里记录每个节点收到和输出的原始 JSON。别嫌日志太长，Agent 应用的排查完全依赖这种细粒度日志。模型输出格式出错、字段缺失、数据被截断，这些问题不看原始 JSON 根本定位不了。
6. 再往上走一步：扩展能力和演进方向
这套轻量 Workflow 运行了一段时间之后，我陆续加了一些增强能力，按价值从高到低排列，你可以评估哪些对你的场景有用。
6.1 并行分发与结果合并
竞品分析场景天然适合并行：收集竞品 A、B、C 的数据是互不依赖的。我在引擎里加了一个简单的ParallelNode，它负责把输入拆分给多个并行执行的节点，最后在汇聚点合并结果。
实现不复杂，本质上是用线程池的map操作。关键是要处理好两个地方：一是各分支的上下文隔离，二是结果合并的 schema 定义。我的做法是：并行分支只读共享状态，各自的结果放在独立命名空间，合并节点再把它们汇总成一个统一的列表。
并行带来的收益非常直观：竞品数量从 3 个增加到 8 个时，串行流程的耗时线性增长，而并行流程几乎不增加额外耗时（受限于 API 速率限制，实际会略有增加）。如果你的业务里有大量这种“同构子任务”，并行绝对值得优先做。
6.2 断点续跑与人工修复
真实生产环境里，Agent 流程随时可能因为 API 超时、网络抖动、解析失败等原因中断。断点续跑不是锦上添花，是刚需。
我的实现思路是：引擎在每个节点执行完成后，把WorkflowContext序列化到本地存储。重启后，加载最近的 context，从失败的节点或指定的节点继续。注意，恢复时需要处理“幂等性”问题：如果一个节点已经执行成功，但还没来得及保存状态进程就崩了，恢复后重跑这个节点可能会产生副作用（比如重复发送工具请求）。我的解决方案是：每个节点执行前检查“该节点是否已在 node_results 中”，已存在的直接跳过。
6.3 走向确定性更强的混合架构
最后想说一个演进方向：Workflow 里的节点，不是“全 Agent”或“全函数”二选一。实际落地一段时间后，我开始把一些 Agent 节点改成“Agent + 确定性后处理”。比如资料提炼节点，模型输出原始提炼结果后，加一层确定性代码做去重、排序、时间戳校验。
这样做的原因是：模型适合做“理解”和“生成”，不适合做“精确计算”和“约束校验”。把这两类逻辑分别交给模型和代码，整个系统的稳定性和可预测性都能上一个台阶。我甚至把一些质量检查也做成了双层：先用正则和规则过滤明显的格式问题，再用审查 Agent 做语义层面的质量判断。规则层负责兜底，模型层负责判断质量，各司其职。
这个方向走深之后，轻量 Workflow 会慢慢变成一个“确定性引擎 + Agent 决策节点”的混合体系。它比纯 Agent 稳定，比纯规则灵活，这也是我个人认为 Agent 应用工程化落地时最务实的形态。
最后再分享一个感受：做了一套轻量 Workflow 之后，我发现它对开发者最大的帮助不是“流程能跑了”，而是让整个 Agent 应用变得可测试、可观察、可干预。Agent 应用最大的风险就是黑盒，而 Workflow 恰恰是打开黑盒的那把钥匙。如果你正在做 Agent 落地，我建议不要一开始就上重型框架，先画一张流程图，想想你的 Agent 在哪些节点决策、哪些节点必须有确定性的代码兜底，然后用一个轻量引擎把它跑起来。这个从简到繁的过程，远比一开始引入一个复杂框架要踏实得多。
