面向Agent的轻量Workflow核心设计与工程实践

1. 项目概述与核心需求解析

1.1 这个 Workflow 要给谁用、解决什么问题

先说结论：面向 Agent 的轻量 Workflow，本质上是一套给大模型智能体准备的“最小执行轨道”。它不追求覆盖所有业务流程场景，不做复杂的 BPMN 建模，也不搞分布式调度那一套，就是让 Agent 在完成多步骤任务时，有一个清晰、可控、可回溯的执行路径。

为什么需要这个东西？我自己在实操里频繁撞到一个问题：直接让 Agent 自由发挥完成复杂任务，比如“整理资料并输出一份行业分析报告”，它经常会做一半就迷失方向。要么步骤顺序混乱，先写了结论再去收集数据；要么某个环节失败后完全不知道该回退到哪一步；最要命的是，任务一旦拆分到五步以上，Agent 的上下文就开始打架，前一步的结果在后一步被重复加工，或者干脆被遗忘。这不是模型能力不够，是缺少一个结构化的执行约束。

你需要一个轻量级的 Workflow 层，做的事就三件：第一，明确任务步骤；第二，维护每一步的输入输出；第三，在步骤之间传递上下文。至于步骤内部怎么发挥，Agent 爱怎么想怎么想，那是它的事。这就是“面向 Agent”的核心区别——传统工作流把每一步的动作也固化了，而我们这里只固化流程，不固化动作，动作由 Agent 动态决策。

适合谁来参考？如果你正在做 AI Agent 类应用，无论是自动化脚本、智能客服、个人助理还是内容生产工具，只要遇到“多步骤任务不可控”的问题，这篇文章的思路可以直接帮你落地一套轻量执行框架。我自己是在一个内部工具项目中实践这套设计的，当时目标很简单：让 Agent 能按固定流程完成资料检索、数据抽取、报告生成三个环节，并且每一步都能被人工干预和复审。

1.2 轻量到底轻在哪里

“轻量”这两个字，我在项目里体会很深。市面上成熟的工作流引擎，比如各类 BPM 中间件，功能确实全：流程图编辑器、定时触发、人工审批节点、子流程嵌套、复杂网关路由，一堆能力。但问题是，Agent 任务很多时候根本用不上这些，反而是这些重能力引出了新麻烦。

我举一个实际的反面案例。初期我尝试过用一套完整的工作流框架，配置了一个只有六个节点的流程。结果发现，光是学习它的节点类型和网关语义就花了两天，写配置文件花了半天，跑起来之后，Agent 回调流程引擎的接口还经常因为鉴权配置不对而失败。那套系统的设计假设是“流程是稳定的、参与者是固定的”，可 Agent 场景恰恰相反：流程经常要调整，比如临时加一步数据清洗，或者删掉一个决策分支；参与者是模型，模型的行为又有不确定性。这一冲突导致我用传统工作流引擎的成本远超收益。

所以后来我总结出轻量的三个标准：

1. 启动轻：不需要独立部署服务，可以在你的应用进程内直接运行，一个类、几个函数就能完成核心逻辑。
2. 改动轻：调整流程结构时，改代码配置即可，不需要重新建模、审批、发布。
3. 心智轻：整套概念模型不超过五个，节点、边、上下文、执行器、钩子，掌握这五个概念就能上手写流程。

1.3 设计目标：给 Agent 一条轨道，而不是一堆枷锁

这个项目在设计之初，我给自己定的目标是：轨道约束要足够强，确保流程不跑偏；但单点自由度要足够高，确保 Agent 的推理能力不被扼杀。说白了，就是在“完全自由”和“完全固化”之间找一个务实的平衡点。

具体到设计目标，我拆成了四条，每条都对应实际踩过的坑：

• 确定性优先：同一个流程定义，无论跑多少次，节点的执行顺序必须是确定的。分支判断可以依赖 Agent 的决策，但决策之后走哪条路要明确。
• 状态可观测：任何时刻，你都要能回答“当前执行到哪个节点了”“上下文里积累了哪些数据”。这一点在后面做错误排查时帮了大忙。
• 失败可恢复：单个节点失败不应该导致整个流程崩溃。好的设计是能让失败节点重试，甚至跳过，同时保留失败现场。
• 运行时可干预：流程执行过程中，人可以暂停、调整参数、重新触发某个节点。这在 Agent 场景里尤其重要，因为模型输出质量波动大，人工复核节点必须能插进去。

