1. 项目概述：当Claude Code遇上工作流自动化

最近在折腾AI工作流自动化的时候，我发现了一个挺有意思的组合：用Claude Code来驱动一个叫Ultraplan的智能体编排框架，目标是实现类似“应用商店”的自动化部署与管理。这听起来有点抽象，简单来说，就是让AI不仅能写代码，还能自己规划、执行一系列复杂的任务，并且能像管理手机App一样，自动化地安装、配置和运行各种AI应用或工具链。

这个想法的核心，源于一个很实际的痛点。现在AI工具很多，每个都能干点事，比如有的擅长数据分析，有的能生成文案，还有的可以处理图片。但要把它们串联起来，完成一个从数据输入到报告输出的完整流程，往往需要人工在中间反复切换、复制粘贴、触发执行，效率低下还容易出错。Ultraplan这类框架的出现，就是为了解决这个“最后一公里”的问题——它像一个智能的“总调度”，能把不同的AI能力（智能体）组织起来，按需协作。而Claude Code，作为Anthropic推出的代码生成模型，其强大的上下文理解、代码生成和逻辑推理能力，恰好能为这个“总调度”注入灵魂，让它不仅能执行预设流程，还能动态理解任务、规划步骤、甚至自我修正。

所以，这个项目本质上是在探索“AI驱动AI”的下一阶段：如何用一个更通用的AI（Claude Code）来编排和管理一系列专用AI工具（智能体），并最终实现一个可以自动发现、部署、运行AI应用的平台。这非常适合那些需要处理重复性、多步骤知识工作的团队或个人，比如内容创作团队、数据分析师、软件开发者，或者任何想把自己从繁琐的流程中解放出来的人。接下来，我就详细拆解一下我是如何设计并实现这个系统的。

2. 核心架构与设计思路拆解

2.1 为什么是Claude Code + Ultraplan？

选择这个技术栈，是经过一番权衡的。市面上智能体（Agent）编排框架不少，比如LangChain、AutoGen、CrewAI等，它们各有侧重。Ultraplan吸引我的地方在于它的设计哲学：极度强调规划的清晰性和可执行性。它不像有些框架那样试图用一个“超级智能体”包办一切，而是明确将“规划”（Plan）和“执行”（Execution）分离。规划阶段，由一个“规划者”（Planner）根据目标分解出具体的、原子化的任务列表；执行阶段，由不同的“执行者”（Executor）去逐个完成这些任务。这种架构非常清晰，也更容易调试和扩展。

而Claude Code，正是扮演这个“规划者”角色的绝佳人选。传统的规划者可能基于规则或简单的决策树，但面对复杂、开放式的任务时（比如“为我创建一个每周市场趋势分析简报应用”），就显得力不从心。Claude Code的优势在于：

1. 强大的自然语言理解 ：它能准确理解用户用自然语言描述的、模糊的初始需求。
2. 出色的代码与逻辑生成能力 ：规划的本质是生成一系列可执行的指令或代码片段。Claude Code能生成结构清晰、逻辑严谨的步骤描述，甚至直接生成调用特定API的代码。
3. 长上下文与连贯性 ：它能在整个会话中保持上下文，根据执行结果动态调整后续计划，处理规划-执行-观察-再规划的循环（ReAct模式）。

简单来说，Ultraplan提供了稳定、可靠的编排“骨架”和“关节”，而Claude Code提供了进行复杂思考、做出智能决策的“大脑”。两者的结合，让整个系统既灵活又可靠。

2.2 “应用商店自动化”的核心构想

“App Store Automation”是这个项目的远景目标，也是设计时的核心指导思想。我们不想只是做一个一次性的、固定的工作流。我们希望构建一个系统，使得新的AI应用（工作流）能够像手机安装App一样被方便地添加和管理。这包含了几个层次：

1. 应用描述与发现 ：每个AI应用（或工作流）需要一个标准化的描述文件（比如一个YAML或JSON文件），定义其名称、功能、输入参数、输出结果、以及所需的智能体或工具。
2. 自动化部署 ：当用户说“我想用那个社交媒体内容生成器”时，系统能根据描述文件，自动配置好所需的环境、智能体、以及它们之间的连接关系。
3. 动态编排与执行 ：部署后，用户只需输入目标（如“为我的新产品写三条推文”），Claude Code作为规划者，会读取该应用的描述，理解其能力范围，然后生成具体的执行计划，并指挥Ultraplan的各个执行者去完成。
4. 生命周期管理 ：包括应用的更新、暂停、卸载等。

