在实际 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, indent2)} 原始数据{json.dumps(raw_data, indent2)} 请执行以下操作 1. 移除空行和完全为空的行 2. 根据规则标准化日期格式 3. 验证必填字段完整性 4. 根据规则转换数值格式 5. 返回清洗后的JSON数组 只返回处理后的JSON数据不要额外解释。 response self.client.complete(prompt, max_tokens2000) 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_branchmain): 为每个子代理创建独立的工作树 worktree_path self.base_work_dir / fworktree_{agent_id} # 创建独立的工作目录 worktree_path.mkdir(exist_okTrue) # 记录工作树信息 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 五阶段记忆进化机制完整的自我改进系统包含五个进化阶段失败记录记录任务执行中的问题和错误根本原因调查分析问题产生的深层原因解决方案验证测试并验证修复方案的有效性经验规则提炼将成功经验抽象为可复用的规则知识库集成将规则集成到系统的长期记忆中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, encodingutf-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, successTrue): 更新技能成功率 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, solutionNone): 记录失败信息 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, encodingutf-8) as f: json.dump(self.state, f, indent2, ensure_asciiFalse)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_threshold0.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_size10): 批量处理任务以减少API调用次数 results [] for i in range(0, len(tasks), batch_size): batch tasks[i:ibatch_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_size1000): 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] result6.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_modelSimon Willison 的 Fable 省钱法本质上是将软件工程中的分层设计原则应用到了 AI 智能体系统架构中。通过让昂贵的模型专注于核心决策廉价模型处理机械任务既保证了系统能力又控制了成本。这种模式特别适合需要长期运行、不断改进的生产环境系统。在实际实施时关键是要建立细粒度的任务分类机制、可靠的状态管理系统以及持续的成本监控体系。随着系统运行时间的积累状态文件中保存的经验教训会让整个系统越来越智能同时成本优化效果也会更加明显。