这套设计目标听起来不算宏大，但每一条在实际编码中都对应着具体的实现取舍，下面几节我会逐个展开。

2. 整体设计与核心抽象

2.1 节点即函数：把 Workflow 的抽象层做薄

我见过不少 Agent 框架，把 Workflow 节点包装得特别复杂，有输入 Schema、输出 Schema、重试策略配置、超时配置、回调钩子，甚至还要配置依赖注入。结果就是写一个节点要写几十行样板代码。我的选择是：节点就是一个异步函数，其余一切皆可选。

在 Python 里，我的节点定义长这样：

async def extract_data(context: WorkflowContext) -> WorkflowContext: # 从上下文中获取输入 raw_text = context.get("raw_text") # 调用 LLM 抽取结构化信息 result = await llm_extract(raw_text) # 把结果写回上下文 context.set("extracted_data", result) return context

你可能注意到，这个函数直接把整个 context 传进来又传出去。是的，我没有设计独立的输入输出参数模型，而是让节点直接读写全局上下文对象。这个选择有得有失：失是节点之间的数据依赖关系不显式，代码可读性会差一点；得是写节点极其快，临时加一个节点只需要写业务逻辑，不用写数据适配代码。

在实际项目中，我还会给节点的函数对象附加两个可选元数据字段，一个是node_name，用于日志和可视化；另一个是retry_count，用于控制失败重试次数。其余如超时、并发控制，都通过统一的执行器配置，而不是分散在每个节点里。这么设计之后，一个节点就从“一个配置实体”简化成了“一个函数加两个装饰器参数”。

2.2 有向无环图还是线性链：根据任务的真实复杂度来选

轻量 Workflow 的拓扑结构，我评估过两种主流方案：线性链（Chain）和有向无环图（DAG）。各有各的适用场景，但我的实践结论是：大部分 Agent 任务用线性链就足够了，DAG 只在少数并行场景才需要。

先说线性链。它的好处是极其直观，执行顺序就是定义顺序，分支逻辑用手写 if 就能解决。比如流程是“先检索后分析”，你写成两步链，第五行代码做完。开发者读代码时，脑子里能生成一条直线，调试时从第一个节点追到最后一个节点，没有任何分叉路径需要跟踪。

但线性链扛不住两类场景：一类是并行执行，比如同时检索两个不同数据库，结果汇总后再进入下一步；另一类是条件分支，比如根据输入类型走不同处理逻辑。这两类如果硬用线性链写，代码会变成一团剪不断理还乱的 if-else。这时候 DAG 就来了。

我实现的是一个极简 DAG，只支持“多入边依赖”和“多出边分支”。节点在注册时需要声明依赖哪些上游节点，执行器用拓扑排序决定执行顺序。核心数据结构是一个邻接表：

class Workflow: def __init__(self): self.nodes = {} self.edges = {} # node_name -> list of downstream node names self.dependencies = {} # node_name -> list of upstream node names

拓扑排序算法用的是经典的 Kahn 算法，每次挑选入度为零的节点执行。

def _topological_sort(self) -> list: indegree = {name: len(deps) for name, deps in self.dependencies.items()} queue = [name for name, deg in indegree.items() if deg == 0] result = [] while queue: node = queue.pop(0) result.append(node) for downstream in self.edges.get(node, []): indegree[downstream] -= 1 if indegree[downstream] == 0: queue.append(downstream) if len(result) != len(self.nodes): raise ValueError("检测到循环依赖，请在流程定义中移除环路") return result

这里有个经验点：即便你只需要线性链，把拓扑结构设计成 DAG 也花不了多少额外成本，但能换来未来的扩展空间。我常用的一句话是：用 DAG 的底层承载线性链的简洁，用不上时假装它是线性链就行。

2.3 上下文对象：Agent 的共享工作台

上下文（Context）是这个设计里最核心的概念，没有之一。它相当于 Agent 在执行过程中的“共享工作台”，所有节点的输入都从这里取，所有节点的产出都往这里写。我把它设计成一个简单的字典包装器，支持 get、set、delete、merge 四个基础操作，外加一个访问历史记录。

