在实际 AI 应用开发中，很多团队面临一个现实问题：大模型 API 调用成本随着任务复杂度线性增长，特别是需要长时间运行、多步骤协作的智能体系统。Simon Willison 提出的 Fable 省钱法核心思路是：用高成本的 Fable 5 模型作为指挥中心，负责复杂决策和工作流编排，而将大量机械性、重复性的子任务交给廉价的子代理模型处理。这种架构既保证了系统智能度，又显著降低了运营成本。

这种设计模式特别适合需要连续运行数天甚至数周的自主系统，比如自动化测试生成、数据清洗流水线、文档处理自动化等场景。下面我们将从架构设计到具体实现，完整介绍如何构建这样一个成本优化的智能体系统。

1. 理解 Fable 5 模型的特性和成本结构

Fable 5 作为 Claude 系列的 Mythos 级模型，与 Sonnet、Opus、Haiku 在能力和定价上存在显著差异。正确理解这些差异是设计省钱架构的基础。

1.1 Claude 各模型的能力定位和成本对比

在实际项目中，需要根据任务复杂度合理分配模型资源。以下是各模型的典型使用场景和成本考量：

模型	能力定位	适合任务类型	成本相对比例	会话持续时间
Fable 5	重型编排角色，复杂决策	多步骤工作流编排、长期自治系统	最高	数天级别
Opus 4.8	复杂但有界的子任务	代码审查、复杂逻辑分析	高	数小时
Sonnet 4.6	高频工人任务	API 调用、数据转换、简单代码生成	中等	分钟到小时
Haiku 4.5	评分子代理	质量检查、简单验证、分类	最低	分钟级别

从成本角度分析，如果整个系统全部使用 Fable 5，即使是中等规模的项目，月度 API 成本也可能达到数千美元。而采用分层架构后，Fable 5 只处理核心决策，大量子任务由 Sonnet 或 Haiku 处理，成本可以降低 60-80%。

1.2 Fable 5 的独特能力：为什么它适合做指挥中心

Fable 5 设计为连续运行数天的模型，这决定了它在长期记忆和状态保持方面的优势：

• 内置自我验证机制 ：能够对自身输出进行质量检查，减少人工干预
• 复杂代码工作能力 ：可以处理涉及多个模块、需要深度理解的编程任务
• 多阶段知识工作 ：能够将复杂任务分解为有序阶段，并保持上下文连贯性
• 状态文件积累 ：每次会话的经验教训可以写入状态文件，实现系统层面的持续改进

这些特性让 Fable 5 非常适合作为智能体系统的"大脑"，而机械性的执行工作完全可以下放给更经济的模型。

2. 设计分层智能体系统架构

构建成本优化的智能体系统需要清晰的架构设计。下面是一个典型的分层架构实现方案。

2.1 四层复利堆栈设计

Simon Willison 提出的复利堆栈包含四个关键层次：

# 智能体系统架构示例（概念代码）
class CostOptimizedAgentSystem:
    def __init__(self):
        self.primitives = {
            'fable_5': Fable5Client(),
            'sonnet_4_6': SonnetClient(), 
            'haiku_4_5': HaikuClient(),
            'tools': ToolRegistry()
        }
        self.orchestration = OrchestrationLayer()
        self.memory = MemoryLayer()
        self.self_improvement = SelfImprovementLayer()
    
    def route_task(self, task_complexity):
        """根据任务复杂度路由到合适的模型"""
        if task_complexity == 'high':
            return self.primitives['fable_5']
        elif task_complexity == 'medium':
            return self.primitives['sonnet_4_6'] 
        else:
            return self.primitives['haiku_4_5']

这个架构的核心思想是：Fable 5 负责最上层的任务分解和决策，然后将具体的子任务分发给合适的廉价模型执行。

2.2 工作流编排模式

在实际实现中，有三种关键的工作流模式特别有效：

扇出-综合模式 ：Fable 5 将大型任务分解为多个并行子任务，由多个 Sonnet 实例同时处理，最后再汇总结果。

