大模型产品选型先算任务成本

大模型产品选型先算任务成本

上季度准备给公司的客服与工单系统做 AI 大模型升级时，团队在选型阶段犯过一个典型错误。当时大家直接参考了市场上几款标杆竞品的选型方案，看到某大厂宣称其最新的 70B 模型在 MMLU 和 GSM8K 榜单上得分第一，支持 128K 超长上下文，而且单次 Token 价格打到了极低的折后价，便不假思索地决定将全线业务切换过去。

然而上线真实打压测试后，现实给了大家狠狠一棒：榜单高分的模型在处理公司内部特定格式的 XML 工单时，Tool Calling 字段经常丢失括号；128K 的上下文在超过 16K 后，首字延迟（TTFT）飙升到了 6 秒以上，用户端等待超时报错频发。算完账才发现，由于大量的失败重试与长上下文无效消耗，综合 ROI（投资回报率）非但没有提升，反而比原来用轻量级 8B 模型配 RAG 的方案贵了 3 倍。

在 LLM 应用产品化落地的过程中，只看宣传参数和公开 Benchmark 选型，属于最昂贵的自嗨。竞品的技术栈有其特定的业务假设，盲目照搬必然踩坑。

1. 竞品拆解：什么可以借鉴，什么千万不能照搬？

在进行大模型选型与竞品分析时，必须把“工程设计模式”与“模型参数体量”彻底解耦。

可借鉴的确定性工程设计

1. 分级路由架构（Model Router）：顶尖的 AI 产品绝对不会用一个昂贵的大模型处理所有请求。简单的意图识别、文本分类用开源小模型（8B 甚至 3B）在本地处理；复杂的逻辑推理、代码生成才路由给昂贵的高性能模型。
2. 结构化重试与 Schema 修复（Output Sanitizer）：优秀竞品在调用 Tool 时，前端/中间件都挂有强类型校验器，模型一旦输出格式错误，自动带着 Error Context 进行 1 轮轻量重试，而不是直接抛异常。
3. 上下文剪枝与 KV Cache 友好设计：竞品如何将 System Prompt 静态化以最大化命中 API 供应商的 KV Cache 优惠，从而降低 50% 以上的 Prompt 费用。

千万不可照搬的陷阱

1. 照搬公开榜单分数的模型：MMLU、HumanEval 等榜单存在严重的刷榜与数据污染现象。榜单高分不代表能准确解析你公司的 DB Schema 或行业术语。
2. 盲目追求超长 Context Window：很多大模型宣传支持 1M Token 上下文，但随着 Context 变长，模型的“大海捞针（Needle In A Haystack）”召回率会断崖式下跌，且首字延迟（TTFT）呈二次方增加。
3. 忽略 API 供应商的真实吞吐上限（TPM/RPM）：竞品可能拥有私有化的 GPU 独占集群，而你使用公有云共享 Endpoint 时，经常遇到高并发下的 Rate Limit（429 报错）。

2. 真实 ROI 评估模型：从 Task 单位成本出发

评估大模型应用的 ROI，不能看“每百万 Token 多少钱”，而要看完成单个有效业务任务（Task）的综合成本。

假设业务场景是“处理一份客户退款工单”：

$$\text{Cost per Task} = \frac{\text{Prompt Tokens} \times \text{Input Price} + \text{Completion Tokens} \times \text{Output Price}}{\text{Task Success Rate}} + \text{Retry Overhead}$$

如果模型 A 价格很便宜，但结构化抽取成功率只有 70%，导致 30% 的请求需要重试甚至人工介入；而模型 B 价格贵一倍，但 Task 成功率 99%。计算下来，模型 B 的真实单位 Task 成本反而远低于模型 A。

为了量化这一指标，我们编写了一套自动化评估与压测脚手架，用于测量不同模型在真实业务 Payload 下的TTFT（首字延迟）、TPS（生成吞吐量）、Task 成功率与实际扣费：

