方糖猫 AI 视频平台：ReAct Agent 任务规划实战

在方糖猫项目中，我们面对的是这样一个问题：用户输入一段文案，系统需要在几分钟内自动完成素材检索、脚本生成、视频渲染和后期处理，输出一条可发布的短视频。这个链路很长、步骤之间依赖复杂、失败场景多。我们最终选择用 ReAct Agent 来编排整个生成流程。这篇文章会还原当时的设计思路、关键实现和踩过的坑。

一、业务场景：为什么不是简单流水线？

短视频自动生成的典型流程大致如下：

flowchart LR
    A["文案输入"] --> B["素材检索"] --> C["故事板生成"] --> D["配音/字幕"] --> E["视频渲染"] --> F["FFmpeg 后处理"] --> G["成片"]

最直觉的做法是用一个硬编码的工作流引擎按顺序执行每个步骤。但我们很快发现几个问题：

1. 步骤之间的依赖不是固定的。比如某些文案不需要 B-roll，某些素材缺失时需要回退到文字卡片，某些场景需要先合成配音再决定画面节奏。
2. 失败恢复成本高。渲染失败后要重试还是跳过？素材检索为空时是终止还是换关键词重搜？每个决策点都需要上下文。
3. 未来要支持更多视频类型。口播、科普、剧情、产品介绍，每种类型的工作流差异很大，硬编码会迅速失控。

所以我们需要一个能够根据输入动态决策、具备工具调用能力、失败后能自主重试或降级的编排机制。ReAct Agent 是其中一个可行方案。

二、ReAct Agent 的核心设计

2.1 任务拆解

我们把整个生成流程拆成 6 个核心步骤，每个步骤对应一个 Agent 任务：

每个任务都是一个 Tool，Agent 通过 Reasoning 决定调用哪个 Tool、以什么参数调用、如何处理返回结果。

2.2 Agent Loop 设计

我们的 Agent 遵循典型的 ReAct 循环：

flowchart LR
    O["观察（Observation）"] --> T["思考（Thought）"] --> A["行动（Action）"] --> O

具体实现中，我把每次循环抽象为三个动作：

1. Plan：根据当前状态生成下一步计划
2. Act：调用 Tool 执行计划
3. Reflect：检查 Tool 输出，决定继续、重试还是终止

一个简化版的状态机如下：

flowchart TD
    Start["Start"] --> AnalyzeIntent["AnalyzeIntent"]
    AnalyzeIntent --> RetrieveMaterials["RetrieveMaterials"]
    RetrieveMaterials -- "素材不足" --> Fallback["FallbackToTextCards"]
    Fallback --> GenerateStoryboard["GenerateStoryboard"]
    RetrieveMaterials --> GenerateStoryboard
    GenerateStoryboard --> SynthesizeSpeech["SynthesizeSpeech"]
    SynthesizeSpeech -- "TTS 失败" --> RetryAudio["Retry / SkipAudio"]
    SynthesizeSpeech --> RenderVideo["RenderVideo"]
    RenderVideo -- "渲染失败" --> RetryRender["Retry / ReduceResolution"]
    RenderVideo --> PostProcess["PostProcess"]
    PostProcess --> EndNode["End"]

三、关键实现

3.1 Tool 注册与统一接口

所有 Tool 都实现统一的 Tool 接口：

from abc import ABC, abstractmethod
from typing import Any, Dict

