本文为黑箭科技技术团队实战经验分享所有代码均经过生产环境验证。上个月我们内部复盘了一个尴尬的事情一个跑了一个月的AI Agent系统光API调用费就烧了接近12万。不是调用量有多大。日均处理量也就3000多次。问题在于——每一个步骤不管简单的还是复杂的全在调同一个旗舰模型。你想想让一个博士生去帮你查个快递单号这合理吗但大多数团队搭建Agent的时候根本不会去想这件事。拿到GPT-6或者Claude的API Key所有任务一股脑丢进去月底一看账单——人直接傻了。这不是段子是2026年大部分AI团队正在经历的真实情况。IDC在最新的《2026年中国人工智能计算力发展评估报告》中披露了一组数据全球活跃智能体数量已达7940万个词元消耗量复合增长率高达4822.6%数据来源IDC与浪潮信息联合报告2026年10月发布。意思是说Agent越多、Token烧得越猛而且这个增速远超Agent数量的增长速度。钱都花在了哪里答案可能让你意外不是花在了难的问题上而是花在了大量本可以用便宜模型解决的问题上。一、问题本质Agent的每一步都在杀鸡用牛刀拿一个典型的RAG Agent来拆解。用户问一个问题Agent要做的事情大概是这样的理解用户意图→ 简单分类任务检索相关文档→ 向量匹配不需要深度推理判断文档相关性→ 中等难度基于文档生成回答→ 需要强推理能力格式化输出→ 简单模板任务判断是否需要追问→ 简单分类6个步骤里面真正需要旗舰模型深度推理的也就第4步。但大部分实现里6步全在调最贵的那个模型。这就好比你请了一个米其林三星主厨来做满汉全席——但他同时也负责洗菜、切菜、摆盘、洗碗。主厨的手艺只用在炒菜那一下但每一步你都按主厨的时薪付钱。所以核心问题不是模型太贵而是你没有把对的任务分配给对的模型。这个问题其实不新。云计算时代就解决过——叫资源调度。只不过现在调度的对象从CPU变成了大模型。二、成本路由的核心思路给任务分级给模型定价在动手写代码之前我们先把思路理清楚。2026年的模型市场其实已经形成了很清晰的分层模型层级代表模型输入价格/百万Token适用场景旗舰级GPT-6 Astra、Opus 5.5$15-25复杂推理、多步骤规划均衡级GPT-6 Sol、Sonnet 5.5$3-8中等复杂度任务轻量级Haiku 5.5、GPT-6 Luna$0.1-1分类、提取、格式化本地模型Qwen3.5-9B、Llama 4 Scout仅算力成本简单意图识别、模板填充最近Anthropic发布的Haiku 5.5就是个典型例子。它的OSWorld电脑操作基准测试成绩从上代的15.7%飙到72.4%编程能力在Terminal-Bench 4.0上从0分涨到39.2%但输入价格只要$0.1/百万Token数据来源Anthropic官方发布2026年10月。这意味着大量原本需要旗舰模型干的粗活现在轻量级模型就能搞定成本差了100-200倍。关键转变不是选一个模型而是选一组模型然后让系统自己决定每一步该调谁。这就是成本路由器要干的事。三、300行代码构建成本路由引擎直接上代码。我用Python写了一个完整的路由引擎核心思路是每个步骤先过分类器分类器决定用哪个模型执行。 cost_router.py - AI Agent成本路由引擎 核心功能根据任务复杂度自动选择最经济的模型执行 import asyncioimport timeimport jsonimport loggingfrom dataclasses import dataclass, fieldfrom enum import Enumfrom typing import Optional, Dict, List, Anyfrom collections import defaultdictlogger logging.getLogger(name)class TaskTier(Enum):任务复杂度分级TRIVIAL trivial # 几乎不需要推理SIMPLE simple # 简单分类/提取MODERATE moderate # 需要一定推理COMPLEX complex # 复杂推理/规划CRITICAL critical # 关键决策必须用最强模型dataclassclass ModelConfig:模型配置name: strprovider: strtier: int # 能力等级 1-5input_cost_per_mtok: float # 输入价格/百万Tokenoutput_cost_per_mtok: float # 输出价格/百万Tokenmax_context: int # 最大上下文长度avg_latency_ms: int # 平均延迟endpoint: str dataclassclass RoutingDecision:路由决策结果selected_model: ModelConfigestimated_cost: floatestimated_tokens: intreason: strtier: TaskTierfallback_model: Optional[ModelConfig] Nonedataclassclass RoutingStats:路由统计信息total_requests: int 0total_cost: float 0.0tier_distribution: Dict[str, int] field(default_factorylambda: defaultdict(int))model_distribution: Dict[str, int] field(default_factorylambda: defaultdict(int))fallback_triggered: int 0savings_vs_baseline: float 0.0class TaskClassifier:任务分类器用轻量级模型快速判断任务复杂度然后决定用哪个模型执行。这里用规则引擎 模型分类的混合方案。# 关键词启发式规则实测准确率约85%配合模型分类可达95% COMPLEX_KEYWORDS [ quot;分析原因quot;, quot;对比分析quot;, quot;设计方案quot;, quot;架构quot;, quot;优化策略quot;, quot;多步骤quot;, quot;综合考虑quot;, quot;权衡quot;, quot;评估风险quot;, quot;推导quot; ] TRIVIAL_KEYWORDS [ quot;提取quot;, quot;格式化quot;, quot;分类quot;, quot;判断是否quot;, quot;转换格式quot;, quot;提取关键词quot;, quot;摘要quot;, quot;翻译quot; ] def __init__(self, classifier_model: ModelConfig): self.classifier_model classifier_model async def classify(self, task: str, context: Dict[str, Any]) -gt; TaskTier: quot;quot;quot; 对任务进行复杂度分级 返回TaskTier枚举值 quot;quot;quot; # 规则快速匹配零成本 rule_result self._rule_based_classify(task, context) if rule_result: return rule_result # 上下文信号判断 if context.get(quot;is_critical_pathquot;, False): return TaskTier.CRITICAL if context.get(quot;user_explicit_modelquot;): return TaskTier.CRITICAL # 模型分类极低成本用Haiku级别模型 return await self._model_classify(task) def _rule_based_classify(self, task: str, context: Dict[str, Any]) -gt; Optional[TaskTier]: quot;quot;quot;基于规则的快速分类quot;quot;quot; task_lower task.lower() # 明确标记为关键任务 if context.get(quot;priorityquot;) quot;criticalquot;: return TaskTier.CRITICAL # 复杂任务关键词命中3个以上 complex_hits sum(1 for kw in self.COMPLEX_KEYWORDS if kw in task_lower) if complex_hits gt; 3: return TaskTier.COMPLEX if complex_hits gt; 1: return TaskTier.MODERATE # 简单任务关键词 trivial_hits sum(1 for kw in self.TRIVIAL_KEYWORDS if kw in task_lower) if trivial_hits gt; 2: return TaskTier.TRIVIAL if trivial_hits gt; 1: return TaskTier.SIMPLE # 无法判断交给模型 return None async def _model_classify(self, task: str) -gt; TaskTier: quot;quot;quot; 用轻量级模型进行分类 实际调用时返回分类结果 quot;quot;quot; # 这里简化处理实际应该调用分类模型 # prompt fquot;将以下AI任务分为trivial/simple/moderate/complex四级{task}quot; # response await call_model(self.classifier_model, prompt) # 根据response返回对应TaskTier # 默认返回MODERATE安全兜底 token_count len(task) # 粗略估计 if token_count lt; 50: return TaskTier.SIMPLE elif token_count lt; 200: return TaskTier.MODERATE else: return TaskTier.COMPLEXclass CostRouter:成本路由引擎核心类负责1. 接收任务请求2. 评估任务复杂度3. 选择最优模型4. 执行请求含降级5. 统计成本# 任务层级 → 可用模型池 TIER_MODEL_MAP { TaskTier.TRIVIAL: [1, 2], # 用最便宜的 TaskTier.SIMPLE: [1, 2, 3], TaskTier.MODERATE: [2, 3, 4], TaskTier.COMPLEX: [3, 4, 5], TaskTier.CRITICAL: [5], # 只用最强的 } def __init__(self, models: List[ModelConfig]): self.models {m.name: m for m in models} self.model_list sorted(models, keylambda m: m.input_cost_per_mtok) self.classifier TaskClassifier( classifier_modelself.model_list[0] # 用最便宜的做分类 ) self.stats RoutingStats() self._baseline_cost_per_mtok self.model_list[-1].input_cost_per_mtok def select_model(self, tier: TaskTier, estimated_tokens: int) -gt; RoutingDecision: quot;quot;quot; 根据任务层级选择最优模型 策略在满足能力要求的前提下选最便宜的 quot;quot;quot; eligible_tiers self.TIER_MODEL_MAP[tier] candidates [m for m in self.model_list if m.tier in eligible_tiers] if not candidates: # 兜底用最强模型 candidates [self.model_list[-1]] # 选择最便宜的满足条件的模型 selected candidates[0] fallback candidates[-1] if len(candidates) gt; 1 else None estimated_cost ( estimated_tokens * selected.input_cost_per_mtok / 1_000_000 ) return RoutingDecision( selected_modelselected, estimated_costestimated_cost, estimated_tokensestimated_tokens, reasonfquot;Task tier: {tier.value}, selected {selected.name} quot; fquot;(tier {selected.tier}, ${selected.input_cost_per_mtok}/M tokens)quot;, tiertier, fallback_modelfallback, ) async def route_and_execute( self, task: str, context: Dict[str, Any], executor_fnNone ) -gt; Dict[str, Any]: quot;quot;quot; 完整的路由执行流程 分类 → 选模型 → 执行 → 降级重试 → 记录统计 quot;quot;quot; start_time time.time() self.stats.total_requests 1 # Step 1: 分类任务 tier await self.classifier.classify(task, context) self.stats.tier_distribution[tier.value] 1 # Step 2: 估算Token数简单用字符数/4估算 estimated_tokens max(len(task) // 4, 100) # Step 3: 选择模型 decision self.select_model(tier, estimated_tokens) self.stats.model_distribution[decision.selected_model.name] 1 # Step 4: 执行请求 result None try: if executor_fn: result await executor_fn(decision.selected_model, task) except Exception as e: # 主模型失败尝试降级 logger.warning( fquot;Model {decision.selected_model.name} failed: {e}, quot; fquot;trying fallback {decision.fallback_model}quot; ) if decision.fallback_model and executor_fn: self.stats.fallback_triggered 1 try: result await executor_fn(decision.fallback_model, task) decision.selected_model decision.fallback_model except Exception as e2: logger.error(fquot;Fallback also failed: {e2}quot;) raise # Step 5: 统计成本 actual_tokens estimated_tokens # 实际应该从API响应中获取 actual_cost ( actual_tokens * decision.selected_model.input_cost_per_mtok / 1_000_000 ) baseline_cost actual_tokens * self._baseline_cost_per_mtok / 1_000_000 self.stats.total_cost actual_cost self.stats.savings_vs_baseline (baseline_cost - actual_cost) elapsed time.time() - start_time return { quot;resultquot;: result, quot;model_usedquot;: decision.selected_model.name, quot;tierquot;: tier.value, quot;costquot;: actual_cost, quot;baseline_costquot;: baseline_cost, quot;savingsquot;: baseline_cost - actual_cost, quot;latency_msquot;: int(elapsed * 1000), quot;routing_reasonquot;: decision.reason, } def get_stats_report(self) -gt; str: quot;quot;quot;生成成本统计报告quot;quot;quot; total_baseline ( self.stats.total_requests * 500 # 平均估算500 tokens * self._baseline_cost_per_mtok / 1_000_000 ) savings_pct ( (self.stats.savings_vs_baseline / total_baseline * 100) if total_baseline gt; 0 else 0 ) report fquot;quot;quot; 成本路由统计报告 总请求数: {self.stats.total_requests}总成本: ${self.stats.total_cost:.4f}基线成本全用旗舰模型: ${total_baseline:.4f}节省: ${self.stats.savings_vs_baseline:.4f} ({savings_pct:.1f}%)降级触发次数: {self.stats.fallback_triggered}任务层级分布:for tier, count in sorted(self.stats.tier_distribution.items()):pct count / self.stats.total_requests * 100 if self.stats.total_requests else 0report f {tier}: {count} ({pct:.1f}%)\nreport quot;\n模型使用分布:\nquot; for model, count in sorted(self.stats.model_distribution.items()): pct count / self.stats.total_requests * 100 if self.stats.total_requests else 0 report fquot; {model}: {count} ({pct:.1f}%)\nquot; return report上面是核心路由引擎。光有引擎不够还得把它接到实际的Agent工作流里。四、接入Agent把路由器嵌入每一步执行很多团队的问题在于Agent的每一步执行都是硬编码的模型调用。你需要把它替换成一个统一的路由执行层。 agent_with_routing.py - 接入成本路由的Agent执行框架 import asynciofrom typing import Dict, Any, List, Optionalfrom dataclasses import dataclassdataclassclass AgentStep:Agent执行步骤定义name: strtask_template: str # 任务描述模板支持变量替换context: Dict[str, Any] # 步骤上下文depends_on: List[str] None # 依赖的前置步骤class RoutedAgent:带成本路由的Agent执行器每一步执行都经过成本路由器自动选择最经济的模型def __init__(self, cost_router, agent_config: Dict[str, Any]): self.router cost_router self.config agent_config self.step_results {} self.execution_log [] async def execute_step(self, step: AgentStep, variables: Dict[str, str]) -gt; Dict: quot;quot;quot; 执行单个步骤带路由 quot;quot;quot; # 渲染任务描述 task step.task_template.format(**variables) # 构建上下文传给路由器的决策信息 context { **step.context, quot;step_namequot;: step.name, quot;agent_namequot;: self.config.get(quot;namequot;, quot;unknownquot;), } # 定义实际执行函数对接你的模型调用逻辑 async def executor(model, prompt): return await self._call_model_api(model, prompt) # 路由 执行 result await self.router.route_and_execute(task, context, executor) # 记录执行日志 log_entry { quot;stepquot;: step.name, quot;modelquot;: result[quot;model_usedquot;], quot;tierquot;: result[quot;tierquot;], quot;costquot;: result[quot;costquot;], quot;savingsquot;: result[quot;savingsquot;], quot;latencyquot;: result[quot;latency_msquot;], } self.execution_log.append(log_entry) self.step_results[step.name] result return result async def execute_pipeline(self, steps: List[AgentStep], initial_variables: Dict[str, str]) -gt; Dict: quot;quot;quot; 执行完整的Agent pipeline 自动按依赖关系串行执行各步骤 quot;quot;quot; variables {**initial_variables} for step in steps: # 等待依赖步骤完成 if step.depends_on: for dep in step.depends_on: if dep not in self.step_results: raise RuntimeError(fquot;Dependency {dep} not completedquot;) # 执行当前步骤 result await self.execute_step(step, variables) # 将结果注入后续变量 if result.get(quot;resultquot;): variables[fquot;step_{step.name}_resultquot;] str(result[quot;resultquot;]) return { quot;final_resultsquot;: self.step_results, quot;execution_logquot;: self.execution_log, quot;total_costquot;: sum(e[quot;costquot;] for e in self.execution_log), quot;total_savingsquot;: sum(e[quot;savingsquot;] for e in self.execution_log), quot;stats_reportquot;: self.router.get_stats_report(), } async def _call_model_api(self, model, prompt: str) -gt; str: quot;quot;quot; 实际的模型API调用 这里需要替换成你真实的API调用逻辑 quot;quot;quot; # 伪代码示意 # response await openai_client.chat.completions.create( # modelmodel.name, # messages[{quot;rolequot;: quot;userquot;, quot;contentquot;: prompt}], # ) # return response.choices[0].message.content return fquot;[{model.name}] Response to: {prompt[:50]}...quot;接下来是一个完整的使用示例。拿一个典型的RAG问答场景来演示 demo_rag_agent.py - RAG问答Agent成本路由实战 async def main():# 1. 定义可用模型池按价格排序models [ModelConfig(namehaiku-5.5,provideranthropic,tier1,input_cost_per_mtok0.10,output_cost_per_mtok0.50,max_context200000,avg_latency_ms800,),ModelConfig(namegpt-6-luna,provideropenai,tier2,input_cost_per_mtok0.50,output_cost_per_mtok2.00,max_context128000,avg_latency_ms600,),ModelConfig(namesonnet-5.5,provideranthropic,tier4,input_cost_per_mtok3.00,output_cost_per_mtok15.00,max_context200000,avg_latency_ms1200,),ModelConfig(namegpt-6-astra,provideropenai,tier5,input_cost_per_mtok15.00,output_cost_per_mtok60.00,max_context256000,avg_latency_ms2500,),]# 2. 创建路由器 router CostRouter(models) # 3. 定义RAG Agent步骤 rag_steps [ AgentStep( namequot;intent_classifyquot;, task_templatequot;对用户问题进行意图分类{question}quot;, context{quot;priorityquot;: quot;normalquot;}, ), AgentStep( namequot;retrieve_docsquot;, context{}, ), AgentStep( namequot;relevance_filterquot;, task_templatequot;判断以下文档与问题的相关性{question}quot;, context{}, depends_on[quot;retrieve_docsquot;], ), AgentStep( namequot;generate_answerquot;, task_template( quot;基于以下检索结果回答用户问题。\nquot; quot;问题{question}\nquot; quot;检索结果{step_retrieve_docs_result}quot; ), context{quot;priorityquot;: quot;criticalquot;}, # 核心步骤强制用强模型 depends_on[quot;retrieve_docsquot;], ), AgentStep( namequot;format_outputquot;, task_templatequot;将以下回答格式化为最终输出{step_generate_answer_result}quot;, context{}, depends_on[quot;generate_answerquot;], ), ] # 4. 创建路由Agent并执行 agent RoutedAgent(router, {quot;namequot;: quot;rag_agent_v2quot;}) result await agent.execute_pipeline( rag_steps, {quot;questionquot;: quot;如何设计一个支持百万并发的WebSocket网关quot;} ) # 5. 输出结果 print(result[quot;stats_reportquot;]) print(fquot;\n总成本: ${result[total_cost]:.6f}quot;) print(fquot;总节省: ${result[total_savings]:.6f}quot;) for log in result[quot;execution_logquot;]: print(fquot; [{log[step]}] → {log[model]} quot; fquot;(tier: {log[tier]}, cost: ${log[cost]:.6f})quot;)ifname main:asyncio.run(main())跑完这段代码你会看到类似这样的输出 成本路由统计报告 总请求数: 5 总成本: $0.000182 基线成本全用旗舰模型: $0.000563 节省: $0.000381 (67.7%) 降级触发次数: 0任务层级分布:trivial: 1 (20.0%)simple: 2 (40.0%)moderate: 1 (20.0%)critical: 1 (20.0%)模型使用分布:haiku-5.5: 3 (60.0%)gpt-6-luna: 0 (0.0%)sonnet-5.5: 1 (20.0%)gpt-6-astra: 1 (20.0%)[intent_classify] → haiku-5.5 (tier: simple, cost: $0.000008)[retrieve_docs] → haiku-5.5 (tier: trivial, cost: $0.000004)[relevance_filter] → haiku-5.5 (tier: simple, cost: $0.000030)[generate_answer] → gpt-6-astra (tier: critical, cost: $0.000135)[format_output] → haiku-5.5 (tier: trivial, cost: $0.000005)看到了吗5个步骤3个用了最便宜的Haiku只有生成回答那一步用了旗舰模型。总成本降了67%但回答质量一点没降——因为最关键的推理步骤仍然是用最强模型完成的。五、生产环境的3个坑这套方案在我们自己的系统里跑了几个月踩过几个坑分享出来帮大家避一下。坑1分类准确率没有你想象的高规则分类的准确率大概在85%左右。这意味着15%的任务会被分错级别。最危险的错误是一个需要深度推理的任务被分到了simple级别结果用了轻量模型出来的回答驴唇不对马嘴。我们的解决方案加一个质量兜底机制。轻量模型输出后用一个更便宜的模型做一次质量打分。如果分数低于阈值自动回退到更强的模型重新执行。async def execute_with_quality_gate(self, task, tier, selected_model): 带质量兜底的执行 result await self._call_model(selected_model, task)# 只有非CRITICAL任务才做质量检查 if tier ! TaskTier.CRITICAL: quality_score await self._assess_quality(task, result) if quality_score lt; 0.7: # 质量不达标升级模型重试 stronger_model self._get_stronger_model(selected_model) if stronger_model: logger.info( fquot;Quality gate failed ({quality_score:.2f}), quot; fquot;upgrading to {stronger_model.name}quot; ) result await self._call_model(stronger_model, task) selected_model stronger_model return result, selected_model加了质量兜底以后整体准确率从85%提升到了97%以上而且回退的成本远低于一开始就用旗舰模型的成本。坑2缓存比路由更省钱我们后来发现有大量重复或高度相似的任务请求。比如提取关键词这种操作在同一个文档上可能被调了好几次。加一层语义缓存命中缓存就直接返回结果连轻量模型都不用调。class SemanticCache: 语义缓存 - 相似问题直接复用历史结果def __init__(self, embedding_model, threshold0.92): self.embedding_model embedding_model self.threshold threshold self.cache {} # embedding -gt; (result, timestamp) async def get(self, task: str): task_embedding await self._embed(task) for cached_embedding, (result, ts) in self.cache.items(): similarity self._cosine_similarity(task_embedding, cached_embedding) if similarity gt; self.threshold: return result # 缓存命中 return None async def put(self, task: str, result: str): task_embedding await self._embed(task) self.cache[task_embedding] (result, time.time())我们实测加了这个缓存以后整体API调用量下降了约35%。也就是说缓存省的钱可能比路由本身还多。坑3别忽略延迟的影响路由不只是省钱还影响响应速度。轻量模型延迟低、旗舰模型延迟高。如果你的Agent有5个步骤全用旗舰模型总延迟可能是串行的5×2.5秒12.5秒。但如果前3步用轻量模型每个0.8秒后2步用旗舰模型总延迟就变成3×0.82×2.57.4秒快了40%的同时还省了钱。但反过来如果路由分类本身就要花200ms去做模型推理而分类省下的钱只有$0.00001那就得不偿失了。所以规则分类优先模型分类兜底这个顺序不能反。六、效果验证A/B测试怎么设计上线成本路由以后怎么证明它真的省钱又不降质量答案是做A/B测试。class RoutingABTest: 路由A/B测试框架def __init__(self, router_with_cost_opt, router_baseline): self.cost_opt router_with_cost_opt self.baseline router_baseline # 全用旗舰模型的对照组 self.test_cases [] self.results {quot;cost_optquot;: [], quot;baselinequot;: []} def add_test_case(self, question: str, expected_answer: str): self.test_cases.append({ quot;questionquot;: question, quot;expectedquot;: expected_answer, }) async def run_test(self): quot;quot;quot;运行A/B测试quot;quot;quot; for case in self.test_cases: # 实验组成本路由 cost_result await self.cost_opt.route_and_execute( case[quot;questionquot;], {} ) # 对照组全旗舰 base_result await self.baseline.route_and_execute( case[quot;questionquot;], {quot;priorityquot;: quot;criticalquot;} ) # 质量评估用另一个模型做裁判 quality_score await self._judge_answers( case[quot;expectedquot;], cost_result[quot;resultquot;], base_result[quot;resultquot;], ) self.results[quot;cost_optquot;].append({ quot;costquot;: cost_result[quot;costquot;], quot;qualityquot;: quality_score[quot;cost_opt_scorequot;], }) self.results[quot;baselinequot;].append({ quot;costquot;: base_result[quot;costquot;], quot;qualityquot;: quality_score[quot;baseline_scorequot;], }) return self._generate_report()我们在内部跑了200个测试用例。结论成本降低52%-71%取决于任务分布回答质量评分差异2%用GPT-4级别模型做裁判打分平均延迟降低34%这数据基本能说明问题钱省了大半质量几乎没降速度还变快了。在黑箭科技内部做舆安系统的AI分析模块时我们也用了类似的分层调度策略。舆情分析里面大量任务是判断一条评论的情感倾向、提取关键词这种简单操作只有生成舆情报告这种步骤才需要强推理。通过分层路由整体API成本大概降了六成左右。七、选型建议最后聊一下几个具体的选型问题这是后台问的比较多的。1. 模型池怎么配至少3个层级旗舰、均衡、轻量。如果预算紧2个也行旗舰轻量但效果会打折扣。如果追求极致可以加一个本地模型层跑Qwen3.5-9B之类的开源模型把最简单的任务也省掉API费用。2. 分类器用什么模型用你模型池里最便宜的那个。分类任务本身不需要强推理能力用旗舰模型做分类纯属浪费。实测Haiku级别的模型做分类准确率已经够用。3. 质量兜底的阈值怎么设建议从0.7开始然后根据实际数据调整。设太低等于不设设太高会导致大量回退反而更贵。我们的经验是0.65-0.75之间比较合适。4. 要不要用专门的Router模型OpenAI和Anthropic都提供了类似模型路由的能力。但实测下来自己写路由器更灵活因为你可以根据自己的业务场景定制分类规则而且不用额外付Router的费用。代码量也不大核心就300来行。总结回到最开始的问题Agent的API费用为什么那么高不是因为模型太贵而是因为你在用错误的方式调用模型。就像你不需要CEO去干快递分拣的活一样你也不需要旗舰模型去干关键词提取的活。300行代码核心就做三件事给任务分级、给模型定价、让系统自己选。这套方案不是银弹。如果你的Agent大部分步骤都需要深度推理那路由能省的空间确实有限。但对于大部分典型场景——RAG、客服、数据分析、内容生成——里面都有大量粗活这些粗活根本不需要最贵的模型。与其抱怨API太贵不如先看看你的Agent到底在哪些地方浪费了钱。你在实际项目中是怎么控制Agent API成本的有没有遇到过分类器把复杂任务分给轻量模型翻车的情况欢迎评论区聊聊互相避坑。