def fan_out_synthesize_pattern(main_task):
    # Fable 5 分析任务并分解
    subtasks = fable5_analyze_and_split(main_task)
    
    # 并行分发给多个 Sonnet 实例
    sonnet_results = []
    for subtask in subtasks:
        result = sonnet_4_6_process(subtask)
        sonnet_results.append(result)
    
    # Fable 5 综合结果
    final_result = fable5_synthesize(sonnet_results)
    return final_result

对抗验证模式 ：Fable 5 生成内容，由专门的 Haiku 验证器进行检查，确保质量的同时控制成本。

def adversarial_validation_pattern(content_generation_task):
    # Fable 5 生成主要内容
    generated_content = fable5_generate(content_generation_task)
    
    # Haiku 进行质量验证（低成本）
    validation_result = haiku_4_5_validate(generated_content)
    
    if not validation_result.passed:
        # 如果验证失败，Fable 5 进行修正
        corrected_content = fable5_correct(generated_content, validation_result.feedback)
        return corrected_content
    
    return generated_content

循环直到完成模式 ：对于需要迭代改进的任务，Fable 5 设定检查点，由 Sonnet 进行进度评估。

def loop_until_done_pattern(iterative_task):
    current_state = iterative_task.initial_state
    max_iterations = 10
    
    for iteration in range(max_iterations):
        # Fable 5 决定下一步行动
        next_action = fable5_decide_next_action(current_state, iterative_task.goal)
        
        # Sonnet 执行具体操作
        execution_result = sonnet_4_6_execute(next_action)
        
        # Haiku 检查是否完成（低成本检查）
        is_complete = haiku_4_5_check_completion(execution_result, iterative_task.goal)
        
        if is_complete:
            return execution_result
        
        current_state = execution_result
    
    return current_state  # 返回最佳可用结果

3. 实现机械任务子代理系统

机械任务的特点是重复性高、逻辑相对简单，但可能数量庞大。这类任务正是省钱法的重点优化对象。

3.1 识别适合子代理的机械任务类型

在实际项目中，以下类型的任务特别适合交给廉价子代理处理：

• 数据提取和转换 ：从结构化文档中提取信息并转换为指定格式
• API 调用和响应处理 ：调用外部服务并处理标准化响应
• 简单代码生成 ：根据模板生成重复性的代码片段
• 质量检查和验证 ：检查输出是否符合基本规范
• 文件操作和批量处理 ：重命名、移动、格式转换等操作

3.2 子代理的实现示例

以下是一个用于处理数据清洗任务的 Sonnet 子代理实现示例：

import json
from typing import List, Dict

class DataCleaningSubAgent:
    def __init__(self, sonnet_client):
        self.client = sonnet_client
        
    def clean_csv_data(self, raw_data: List[Dict], cleaning_rules: Dict) -> List[Dict]:
        """
        使用 Sonnet 4.6 清洗 CSV 数据
        成本远低于使用 Fable 5 处理同样任务
        """
        prompt = f"""
        请根据以下清洗规则处理CSV数据：
        清洗规则：{json.dumps(cleaning_rules, indent=2)}
        
        原始数据：{json.dumps(raw_data, indent=2)}
        
        请执行以下操作：
        1. 移除空行和完全为空的行
        2. 根据规则标准化日期格式
        3. 验证必填字段完整性
        4. 根据规则转换数值格式
        5. 返回清洗后的JSON数组
        
        只返回处理后的JSON数据，不要额外解释。
        """
        
        response = self.client.complete(prompt, max_tokens=2000)
        cleaned_data = self._extract_json_from_response(response)
        return cleaned_data
    
    def _extract_json_from_response(self, response: str) -> List[Dict]:
        """从响应中提取JSON数据"""
        try:
            # 尝试直接解析JSON
            return json.loads(response)
        except json.JSONDecodeError:
            # 如果直接解析失败，尝试提取JSON部分
            import re
            json_match = re.search(r'\[.*\]', response, re.DOTALL)
            if json_match:
                return json.loads(json_match.group())
            raise ValueError("无法从响应中提取有效的JSON数据")

3.3 工作树并行安全机制

当多个子代理并行工作时，需要确保文件操作不会冲突。Worktrees 机制可以很好地解决这个问题：

import os
import tempfile
from pathlib import Path