import asyncio import time import json import logging from typing import List, Dict, Any import aiohttp logger = logging.getLogger("llm.roi_evaluator") class LLMModelEvaluator: def __init__(self, api_base: str, api_key: str, model_name: str): self.api_base = api_base self.api_key = api_key self.model_name = model_name async def benchmark_single_request( self, prompt: str, expected_schema_keys: List[str] ) -> Dict[str, Any]: """ 测量单次 API 调用的真实延迟、首字耗时(TTFT)、吞吐量与结构化正确性 """ headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "model": self.model_name, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "stream": True # 开启流式以准确测量 TTFT } start_time = time.time() ttft = 0.0 first_token_received = False full_content = "" completion_tokens = 0 try: async with aiohttp.ClientSession() as session: async with session.post( f"{self.api_base}/chat/completions", headers=headers, json=payload ) as response: if response.status != 200: return {"success": False, "error": f"HTTP {response.status}"} async for chunk in response.content: if not chunk: continue chunk_str = chunk.decode("utf-8") if not first_token_received: ttft = time.time() - start_time first_token_received = True # 简单解析 SSE 增量数据 lines = chunk_str.split("\n") for line in lines: if line.startswith("data: ") and not line.endswith("[DONE]"): try: data = json.loads(line[6:]) delta = data["choices"][0]["delta"].get("content", "") full_content += delta completion_tokens += 1 except Exception: pass total_latency = time.time() - start_time tps = completion_tokens / (total_latency - ttft) if (total_latency - ttft) > 0 else 0 # 结构化输出强校验 is_valid_json = False has_all_keys = False try: parsed_json = json.loads(full_content) is_valid_json = True has_all_keys = all(k in parsed_json for k in expected_schema_keys) except Exception: pass task_success = is_valid_json and has_all_keys return { "success": True, "model": self.model_name, "ttft_seconds": round(ttft, 3), "total_latency_seconds": round(total_latency, 3), "tps": round(tps, 2), "completion_tokens": completion_tokens, "task_success": task_success, "output_preview": full_content[:100] } except Exception as e: return {"success": False, "error": str(e)} async def run_batch_evaluation( self, test_dataset: List[Dict[str, Any]], concurrency: int = 5 ) -> Dict[str, Any]: """ 高并发批量基准测试，计算综合 Task 成功率与 ROI 性能指标 """ semaphore = asyncio.Semaphore(concurrency) async def worker(item): async with semaphore: return await self.benchmark_single_request( item["prompt"], item["expected_keys"] ) tasks = [worker(item) for item in test_dataset] results = await asyncio.gather(*tasks) valid_results = [r for r in results if r.get("success")] if not valid_results: return {"error": "全量请求失败"} avg_ttft = sum(r["ttft_seconds"] for r in valid_results) / len(valid_results) avg_tps = sum(r["tps"] for r in valid_results) / len(valid_results) success_rate = sum(1 for r in valid_results if r["task_success"]) / len(valid_results) return { "model_name": self.model_name, "total_requests": len(test_dataset), "avg_ttft_sec": round(avg_ttft, 3), "avg_tps": round(avg_tps, 2), "task_success_rate": f"{success_rate * 100:.1f}%", "effective_roi_score": round((success_rate * avg_tps) / (avg_ttft + 0.1), 2) }

3. 多模型混合分流（Router）架构的落地实践

在完成真实测评后，我们发现没有一个单体模型能在“速度、精度、价格”上全胜。

最终落地的生产级架构是动态 Router 分流模式：

• 前置意图过滤层（Routing Layer）：采用轻量级开源模型（如 Qwen-2.5-7B 或 Llama-3-8B），负责意图分类、抽取简单参数。90% 的常规请求在这个阶段被快速解决，TTFT 控制在 300ms 以内，成本极低。
• 复杂推理兜底层（Fallback Layer）：仅当路由层标记为“复杂异常逻辑”或第一轮 Tool Calling 校验失败时，才将 Context 升级投递给高阶模型（如 Claude-3.5-Sonnet 或 GPT-4o）。

这种组合架构既保证了 99%+ 的业务成功率，又将单次 Task 的平均 Token 费用降低了70% 以上。

4. 模型选型与 ROI 决策检查清单

在最终定型大模型应用选型时，团队应当对着下表逐一核对，切忌被单一参数遮蔽双眼：

总结：

大模型产品化的关键不是去抢最新的“参数王者”，而是建立一套属于你自家业务的自动化 Evaluation 机制与 ROI 成本计算脚手架。用工程的确定性（Router 路由 + 结构化校验 + 降级兜底）去消化模型的局限性，才是技术选型不踩坑的根本保障。