这个构想将系统的复杂度从“如何编码一个特定工作流”提升到了“如何描述和管理任意工作流”，对规划者的通用性和框架的扩展性提出了更高要求，也是Claude Code大显身手的地方。

3. 系统实现与核心模块解析

3.1 环境搭建与基础配置

首先，我们需要搭建基础环境。项目基于Python，因此需要一个干净的虚拟环境。

# 创建并激活虚拟环境
python -m venv ultraplan_env
source ultraplan_env/bin/activate  # Linux/macOS
# 或者 ultraplan_env\Scripts\activate  # Windows

# 安装核心依赖
pip install ultraplan  # 假设Ultraplan已发布到PyPI，或从GitHub安装
pip install anthropic  # Claude API官方库
pip install python-dotenv  # 用于管理API密钥等环境变量

接下来是关键的配置文件 .env ，用于安全地存储敏感信息：

# .env 文件
ANTHROPIC_API_KEY=your_claude_api_key_here
# 可以添加其他服务的API密钥，如OpenAI, Serper (搜索), 等
LOG_LEVEL=INFO
PLAN_CACHE_DIR=./cache/plans

注意 ：API密钥是最高机密，务必通过环境变量或 .env 文件管理，绝对不要硬编码在源码中。 .env 文件也应加入 .gitignore 。

3.2 定义“AI应用”的元数据规范

为了实现“应用商店”的概念，我们首先要定义什么是“AI应用”。我设计了一个简单的YAML规范来描述一个应用：

# apps/social_media_generator.yaml
name: "社交媒体内容生成器"
version: "1.0.0"
description: "根据产品描述和目标受众，生成适用于不同平台（Twitter, LinkedIn）的推广文案。"
author: "Your Name"
inputs:
  - name: "product_description"
    type: "string"
    description: "产品的详细描述"
    required: true
  - name: "target_audience"
    type: "string"
    description: "目标受众特征"
    required: true
  - name: "platform"
    type: "enum"
    options: ["twitter", "linkedin"]
    default: "twitter"
outputs:
  - name: "generated_post"
    type: "string"
    description: "生成的文案内容"
capabilities:
  - "text_generation"
  - "audience_analysis"
agents_required:
  - "copywriter_agent"
  - "tone_adjuster_agent"
workflow_steps: |
  1. 分析产品描述和目标受众。
  2. 根据平台特性（如Twitter的简洁、LinkedIn的专业）确定文案风格。
  3. 生成初稿文案。
  4. 根据品牌语调进行优化调整。
  5. 输出最终文案。

这个描述文件不包含具体的执行逻辑，只声明了应用的“能力”和“需求”。具体的执行逻辑，将由Claude Code在运行时，结合这些描述来动态生成。这种“声明式”而非“命令式”的设计，是实现灵活性的关键。

3.3 构建基于Claude Code的智能规划器（Planner）

这是整个系统的“大脑”。我们需要创建一个类，它负责与Claude Code对话，将用户需求和应用描述转化为Ultraplan可执行的计划。

import anthropic
import yaml
import json
import os
from typing import Dict, Any, List
from dataclasses import dataclass
from ultraplan import Plan, Task  # 假设Ultraplan有这样的数据结构

@dataclass
class AppManifest:
    """封装应用描述文件的数据类"""
    name: str
    description: str
    inputs: List[Dict]
    outputs: List[Dict]
    capabilities: List[str]
    agents_required: List[str]