class ParallelWorktreeManager:
    def __init__(self, base_work_dir):
        self.base_work_dir = Path(base_work_dir)
        self.active_worktrees = {}
        
    def create_worktree_for_agent(self, agent_id, base_branch="main"):
        """为每个子代理创建独立的工作树"""
        worktree_path = self.base_work_dir / f"worktree_{agent_id}"
        
        # 创建独立的工作目录
        worktree_path.mkdir(exist_ok=True)
        
        # 记录工作树信息
        self.active_worktrees[agent_id] = {
            'path': worktree_path,
            'branch': base_branch,
            'created_at': datetime.now()
        }
        
        return worktree_path
    
    def cleanup_worktree(self, agent_id):
        """清理工作树"""
        if agent_id in self.active_worktrees:
            worktree_path = self.active_worktrees[agent_id]['path']
            # 安全删除工作目录
            import shutil
            shutil.rmtree(worktree_path)
            del self.active_worktrees[agent_id]

4. 状态管理和经验积累系统

Fable 省钱法的另一个关键优势是状态文件的持续积累，这让系统能够越用越聪明，而不是每次从头开始。

4.1 五阶段记忆进化机制

完整的自我改进系统包含五个进化阶段：

1. 失败记录 ：记录任务执行中的问题和错误
2. 根本原因调查 ：分析问题产生的深层原因
3. 解决方案验证 ：测试并验证修复方案的有效性
4. 经验规则提炼 ：将成功经验抽象为可复用的规则
5. 知识库集成 ：将规则集成到系统的长期记忆中

4.2 状态文件设计

状态文件应该包含结构化信息，支持系统的持续学习：

{
  "system_state": {
    "version": "1.0",
    "last_updated": "2026-06-15T10:30:00Z",
    "total_tasks_processed": 1567
  },
  "skill_library": {
    "data_cleaning": {
      "success_rate": 0.94,
      "common_issues": [
        {
          "issue": "日期格式不一致",
          "solution": "统一使用ISO 8601格式",
          "last_occurred": "2026-06-10",
          "occurrence_count": 23
        }
      ],
      "best_practices": [
        "在处理前先验证数据源编码",
        "对于大型文件使用流式处理"
      ]
    }
  },
  "agent_performance": {
    "sonnet_4_6": {
      "tasks_completed": 1245,
      "average_cost_per_task": 0.02,
      "success_rate": 0.89
    },
    "haiku_4_5": {
      "tasks_completed": 322,
      "average_cost_per_task": 0.005,
      "success_rate": 0.92
    }
  },
  "failure_log": [
    {
      "timestamp": "2026-06-14T15:22:00Z",
      "task_type": "api_integration",
      "error_message": "API速率限制 exceeded",
      "resolved": true,
      "solution_applied": "添加指数退避重试机制"
    }
  ]
}

4.3 状态文件的读写机制

实现一个高效的状态管理系统：

import json
from datetime import datetime
from pathlib import Path