class Tool(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        pass

    @abstractmethod
    def run(self, params: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
        """
        params: LLM 生成的调用参数
        context: 当前任务上下文，包含前面步骤的输出
        """
        pass

所有 Tool 注册到一个 Registry 中，Agent 根据 LLM 的输出动态调用：

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool:
        return self._tools[name]

这种设计的好处是：新增一种视频类型或新增一个处理步骤，只需要新增一个 Tool，Agent 的编排逻辑不需要改动。

3.2 Agent Loop 核心代码

class ReActAgent:
    def __init__(self, llm_client, tool_registry, max_steps=15):
        self.llm = llm_client
        self.tools = tool_registry
        self.max_steps = max_steps

    def run(self, user_input: str) -> Dict[str, Any]:
        context = {
            "user_input": user_input,
            "history": [],
            "status": "running"
        }

        for step in range(self.max_steps):
            # 1. 让 LLM 根据当前状态决定下一步
            prompt = self._build_prompt(context)
            action = self.llm.generate(prompt)

            # 2. 解析 LLM 输出：Tool 名称 + 参数
            tool_name, params = self._parse_action(action)

            if tool_name == "Finish":
                context["status"] = "completed"
                break

            # 3. 执行 Tool
            tool = self.tools.get(tool_name)
            try:
                result = tool.run(params, context)
                result["success"] = True
            except Exception as e:
                result = {"success": False, "error": str(e)}

            # 4. 更新上下文
            context["history"].append({
                "step": step,
                "tool": tool_name,
                "params": params,
                "result": result
            })

            # 5. 失败时触发重试或降级
            if not result["success"]:
                fallback = self._decide_fallback(tool_name, result)
                if fallback:
                    context["history"].append({"fallback": fallback})

        return context

3.3 Prompt 设计

Prompt 的核心不是让 LLM「自由发挥」，而是把它限定在一个结构化的决策空间里：

你是一名短视频生成助手。当前任务状态如下：

- 用户输入：{user_input}
- 已执行步骤：{history}
- 可用工具：
  1. AnalyzeIntent: 分析文案意图
  2. RetrieveMaterials: 检索素材
  3. GenerateStoryboard: 生成故事板
  4. SynthesizeSpeech: 合成配音
  5. RenderVideo: 渲染视频
  6. PostProcess: 后处理
  7. Finish: 任务完成

请输出下一步操作，格式如下：

Thought: <你的推理过程>
Action: <Tool 名称>
ActionInput: <JSON 格式的参数>

为了让输出更稳定，我们做了几层约束：

1. JSON Schema 约束：要求 LLM 输出固定字段
2. Few-shot 示例：在 Prompt 里放 2-3 个正确示例
3. Pydantic 校验 + 重试：后端严格校验输出格式，失败自动重试

四、踩过的坑

4.1 LLM 输出格式不稳定

最早用纯 Prompt 约束时，LLM 有时输出 Action: GenerateStoryboard，有时输出 下一步：生成故事板，导致解析失败率很高。

解决方案：

• 引入 JSON Schema，明确要求输出 JSON
• 后端用 Pydantic 校验，失败自动重试 3 次
• 重试时 temperature 从 0.7 降到 0.2，让输出更稳定

效果：结构解析成功率从约 75% 提升到 98% 以上。

4.2 任务卡住没有超时机制

早期 Agent 会无限循环，比如某一步失败后 LLM 反复决定重试同一个 Tool。

解决方案：

• 设置 max_steps=15
• 每个 Tool 设置独立超时
• 同一 Tool 连续失败 2 次后强制降级或人工介入

4.3 上下文过长导致成本飙升

随着步骤推进，history 越来越长，LLM Token 消耗快速增长。

解决方案：

• 对 history 做摘要，只保留关键结果
• 每个 Tool 输出只保留必要字段
• 长视频场景下采用「分段生成 + 本地组装」策略

五、当前效果与待优化方向

目前这套 Agent 编排已经支撑方糖猫平台的核心生成链路，能够覆盖常见的口播、科普类视频生成场景。

已验证的收益：

• 新增一种视频类型的支持成本明显降低，从改动整个工作流变成新增一个 Tool
• 失败恢复路径清晰，常见异常无需人工介入
• 渲染失败、素材缺失等场景可以自动降级，任务完成率稳定在较高水平

还在探索的方向：

• 更细粒度的 Agent 评估：不是只看任务是否完成，还要看成片质量
• 多 Agent 协作：把创意策划、素材审核、质量评估拆成独立 Agent
• 成本优化：不同步骤路由到不同模型，复杂推理用大模型，简单调用用小模型

六、什么场景适合 ReAct，什么不适合

适合：

• 步骤之间有复杂依赖，需要动态决策
• 失败场景多，需要自动重试或降级
• 流程经常变化，硬编码维护成本高

不适合：

• 步骤固定、成功率高、性能要求极高的场景，传统工作流更合适
• 对延迟极度敏感的场景，Agent 的多次 LLM 调用会显著增加耗时

七、总结

ReAct Agent 不是银弹，但它确实帮我们解决了一个核心问题：如何把复杂的 AI 视频生成流程，从「硬编码脚本」变成「可编排、可观测、可恢复」的工程系统。

关键经验：

1. Tool 接口要统一，新增能力时不要改动 Agent 核心逻辑
2. Prompt 输出要做 Schema 约束和后端校验，否则线上会不稳定
3. 失败处理、超时、降级必须一开始就想清楚
4. Agent 的评估指标不能只看「跑通」，要看「生成质量」和「成本」

后续我会继续分享 RAG 素材库、多 Worker 分布式渲染、以及 Whisper + Remotion 渲染链路的实践，欢迎关注。

如果你也在做 AI Agent 相关的项目，欢迎在评论区交流。

• AI 工程

• AI Agent
• React
• LLM
• AI 视频生成

方糖猫（大可）

18 年后端架构，现在独立做 AI 产品。这里记录 AI 视频、AI Agent 与工程实践。