class ClaudeCodePlanner:
    def __init__(self, api_key: str, model: str = "claude-3-5-sonnet-20241022"):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = model
        # 系统提示词（System Prompt）是引导Claude行为的关键
        self.system_prompt = """你是一个高级工作流规划专家。你的任务是根据用户的需求和一个AI应用的描述，生成一个详细、可执行的工作流计划。
        计划必须由一系列具体的、原子化的任务组成。每个任务应该清晰描述要做什么、由哪个执行者（Agent）负责、需要什么输入、以及产生什么输出。
        请以JSON格式输出计划，结构如下：
        {
          "plan_name": "计划名称",
          "tasks": [
            {
              "id": "任务唯一ID，如task_1",
              "description": "对任务清晰、无歧义的描述",
              "agent": "负责此任务的智能体名称，必须与App描述中`agents_required`里的名称对应",
              "inputs": {"input_key": "value or reference to previous task output"},
              "expected_output": "期望输出的描述"
            }
          ]
        }
        确保任务之间有清晰的依赖关系，后一个任务的输入可以引用前一个任务的输出（使用类似`{{task_1.output}}`的模板语法）。
        """

    def load_app_manifest(self, app_yaml_path: str) -> AppManifest:
        """加载并解析应用描述文件"""
        with open(app_yaml_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
        return AppManifest(**data)

    def generate_plan(self, user_request: str, app_manifest: AppManifest) -> Dict[str, Any]:
        """核心方法：调用Claude Code生成计划"""
        # 构建给Claude的对话消息
        user_message = f"""
        用户请求：{user_request}

        目标AI应用描述：
        名称：{app_manifest.name}
        功能：{app_manifest.description}
        所需能力：{', '.join(app_manifest.capabilities)}
        可用智能体：{', '.join(app_manifest.agents_required)}
        输入参数：{json.dumps(app_manifest.inputs, indent=2, ensure_ascii=False)}
        期望输出：{json.dumps(app_manifest.outputs, indent=2, ensure_ascii=False)}

        请根据以上信息，生成一个可执行的工作流计划。
        """

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=4000,
                system=self.system_prompt,
                messages=[{"role": "user", "content": user_message}]
            )
            # 解析Claude返回的JSON
            plan_json = json.loads(response.content[0].text)
            return plan_json
        except json.JSONDecodeError as e:
            print(f"Claude返回的不是有效JSON: {e}")
            print(f"原始返回: {response.content[0].text}")
            # 可以在这里添加重试或错误处理逻辑，例如让Claude重新生成
            raise
        except Exception as e:
            print(f"调用Claude API失败: {e}")
            raise

    def convert_to_ultraplan(self, claude_plan: Dict[str, Any]) -> Plan:
        """将Claude生成的计划转换为Ultraplan的Plan对象"""
        plan = Plan(name=claude_plan.get("plan_name", "Unnamed Plan"))
        for task_data in claude_plan["tasks"]:
            task = Task(
                id=task_data["id"],
                description=task_data["description"],
                # 这里需要根据agent名称映射到具体的执行器实例，后面会讲
                executor=task_data["agent"],
                inputs=task_data.get("inputs", {}),
                expected_output=task_data.get("expected_output", "")
            )
            plan.add_task(task)
        # 这里可以添加任务依赖关系解析逻辑，如果Claude在描述中隐含了依赖
        # 例如，通过分析inputs中是否包含对前序任务的引用
        return plan

这个规划器的核心是 generate_plan 方法。它把用户请求和应用描述拼接成一个详细的提示（Prompt），发送给Claude Code。Claude Code在系统提示词的约束下，会输出一个结构化的JSON计划。 系统提示词的质量直接决定了规划的成功率 ，需要精心设计，明确输出格式、任务粒度和依赖关系的表达方式。

实操心得 ：在调试初期，Claude可能会返回格式不正确的JSON，或者在任务描述上过于模糊。我的经验是，在系统提示词中提供更具体的例子，并明确要求“原子化”（即一个任务只做一件事），能显著提高生成计划的质量。另外，对Claude的返回结果一定要做健壮的异常处理（ try...except ），并考虑加入重试机制。

3.4 实现智能体（Agent）执行器与注册中心

规划器生成了计划，指明每个任务由哪个“Agent”执行。接下来，我们需要实现这些Agent，并建立一个注册中心来管理它们。

首先，定义一个基础的Agent接口：

from abc import ABC, abstractmethod

class BaseAgent(ABC):
    """所有智能体执行器的基类"""
    def __init__(self, name: str):
        self.name = name

    @abstractmethod
    def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """执行任务，接收输入字典，返回输出字典"""
        pass

    def get_name(self) -> str:
        return self.name

然后，实现几个具体的Agent。例如，一个调用Claude本身来生成文案的Agent：