class StateManager:
    def __init__(self, state_file_path):
        self.state_file = Path(state_file_path)
        self.state = self._load_state()
    
    def _load_state(self):
        """加载状态文件"""
        if self.state_file.exists():
            with open(self.state_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        else:
            return self._create_initial_state()
    
    def _create_initial_state(self):
        """创建初始状态"""
        return {
            "system_state": {
                "version": "1.0",
                "created_at": datetime.now().isoformat(),
                "total_tasks_processed": 0
            },
            "skill_library": {},
            "agent_performance": {},
            "failure_log": []
        }
    
    def update_skill_success(self, skill_name, success=True):
        """更新技能成功率"""
        if skill_name not in self.state["skill_library"]:
            self.state["skill_library"][skill_name] = {
                "success_count": 0,
                "total_attempts": 0,
                "success_rate": 0.0
            }
        
        skill_data = self.state["skill_library"][skill_name]
        skill_data["total_attempts"] += 1
        if success:
            skill_data["success_count"] += 1
        skill_data["success_rate"] = skill_data["success_count"] / skill_data["total_attempts"]
        
        self._save_state()
    
    def log_failure(self, task_type, error_message, solution=None):
        """记录失败信息"""
        failure_entry = {
            "timestamp": datetime.now().isoformat(),
            "task_type": task_type,
            "error_message": error_message,
            "resolved": solution is not None,
            "solution_applied": solution
        }
        
        self.state["failure_log"].append(failure_entry)
        # 保持失败日志不超过1000条
        if len(self.state["failure_log"]) > 1000:
            self.state["failure_log"] = self.state["failure_log"][-1000:]
        
        self._save_state()
    
    def _save_state(self):
        """保存状态到文件"""
        self.state["system_state"]["last_updated"] = datetime.now().isoformat()
        with open(self.state_file, 'w', encoding='utf-8') as f:
            json.dump(self.state, f, indent=2, ensure_ascii=False)

5. 成本监控和优化策略

实施省钱法后，需要建立监控机制来确保成本优化效果符合预期。

5.1 成本监控仪表板

创建一个简单的成本监控系统：

class CostMonitor:
    def __init__(self):
        self.daily_costs = {}
        self.model_usage = {
            'fable_5': {'count': 0, 'cost': 0.0},
            'sonnet_4_6': {'count': 0, 'cost': 0.0},
            'haiku_4_5': {'count': 0, 'cost': 0.0}
        }
    
    def record_api_call(self, model_name, cost_estimate, task_description=""):
        """记录API调用成本"""
        today = datetime.now().date().isoformat()
        
        if today not in self.daily_costs:
            self.daily_costs[today] = 0.0
        
        self.daily_costs[today] += cost_estimate
        self.model_usage[model_name]['count'] += 1
        self.model_usage[model_name]['cost'] += cost_estimate
        
        # 记录详细日志（可选）
        self._log_detailed_usage(model_name, cost_estimate, task_description)
    
    def get_cost_savings_report(self):
        """生成成本节省报告"""
        total_cost = sum(self.daily_costs.values())
        fable_ratio = self.model_usage['fable_5']['cost'] / total_cost if total_cost > 0 else 0
        
        report = {
            'total_cost': total_cost,
            'cost_distribution': self.model_usage,
            'fable_5_usage_ratio': fable_ratio,
            'estimated_savings': self._estimate_savings()
        }
        
        return report
    
    def _estimate_savings(self):
        """估算相比全量使用Fable 5节省的成本"""
        total_tasks = sum(usage['count'] for usage in self.model_usage.values())
        # 假设所有任务都用Fable 5的成本（估算）
        all_fable_cost = total_tasks * 0.10  # 假设每个Fable 5任务成本$0.10
        actual_cost = sum(usage['cost'] for usage in self.model_usage.values())
        
        return {
            'savings_amount': all_fable_cost - actual_cost,
            'savings_percentage': (all_fable_cost - actual_cost) / all_fable_cost * 100
        }

5.2 动态任务路由优化

基于成本监控数据，可以进一步优化任务路由策略：

class DynamicTaskRouter:
    def __init__(self, cost_monitor, performance_threshold=0.85):
        self.cost_monitor = cost_monitor
        self.performance_threshold = performance_threshold
        self.routing_rules = self._load_routing_rules()
    
    def determine_best_model(self, task_complexity, task_type, historical_success_rate):
        """根据任务特征确定最合适的模型"""
        # 如果历史成功率低，可能需要更强大的模型
        if historical_success_rate < self.performance_threshold:
            return 'fable_5'
        
        # 根据任务复杂度路由
        if task_complexity == 'high':
            return 'fable_5'
        elif task_complexity == 'medium':
            # 检查Sonnet在该类任务上的表现
            sonnet_performance = self._get_model_performance('sonnet_4_6', task_type)
            if sonnet_performance > self.performance_threshold:
                return 'sonnet_4_6'
            else:
                return 'fable_5'
        else:  # low complexity
            # 优先使用成本最低的模型
            haiku_performance = self._get_model_performance('haiku_4_5', task_type)
            if haiku_performance > self.performance_threshold:
                return 'haiku_4_5'
            elif self._get_model_performance('sonnet_4_6', task_type) > self.performance_threshold:
                return 'sonnet_4_6'
            else:
                return 'fable_5'
    
    def _get_model_performance(self, model_name, task_type):
        """获取模型在特定任务类型上的历史表现"""
        # 从状态管理器中获取性能数据
        # 简化实现，实际项目中需要查询详细的历史数据
        return 0.9  # 默认返回90%成功率

6. 常见问题排查和优化建议

在实际部署分层智能体系统时，可能会遇到一些典型问题。

6.1 成本优化不明显的问题排查

如果发现成本节省效果不如预期，可以按照以下步骤排查：

问题现象	可能原因	检查方式	解决方案
Fable 5 使用比例过高	任务路由策略过于保守	检查路由决策日志，分析任务复杂度判断是否准确	调整路由阈值，让更多任务流向廉价模型
子代理任务失败率偏高	子代理能力不足以处理分配的任务	分析失败任务的类型和错误信息	重新评估任务分配策略，或为子代理提供更详细的指令
状态文件过大影响性能	积累了过多历史数据	检查状态文件大小和加载时间	实现数据归档机制，只保留最近的关键信息

6.2 系统性能优化建议

基于实际项目经验，以下优化措施通常能显著提升系统效果：

指令优化 ：为子代理提供更精确的指令模板，减少理解偏差。

# 优化前的指令
"请处理这个数据文件"

# 优化后的指令
"""
请执行以下数据处理步骤：
1. 读取CSV文件，跳过前2行表头
2. 将第3列的数字字符串转换为浮点数
3. 删除任何包含空值的行
4. 将结果保存为新的CSV文件，使用UTF-8编码
"""

批量处理 ：将多个小任务合并为批量任务，减少API调用开销。

def batch_process_tasks(tasks, batch_size=10):
    """批量处理任务以减少API调用次数"""
    results = []
    for i in range(0, len(tasks), batch_size):
        batch = tasks[i:i+batch_size]
        batch_prompt = self._create_batch_prompt(batch)
        batch_result = self.sonnet_client.complete(batch_prompt)
        results.extend(self._parse_batch_result(batch_result))
    return results

缓存机制 ：对重复性查询结果进行缓存，避免重复计算。

from functools import lru_cache
import hashlib

class SmartCache:
    def __init__(self, max_size=1000):
        self.cache = {}
        self.max_size = max_size
    
    def get_cache_key(self, prompt, parameters):
        """生成基于内容和参数的缓存键"""
        content = prompt + str(parameters)
        return hashlib.md5(content.encode()).hexdigest()
    
    def get_cached_result(self, prompt, parameters):
        """获取缓存结果"""
        key = self.get_cache_key(prompt, parameters)
        return self.cache.get(key)
    
    def set_cached_result(self, prompt, parameters, result):
        """设置缓存结果"""
        if len(self.cache) >= self.max_size:
            # 简单的LRU策略：移除最早的一半缓存
            keys_to_remove = list(self.cache.keys())[:self.max_size//2]
            for key in keys_to_remove:
                del self.cache[key]
        
        key = self.get_cache_key(prompt, parameters)
        self.cache[key] = result

6.3 安全边界和降级机制

Fable 5 在特定敏感领域会自动降级到 Opus 4.8，在实际项目中也需要建立类似的安全机制：

class SafetyChecker:
    def __init__(self):
        self.sensitive_domains = [
            'cybersecurity', 'biotechnology', 'chemistry', 'model_distillation'
        ]
    
    def check_task_safety(self, task_description):
        """检查任务是否涉及敏感领域"""
        for domain in self.sensitive_domains:
            if domain in task_description.lower():
                return False, f"任务涉及敏感领域: {domain}"
        return True, "任务安全"
    
    def apply_safety_downgrade(self, task_description, intended_model):
        """必要时应用安全降级"""
        is_safe, reason = self.check_task_safety(task_description)
        if not is_safe and intended_model == 'fable_5':
            print(f"安全降级: {reason}, 使用Opus 4.8代替Fable 5")
            return 'opus_4_8'
        return intended_model

Simon Willison 的 Fable 省钱法本质上是将软件工程中的分层设计原则应用到了 AI 智能体系统架构中。通过让昂贵的模型专注于核心决策，廉价模型处理机械任务，既保证了系统能力又控制了成本。这种模式特别适合需要长期运行、不断改进的生产环境系统。

在实际实施时，关键是要建立细粒度的任务分类机制、可靠的状态管理系统以及持续的成本监控体系。随着系统运行时间的积累，状态文件中保存的经验教训会让整个系统越来越智能，同时成本优化效果也会更加明显。