为什么强调访问历史？这是我在调试时踩了大坑后加的功能。Agent 节点的输入输出都是自然语言或结构化文本，体积大且内容相似，如果只记录最终上下文，出了问题根本不知道是哪个节点写脏的。加了访问历史之后，每个键的写入时间、写入节点、旧值、新值都有了记录，排查问题效率翻倍。

class WorkflowContext: def __init__(self): self._data = {} self._history = [] def set(self, key, value, node_name="unknown"): old_value = self._data.get(key) self._data[key] = value self._history.append({ "key": key, "node": node_name, "old": old_value, "new": value, "ts": time.time() }) def get(self, key, default=None): return self._data.get(key, default) def merge(self, data: dict, node_name="unknown"): for k, v in data.items(): self.set(k, v, node_name)

命名规范上我也定了一条：节点写入的键名必须带节点名前缀，比如extract_data.result。理由是避免两个节点写同一个键导致隐性覆盖。这个规范在五个节点以内看不出价值，一旦节点超过八个，就非常救命。

2.4 执行引擎：串行靠状态机，并行靠队列

执行器是整个 Workflow 的心脏。我的实现分两层：底层是一个简单的状态机，负责维护节点生命周期；上层是一个调度器，负责从拓扑排序结果中取出可执行节点，分发执行。

节点生命周期我定义了四个状态：pending、running、success、failed。初始时所有节点都是 pending，执行器循环扫描，把“依赖全部成功”的 pending 节点取出，放入 running；节点执行完毕后根据返回结果置为 success 或 failed。全部节点 success 流程结束，任一 failed 触发重试或终止。

async def run(self, context): execution_order = self._topological_sort() for node_name in execution_order: node = self.nodes[node_name] context.set("_current_node", node_name, node_name) try: result = await self._run_with_retry(node, context) if result is False: raise NodeExecutionError(f"节点 {node_name} 返回失败") node.status = "success" except Exception as e: if not self._handle_failure(node, context, e): raise return context

并行调度的实现自然引入了多路复用。当 DAG 中存在多个入度为零的节点时，我用asyncio.gather并发执行它们。这里有个非常关键的性能优化点：并行节点之间的上下文写入冲突。我的解决办法很简单——每个并发节点写自己的键前缀，最后汇总节点再 merge。这个模式避免了锁机制，性能又高，逻辑又清晰。

3. 与 LLM 的集成：让 Agent 动起来

3.1 工具调用与函数注册机制

Workflow 要驱动 Agent，就得让 Agent 具备调用外部工具的能力。我的做法是建立一个工具注册表，把函数、描述、参数 Schema 三者绑定，然后在节点里给 LLM 提供这个注册表，让它自主决策该调用哪个工具。

class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, description, parameters_schema, func): self._tools[name] = { "description": description, "parameters_schema": parameters_schema, "func": func } def list_tools(self): return [{"name": k, "description": v["description"], "parameters_schema": v["parameters_schema"]} for k, v in self._tools.items()]

这里有一个我之前反复踩的坑：给 LLM 的工具描述写得太随意。后来我意识到，大部分工具调用失败都是因为参数不匹配，比如 LLM 传了一个不存在的字段名，或者把字符串传给了整数参数。修复办法是，在register里强制校验函数签名，调用前用 Schema 校验并尝试做一次类型转换。

async def call(self, name, **kwargs): tool = self._tools.get(name) if not tool: raise ToolNotFoundError(f"工具 {name} 不存在") # Schema 校验与类型修正 validated = self._validate_and_coerce( tool["parameters_schema"], kwargs ) return await tool["func"](**validated)

底层原理上，LLM 工具调用就是把函数描述转成 JSON Schema，让模型“假装”理解，再通过生成结构化参数来触发真实函数。这个机制本身并不复杂，但工程上坑很多。比如模型返回的 JSON 里可能夹带 Markdown 代码块标记，或者同时选择了多个工具。我的处理是写一个健壮的解析函数，先剥离代码块标记，再尝试直接解析，最后用正则提取 JSON 片段，按优先级一次一次尝试。

3.2 决策节点设计：让模型在流程岔路口做选择

流程中的一个重要节点类型是决策节点。它不执行具体业务动作，而是根据上下文内容判断下一步该走哪个分支。实现思路是：提前定义好分支名称和每个分支的选项说明，让 LLM 从候选里选择一个，再把它映射到具体的下游节点。