class CopywriterAgent(BaseAgent):
    def __init__(self, api_key: str):
        super().__init__("copywriter_agent")
        self.client = anthropic.Anthropic(api_key=api_key)

    def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        product_desc = inputs.get("product_description", "")
        audience = inputs.get("target_audience", "")
        platform = inputs.get("platform", "twitter")

        tone = "简洁、有网感、带话题标签" if platform == "twitter" else "专业、详尽、突出价值"

        prompt = f"""你是一名专业的文案写手。请为以下产品创作一条适合在{platform}平台发布的推广文案。
        产品描述：{product_desc}
        目标受众：{audience}
        文案要求：风格应{tone}。长度适中。
        请直接输出文案内容，无需额外解释。"""

        try:
            response = self.client.messages.create(
                model="claude-3-haiku-20240307",  # 使用更快的模型执行具体任务
                max_tokens=500,
                messages=[{"role": "user", "content": prompt}]
            )
            generated_text = response.content[0].text
            return {"generated_post": generated_text.strip()}
        except Exception as e:
            return {"error": f"文案生成失败: {str(e)}", "generated_post": ""}

再比如，一个进行语气调整的Agent：

class ToneAdjusterAgent(BaseAgent):
    def __init__(self, api_key: str):
        super().__init__("tone_adjuster_agent")
        self.client = anthropic.Anthropic(api_key=api_key)

    def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        draft_text = inputs.get("draft_text", "")
        brand_tone = inputs.get("brand_tone", "专业且友好") # 可以从应用配置或用户输入获取

        prompt = f"""请将以下文案调整为符合品牌语调的风格。品牌语调是：{brand_tone}。
        原文案：
        {draft_text}
        请只输出调整后的文案。"""

        try:
            response = self.client.messages.create(
                model="claude-3-haiku-20240307",
                max_tokens=500,
                messages=[{"role": "user", "content": prompt}]
            )
            adjusted_text = response.content[0].text
            return {"adjusted_post": adjusted_text.strip()}
        except Exception as e:
            return {"error": f"语调调整失败: {str(e)}", "adjusted_post": draft_text} # 失败时返回原稿

有了这些Agent，我们需要一个注册中心来统一管理，方便规划器和执行引擎查找和调用：

class AgentRegistry:
    """智能体注册中心，单例模式"""
    _instance = None
    _agents: Dict[str, BaseAgent] = {}

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(AgentRegistry, cls).__new__(cls)
        return cls._instance

    def register_agent(self, agent: BaseAgent):
        """注册一个智能体"""
        self._agents[agent.get_name()] = agent
        print(f"Agent注册成功: {agent.get_name()}")

    def get_agent(self, name: str) -> BaseAgent:
        """根据名称获取智能体"""
        agent = self._agents.get(name)
        if not agent:
            raise KeyError(f"未找到名为 '{name}' 的Agent。请检查是否已注册。")
        return agent

    def list_agents(self) -> List[str]:
        """列出所有已注册的Agent名称"""
        return list(self._agents.keys())

# 初始化注册中心并注册Agent
registry = AgentRegistry()
registry.register_agent(CopywriterAgent(api_key=os.getenv("ANTHROPIC_API_KEY")))
registry.register_agent(ToneAdjusterAgent(api_key=os.getenv("ANTHROPIC_API_KEY")))

注意事项 ：这里用一个简单的单例模式实现注册中心。在实际生产环境中，你可能需要考虑更复杂的生命周期管理、依赖注入、以及从配置文件动态加载Agent。另外，同一个API密钥被多个Agent共用，在高并发下可能遇到速率限制，需要考虑密钥轮询或池化。

3.5 集成Ultraplan执行引擎

现在，我们有了解析应用描述的 AppManifest ，有了生成计划的 ClaudeCodePlanner ，有了具体干活的 BaseAgent 和 AgentRegistry 。最后一步，就是将Claude生成的计划交给Ultraplan来执行。我们需要一个适配层，将我们自定义的Agent与Ultraplan的执行器接口对接。

假设Ultraplan需要一个 Executor 类来执行每个 Task ，我们可以这样包装我们的Agent：

from ultraplan import Executor as UltraExecutor
from ultraplan import Task as UltraTask