async def make_decision(context, question, options): prompt = f""" 根据当前上下文，回答以下问题：{question} 选项： {chr(10).join(f"- {opt}" for opt in options)} 只输出选项本身，不要输出其他内容。 """ response = await llm_complete(prompt, context.get("_current_window", "")) cleaned = response.strip().strip("`") for opt in options: if cleaned == opt or cleaned.startswith(opt): return opt # 兜底：匹配失败，走第一个选项 logger.warning(f"决策解析失败，输出了: {cleaned}，回退到首个选项") return options[0]

这个兜底逻辑不是偷懒，而是经过实践后的务实选择。LLM 在开放问题时返回格式错误的概率不低，尤其当上下文很长、模型注意力分布不稳定的时候。与其让整个流程崩溃，不如让它回退到默认分支，至少保证流程可以继续跑，同时把异常记录到日志里提醒人工审查。

这里有一个值得强调的设计原则：决策节点不要做太细粒度判断。比如让模型决定“温度参数取 0.1 还是 0.2”就是浪费；适合交给模型判断的是“这个用户输入是咨询还是投诉”“这一步的结果是否满足质量标准”这类宏观方向性问题。细粒度的判断应该用规则，大方向的判断才用模型。

3.3 结果校验与自省回路

Agent 流程最容易被忽视的环节就是输出质量校验。模型生成的文本大概率结构完整、言辞流畅，但这不等于内容正确。为了建立基本的质量防线，我实现了两种校验器：

一是规则校验器，用来检查硬性条件，比如字段是否缺失、长度是否超出限制、是否包含敏感词。这类校验机快准狠，成本为零，适合前置过滤。

二是模型校验器，用来检查软性条件，比如“抽取的金额字段与原文是否一致”“总结是否覆盖了所有要点”。这类校验基于模型自省（self-reflection），用一次额外的 LLM 调用做二次确认。

async def verify_by_llm(context, result, requirements): prompt = f""" 请验证以下结果是否满足所有要求： 要求：{requirements} 结果：{json.dumps(result, ensure_ascii=False)} 请只回答 PASS 或 FAIL。 """ verdict = await llm_complete(prompt) if "PASS" in verdict.upper(): return True, "PASS" return False, verdict

校验失败后走自省回路：把失败原因打包发给一个修复节点，让 LLM 根据原因修正上一次的结果，修正完再回到校验节点重新验证。这就是一个典型的“执行-校验-修复”循环，也是 Agent 流程从“线性跑一遍”升级到“质量自适应”的关键一步。实际的循环上限我配置成三次，超过三次直接置为失败，防止模型陷入自说自话的死循环。

4. 实操阶段：一个可落地的 Agent 流水线

4.1 场景设定：从客户留言到结构化工单

光讲抽象设计不够，我拿一个真实改造过的场景做全程演示。背景是一个内部服务 Team，每天收到大量非结构化的客户留言，历史做法是人工阅读、分类、填工单系统。我们做的工作是：把这段流程交给 Agent Workflow 自动跑，人工只负责最终确认。

流程共五步，设计如下：

1. register_ticket：接收原始留言文本，生成工单号，写入上下文。
2. classify_intent：让 LLM 判断留言类型：咨询、故障、投诉、建议。
3. extract_entities：根据类型抽取关键实体。故障类抽设备型号和错误码，投诉类抽涉及部门和问题描述。
4. priority_assess：根据抽取结果和紧急度关键词，评估优先级 P0/P1/P2。
5. human_confirm：把整理好的结构化工单交给人工确认，人工可一键通过或退回修改。

这正是线性链的典型用法。每一步的依赖关系简单明确，不需要 DAG 并行，甚至不需要条件分支。如果按传统工作流引擎的建模思路，这五步要做成五个服务、配五个接口、画一张流程图，而在我这套轻量设计里，就是五个函数加一个线性链注册。

4.2 流程编排代码详解

下面这段代码呈现了完整可运行的流程定义，省去了真正的 LLM 调用实现，用模拟函数替代，便于你聚焦在编排逻辑上。

import asyncio from typing import Optional class LLMClient: async def complete(self, prompt: str) -> str: # 实际场景替换为具体模型调用，注意设置超时与重试 await asyncio.sleep(0.5) return "fault" class WorkflowContext: def __init__(self): self._data = {} self._history = [] def set(self, key, value, node_name="unknown"): old = self._data.get(key) self._data[key] = value self._history.append({"key": key, "node": node_name, "old": old, "new": value}) def get(self, key, default=None): return self._data.get(key, default) class WorkflowNode: def __init__(self, name, fn, retries=2): self.name = name self.fn = fn self.retries = retries self.status = "pending" class Workflow: def __init__(self): self.nodes = {} self.sequence = [] def add_node(self, node): self.nodes[node.name] = node self.sequence.append(node.name) async def run(self, context): for name in self.sequence: node = self.nodes[name] context.set("_current_node", name, name) last_error = None for attempt in range(node.retries + 1): try: result = await node.fn(context) if result is False: raise RuntimeError("node returned False") node.status = "success" break except Exception as e: last_error = e logger.warning(f"{name} 第 {attempt+1} 次失败：{e}") if attempt == node.retries: node.status = "failed" raise context.set(f"{name}.node_status", node.status, name) return context

节点函数的基本样例如下。实际项目中，每个节点内部都会调用 LLM 或外部工具，但是节点的输入输出约定保持统一：

async def classify_intent(ctx): raw_text = ctx.get("raw_text") prompt = f"将以下留言分类为：咨询、故障、投诉、建议。只输出一个词。\\n留言：{raw_text}" intent = await llm.complete(prompt) ctx.set("intent", intent.strip(), "classify_intent") return True

如果你把上面这段跑一遍，再打印context._data，会看到每一步的产出都牢牢地存在上下文对象里。这个就是我们唯一定义的数据协议：上下文对象就是节点间通信的全部。

4.3 人工确认节点与非阻塞干预

第五步 human_confirm 比较特殊，它不能自动通过，必须等人工响应。在设计上我用了回调与轮询两种方式：回调适合嵌在已有的 IM 通知体系里，轮询适合作为独立命令行工具运行。我们在内部工具里做得更简单——这个节点会发送一条企微通知给值班人，点“通过”就调用确认接口，点“退回”就带上原因并触发一个重做节点。

这个设计表面上只是加了一个节点，实际上改变了整个流程的可靠性模型。因为有人工兜底，前四步的模型错误不会直接变成线上的坏数据，而是被人在最后一步截住。我强烈建议所有 Agent 流水线，在早期阶段不要追求全自动，而是在关键结果落地前保留一道人工确认。

---

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

5.1 上下文越走越大，把模型“撑”坏了

这是我遇到最多的性能问题。Agent 流程跑多轮之后，上下文里堆积了大量中间结果，尤其是一次性把原始文本、抽取结果、修复版本、校验历史全塞进去，LLM 的输入窗口迟早爆掉。表现是突然有一轮响应变慢，然后报请求过大错误。

排查方法：用WorkflowContext._history做一次统计，找出哪个键占了最大空间。修复方案有两个：一是节点写回时只保留必要字段，二是设计一个“上下文压缩节点”，定期把旧的中间结果做一次摘要，替代原始内容继续往下传。我很多流程里都会加一个summarize_context节点，跑在长文本处理步骤之前，这样既保留关键信息又控制体积。

5.2 LLM 输出格式五花八门，解析稳定不了

模型输出的格式波动是 Agent 工程里最搞心态的问题，没有之一。同一个 Prompt 这次老老实实输出 “故障”，下次可能以 “故障。原因如下：...” 开头。我在决策节点和抽取节点上都栽过跟头。

我的三层兜底策略，按优先级排序：

1. 字符串精确匹配：尝试用完整句对齐候选选项。
2. 包含匹配：检查候选选项是否出现在输出文本中。
3. 正则提取：用re.search提取特定模式，比如引号内的内容。

如果三层都失败，就记录原始输出到日志，然后返回默认值。这套策略不完美，但实测把决策成功率从 90% 提升到了 98.5%。剩下的 1.5%，交给人工兜底。

5.3 重试机制反而变成了故障放大器

重试是双刃剑。早期我天真地给所有节点配置了 3 次重试，结果一个模型调用超时，每个节点都重试三次，整个流程跑了四倍时间，还把外部 API 的限流打满了。后来的修正方案是：区分错误类型，超时类错误自动重试，业务逻辑错误不重试，立刻终止流程。