class AgentExecutor(UltraExecutor):
    """适配器，将Ultraplan的Executor接口桥接到我们的BaseAgent"""
    def __init__(self, agent_registry: AgentRegistry):
        self.registry = agent_registry

    def execute(self, task: UltraTask, context: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行Ultraplan的任务。
        task: Ultraplan的任务对象，其`executor`属性应为我们注册的Agent名称。
        context: 执行上下文，包含之前任务的输出等。
        """
        agent_name = task.executor  # 例如 "copywriter_agent"
        try:
            # 1. 从注册中心获取对应的Agent实例
            agent = self.registry.get_agent(agent_name)

            # 2. 准备输入参数。这里需要解析task.inputs，它可能包含对上下文变量的引用。
            # 例如， inputs: {"product_description": "{{user_input}}", "target_audience": "{{task_1.output.audience_analysis}}"}
            # 我们需要将`{{...}}`的模板替换为context中实际的值。
            resolved_inputs = self._resolve_inputs(task.inputs, context)

            # 3. 调用Agent执行
            print(f"[执行器] 开始执行任务 {task.id}，使用Agent: {agent_name}")
            result = agent.execute(resolved_inputs)

            # 4. 将结果存入上下文，供后续任务使用
            # Ultraplan通常会自动处理，这里我们确保结果格式正确
            execution_result = {
                "task_id": task.id,
                "status": "success",
                "output": result
            }
            # 将Agent的输出合并到上下文中，key为任务ID，方便后续引用
            context[task.id] = result
            return execution_result

        except KeyError as e:
            return {"task_id": task.id, "status": "failed", "error": f"Agent未找到: {e}"}
        except Exception as e:
            return {"task_id": task.id, "status": "failed", "error": f"执行异常: {str(e)}"}

    def _resolve_inputs(self, inputs: Dict, context: Dict) -> Dict:
        """解析输入参数中的模板变量。这是一个简单的实现。"""
        resolved = {}
        for key, value in inputs.items():
            if isinstance(value, str) and value.startswith("{{") and value.endswith("}}"):
                # 提取变量名，如 "{{task_1.output.generated_post}}"
                var_path = value[2:-2].strip()  # 去掉大括号
                # 简单起见，假设路径是 task_id.output.key 的格式
                parts = var_path.split('.')
                current = context
                for part in parts:
                    if isinstance(current, dict) and part in current:
                        current = current[part]
                    else:
                        # 如果解析失败，保留原模板字符串或使用空值
                        current = value
                        break
                resolved[key] = current
            else:
                resolved[key] = value
        return resolved

最后，我们编写主程序，将以上所有模块串联起来：

import os
from dotenv import load_dotenv

def main():
    # 1. 加载环境变量
    load_dotenv()
    api_key = os.getenv("ANTHROPIC_API_KEY")

    # 2. 初始化核心组件
    planner = ClaudeCodePlanner(api_key=api_key)
    registry = AgentRegistry()
    # ... 注册所有需要的Agent到registry ...
    executor = AgentExecutor(registry)

    # 3. 用户选择应用并输入需求
    app_manifest_path = "apps/social_media_generator.yaml"
    user_request = "为我们的新产品‘智能笔记AI’生成一条推文。产品是一个能自动总结会议录音的SaaS工具，目标受众是忙碌的经理人和创业者。"

    # 4. 加载应用描述
    app_manifest = planner.load_app_manifest(app_manifest_path)
    print(f"加载应用: {app_manifest.name}")

    # 5. 调用Claude Code生成计划
    print("正在请求Claude Code生成工作流计划...")
    claude_plan_dict = planner.generate_plan(user_request, app_manifest)
    print("计划生成成功。")

    # 6. 转换为Ultraplan Plan对象
    ultra_plan = planner.convert_to_ultraplan(claude_plan_dict)

    # 7. 创建Ultraplan引擎并执行计划
    # 假设Ultraplan有一个简单的Engine类
    from ultraplan import Engine  # 示例，实际API可能不同
    engine = Engine(plan=ultra_plan, executor=executor)

    # 8. 准备初始上下文（用户输入）
    initial_context = {
        "user_input": user_request,
        "product_description": "智能笔记AI：一款能自动转录、总结会议录音，并生成行动项和待办清单的SaaS工具。",
        "target_audience": "忙碌的经理人、创业者、远程团队负责人",
        "platform": "twitter"
    }

    print("开始执行工作流...")
    final_results = engine.run(context=initial_context)

    # 9. 输出结果
    print("\n" + "="*50)
    print("工作流执行完成！")
    for task_id, result in final_results.items():
        if result.get("status") == "success":
            output = result.get("output", {})
            # 查找最终生成的文案
            if "generated_post" in output:
                print(f"生成的推文：\n{output['generated_post']}")
            elif "adjusted_post" in output:
                print(f"优化后的文案：\n{output['adjusted_post']}")
        else:
            print(f"任务 {task_id} 失败: {result.get('error')}")

if __name__ == "__main__":
    main()

4. 关键问题与优化策略实录

在实际搭建和测试这个系统的过程中，我遇到了不少坑，也总结出一些优化点。

4.1 规划可靠性问题与提示词工程

问题 ：初期，Claude Code生成的计划有时会“天马行空”，比如调用一个不存在的Agent，或者任务描述过于模糊（如“分析数据”），导致执行器无法理解。

解决方案 ：这本质上是提示词工程（Prompt Engineering）的问题。我通过迭代优化系统提示词来解决：

1. 提供更严格的约束 ：在提示词中明确列出可用的Agent名称列表，并要求Claude必须从中选择。例如：“可用智能体包括： copywriter_agent , tone_adjuster_agent , data_fetcher_agent 。你的计划中每个任务的 agent 字段必须严格使用这些名称之一。”
2. 定义“原子化任务”的标准 ：给出正面和反面例子。例如：

   好的任务描述 ：“调用 copywriter_agent ，输入为 product_description （来自用户请求）和 target_audience （来自用户请求），生成一篇初稿文案，输出到变量 draft_post 。” 坏的任务描述 ：“生成文案。”（过于模糊）

3. 要求输出结构化更强的JSON ：除了基本结构，还可以要求Claude为每个任务标注依赖的前置任务ID，这样能更清晰地构建DAG（有向无环图）。

优化后的提示词片段示例 ：

...（前文省略）...
请确保：
1.  每个任务描述必须清晰、具体、可操作，避免使用“处理”、“分析”等模糊词汇。
2.  每个任务的`agent`字段，必须从以下列表中选择：[copywriter_agent, tone_adjuster_agent]。
3.  如果任务B需要任务A的输出作为输入，请在任务B的`inputs`字段中明确引用，格式为 `{{task_A_id.output.key_name}}`。
4.  最终输出必须是一个合法的、可被`json.loads`解析的JSON对象。

4.2 执行上下文与变量传递

问题 ：Claude生成的计划中，任务间的输入输出引用（如 {{task_1.output.draft}} ）需要被正确解析。我最初实现的简单字符串替换在嵌套数据结构或复杂引用路径时容易出错。

解决方案 ：实现一个更健壮的上下文解析器。可以使用类似Jinja2的模板引擎，或者实现一个支持点号路径（如 task_1.output.data.summary ）的解析函数。关键是要统一整个系统中的数据格式约定，比如规定每个Agent的 execute 方法返回的必须是一个字典。

import re
def resolve_context(template: str, context: Dict) -> Any:
    """使用正则表达式解析简单的 {{path.to.value}} 模板"""
    pattern = r"\{\{\s*([\w\.]+)\s*\}\}"
    def replacer(match):
        path = match.group(1)
        parts = path.split('.')
        current = context
        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            else:
                # 如果路径不存在，可以返回None或原字符串，并记录警告
                return match.group(0) # 返回原模板
        return current
    return re.sub(pattern, replacer, template)

# 使用示例
context = {"task_1": {"output": {"draft": "Hello World"}}}
input_str = "前缀 {{task_1.output.draft}} 后缀"
resolved = resolve_context(input_str, context) # 输出： “前缀 Hello World 后缀”

4.3 错误处理与任务重试

问题 ：网络波动、API限额、或Agent内部逻辑错误都可能导致单个任务失败。一个任务失败不应导致整个工作流崩溃，而应有相应的容错机制。

解决方案 ：在 AgentExecutor.execute 方法中加强错误捕获，并设计重试逻辑。同时，规划器Claude Code也可以被赋予“观察-再规划”的能力。

1. 本地重试 ：对于网络等临时错误，可以加入指数退避重试。

   import time
   def execute_with_retry(self, task, context, max_retries=3):
       for attempt in range(max_retries):
           try:
               return self._execute_once(task, context)
           except TemporaryError as e: # 自定义临时错误异常
               if attempt == max_retries - 1:
                   raise
               wait_time = (2 ** attempt) + random.random()
               time.sleep(wait_time)
   

2. 全局重规划 ：当任务失败且无法自动恢复时，可以将错误信息（如“ copywriter_agent 因内容政策失败”）和当前上下文反馈给Claude Code规划器，请求它生成一个新的、绕过该问题的计划。这实现了简单的ReAct（Reasoning and Acting）循环。

4.4 性能与成本考量

问题 ：每次执行工作流都要调用Claude Code进行规划，对于简单或重复的工作流，这会产生不必要的延迟和API调用成本。

解决方案 ：引入计划缓存（Plan Caching）。

1. 缓存键 ：将“用户请求”和“应用描述文件的哈希值”组合起来作为缓存键。如果相同的请求再次出现，直接使用缓存的计划，无需再次调用Claude。
2. 缓存失效 ：当应用描述文件（YAML）更新时，使对应的缓存失效。
3. 成本监控 ：记录每次调用Claude的Token消耗，便于分析和优化提示词，控制成本。

import hashlib
import pickle
import os

class CachedPlanner(ClaudeCodePlanner):
    def __init__(self, api_key: str, cache_dir: str = "./cache/plans"):
        super().__init__(api_key)
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def generate_plan(self, user_request: str, app_manifest: AppManifest) -> Dict[str, Any]:
        # 生成缓存键
        cache_key_data = f"{user_request}_{app_manifest.name}_{app_manifest.version}"
        cache_key = hashlib.md5(cache_key_data.encode()).hexdigest()
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")

        # 检查缓存
        if os.path.exists(cache_file):
            print(f"从缓存加载计划: {cache_key}")
            with open(cache_file, 'rb') as f:
                return pickle.load(f)

        # 缓存未命中，调用父类方法生成新计划
        print("缓存未命中，调用Claude生成新计划...")
        plan = super().generate_plan(user_request, app_manifest)

        # 保存到缓存
        with open(cache_file, 'wb') as f:
            pickle.dump(plan, f)
        print(f"计划已缓存: {cache_key}")
        return plan

5. 扩展方向与应用场景探讨

这个基础框架搭建起来后，可以朝多个方向扩展，以适应更复杂的场景。

5.1 扩展一：集成外部工具与API

真正的自动化工作流离不开与外部世界的交互。我们可以创建更多类型的Agent：

• Web搜索Agent ：集成Serper API或类似服务，让工作流能获取实时信息。
• 代码执行Agent ：在安全沙箱中运行Python代码，进行数据处理或计算。
• 软件操作Agent ：通过Selenium或Playwright控制浏览器，或通过RPA工具操作桌面软件。
• 文件操作Agent ：读写本地或云存储的文件。

关键是为这些工具设计统一的 BaseAgent 接口，并妥善处理认证、安全（特别是代码执行）和错误。

5.2 扩展二：实现真正的“应用商店”

目前“应用商店”还只是一个概念。可以将其具体化：

1. 中央应用仓库 ：建立一个存储所有应用描述文件（YAML）的Git仓库或数据库。
2. 应用发现与安装CLI ：开发一个命令行工具，用户可以浏览仓库、搜索应用，并通过一条命令（如 ultraplan-app install social-media-generator ）将应用描述文件下载到本地 apps/ 目录。
3. 依赖管理 ：在应用描述文件中增加 dependencies 字段，声明需要哪些Agent或Python包。安装时自动检查并提示用户配置。
4. 用户界面 ：开发一个简单的Web UI，让非技术用户也能通过点击来选择和运行AI应用。

5.3 典型应用场景

这个系统非常适合以下场景：

• 个性化内容工厂 ：输入一个主题，自动完成从搜集资料、生成文章大纲、撰写初稿、润色排版到发布到CMS的全流程。
• 智能数据分析助手 ：连接数据库或上传数据文件，自动完成数据清洗、分析、可视化图表生成，并撰写分析结论。
• 自动化客户支持 ：监控客服工单，自动分类、提取关键信息、生成初步回复建议，甚至直接处理简单查询。
• 个人知识管理 ：自动整理收藏的文章、视频，进行摘要总结，并按照自定义标签体系归档。

这个项目的魅力在于，一旦核心的“规划-执行”引擎搭建完成，新的应用（工作流）就变成了一个描述文件（YAML）和几个Agent的组合，极大地降低了创建复杂自动化流程的门槛。它把编程的灵活性交给了Claude Code这样的AI，而人类则专注于定义“做什么”和“用什么做”，而不是“怎么做”的每一个细节。