class TransientError(Exception): pass class PermanentError(Exception): pass async def _run_with_retry(self, node, context): for attempt in range(node.retries + 1): try: return await node.fn(context) except TransientError: if attempt < node.retries: await asyncio.sleep(2 ** attempt) # 指数退避 continue raise except PermanentError: raise

5.4 状态不同步：并行节点把数据写乱了

并行分支的上下文写入冲突是最隐蔽的 bug。两个节点同时写同一个键，后完成的那一方覆盖先完成的，而你不知道哪次会后完成。前面我提过用键前缀隔离，这里展开说一下。假设有retrieve_a和retrieve_b两个并行检索节点，它们分别写retrieve_a.result和retrieve_b.result，汇总节点读取这两个键做融合。这样即使两个节点执行顺序不稳定，也不会互相污染数据。关键是汇总节点的实现要容错，比如ctx.get(f"{node}.result")时万一某个键不存在，要有默认值。

5.5 死循环：自省回路失控

自省修复环节最常见的问题是陷入死循环。模型每次 “修复” 后的结果跟上一版差不多，但校验器总是不通过，于是一次又一次地调用，直到打爆 token 配额。现在我的修复循环做了三重保险：最大重试次数（3 次）、时间戳死线（从进入修复环开始算，超过 60 秒退出）、以及修复内容差异检测（如果两次修复结果的相似度超过 90%，直接判定修复无效）。

6. 再进一步：扩展性、持久化与可观测性

6.1 持久化与回放：让 Agent 流程可复盘

Agent 流程是状态型系统，一旦开始执行，它的中间状态就应该能落盘。我用了最朴素也最实用的方案：每次节点执行完成后，把整个上下文和节点状态序列化到硬盘或数据库里。

def save_snapshot(context, snapshot_path): snapshot = { "data": context._data, "history": context._history, "timestamp": time.time() } with open(snapshot_path, "w", encoding="utf-8") as f: json.dump(snapshot, f, ensure_ascii=False, indent=2)

回放功能就是读取这个快照，把执行过程在日志里重新打印一遍。这个能力在复盘一次失败的 Agent 任务时特别有用，你能清清楚楚看到模型在哪个节点做了错误判断，上下文到底在哪一步被污染。没有这个能力，排一次错平均要半小时以上；有了它，五到十分钟就能定位问题。

6.2 日志与追踪：不要小看一条结构化日志

早期我的日志就是随手 print，后来发现排查 Agent 问题完全不够用。我改进后的日志格式是 JSON 化的，每条日志至少带上节点名、执行编号、耗时、状态、上下文快照的容量。这样无论是 grep 还是接入日志分析平台，都能快速筛选出异常点。比如我常做的一个操作：搜索"status":"failed"，然后按时间线看它前后的日志，就能还原整个故障现场。

6.3 配置化：把流程定义从代码中剥离

节点是函数，流程定义是数据，这两者分开之后，就有了配置化的空间。我的流程定义存在 YAML 文件里，代码负责把节点函数按名字注册到执行器。这样一来，调整流程顺序、增减节点，只需修改 YAML 后重启应用，不需要改代码、走发布流程。对于 Agent 流程这种迭代频率极高的场景，这个特性重要程度排在前面，能省掉大量无意义的开发排期。

workflow: name: ticket_pipeline nodes: - name: register_ticket module: nodes.ticket_ops function: register_ticket - name: classify_intent module: nodes.intent function: classify_intent - name: human_confirm module: nodes.human function: human_confirm retries: 0

6.4 不要在没想清楚时引入消息队列和外部存储

这是最后想提醒的一点。很多开发者一听到 Workflow，就联想到消息队列、分布式任务、独立存储，觉得一套流程引擎天生就该这样。但面向 Agent 的轻量 Workflow，最大的价值恰恰在于简单，在于跑在应用进程内，在于一个字典搞定上下文。当你的流程超过 20 个节点、需要跨服务调用、或者有高并发要求时，再考虑迁移到消息队列也不迟，到时候你的流程代码因为节点和上下文都做了合理抽象，迁移成本也会低很多。过早引入分布式组件，只会让项目的上手门槛和运维成本同时增加，却解决不了 Agent 流程当前真正的问题——可控性。先把可控性做好，后续再谈规模。