大模型产品别只看 Demo：用任务分流和单次贡献毛利算 ROI
大模型产品别只看 Demo：用任务分流和单次贡献毛利算 ROI
Demo 能回答几道预设问题，不等于这项能力有正向 ROI。产品化要算每个任务的成功率、人工接管、Token 费用、响应时间和能归因的收入，不是只看回答“像不像人”。
先把确定性查询交给规则或数据库，把模型留给意图理解和非结构化转换。这个分流是否划算，再用灰度样本验证。
1. 先建立任务分流与成本口径
从脱敏请求或按产品范围构造的任务集开始，先定义分类规则，再让两人抽样复核分类一致性。样本量由流量分布和置信要求决定，不预设为某个整齐数字。每类任务记录这些字段：
| 场景类型 | 流量占比 | 平均 Token 消耗/次 | 调用成功率 | 业务真实价值 |
|---|---|---|---|---|
| A. 简单基础指标查询 | 从分类结果汇总 | 模型 usage | 与 SQL 或规则答案对照 | 是否值得走确定性路径 |
| B. 复杂多维组合分析 | 从分类结果汇总 | 模型 usage | 固定任务集 + 人工复核 | 是否需要模型与复核 |
| C. 闲聊与无关提问 | 从分类结果汇总 | 模型 usage | 按产品范围判定 | 接收、提示还是拒绝 |
| D. 格式化数据导出请求 | 从分类结果汇总 | 模型 usage | 与传统 API 对照 | 哪条路径更易维护 |
排查结果表明：
- 简单查询单独统计。如果 SQL 或规则引擎能稳定完成，就比较两条路径的耗时、正确性与维护成本，再决定是否绕过模型。
- 复杂多维分析记录人工复核。模型处理嵌套表结构时，成功率和重试次数要从固定任务集计算，不提前写成某个比例。
只有在对照数据表明规则路径的质量不低于模型路径，同时费用或延迟更低时，才能把前置分流列为候选改动。ROI 不达标也可能来自模型选型、人工复核或收益口径，不能只归因于路由。
2. 基于 MVP 范围切分的混合分流架构
针对全能型架构的缺陷，需要对 MVP 的功能范围进行重新切分。
切分的核心原则是：解耦确定性逻辑与非确定性 AI 能力，将大模型限定在其擅长的“意图理解与非结构化文本转换”边界内，而将结构化数据计算与基础查询全量交由确定性引擎处理。
重新切分后的 MVP 应由入口限流、核心服务和必要的观测能力组成，先保障关键请求可控。
在该 MVP 架构中：
- 剔除了非必要的全自动复杂报表生成；
- 增加了前置静态规则分流闸门，直接过滤简单查询与无效闲聊；
- 聚焦于“自然语言转标准化查询参数”确切场景。
这种切分让每条路径的调用次数和费用更容易计量，但预算是否满足仍取决于真实流量和模型价格。
3. 混合路由与 ROI 可观测性系统代码实现
为了支撑上述 MVP 的范围切分，在工程层实现一套“混合路由与 ROI 实时计算网关”。
基于 Python 实现的实现示例如下：
import re import time from typing import Dict, Any, Tuple from pydantic import BaseModel class UserRequest(BaseModel): user_id: str query: str class RoutingResult(BaseModel): execution_type: str # RULE_ENGINE, TEMPLATE_SQL, LLM_AGENT response_data: str token_cost_usd: float latency_ms: float class ROIManagedGateway: """带 ROI 成本治理与 MVP 范围切分的智能网关""" # 固定的 Token 单价 (单位: USD/1K token) PROMPT_COST_PER_1K = 0.0015 COMPLETION_COST_PER_1K = 0.0020 def __init__(self, llm_client, db_pool): self.llm_client = llm_client self.db_pool = db_pool # 预编译正则规则，用于确定性拦截 self.simple_query_pattern = re.compile(r"^查(下|一下)?(?P<name>[\u4e00-\u9fa5]{2,4})的销售额$") def _try_rule_routing(self, query: str) -> Tuple[bool, str]: """确定性规则路由防线：0 Token 解决简单高频场景""" match = self.simple_query_pattern.match(query.strip()) if match: name = match.group("name") # 直接走 SQL 模板查询，不经过 LLM return True, f"【确定性引擎结果】销售员 {name} 上月销售额为 128,500 元。" # 拦截闲聊 if len(query) < 4 or query in ["你好", "在吗", "谢谢"]: return True, "您好！我是销售数据助手，请告诉我您想查询的具体业务数据。" return False, "" def process_request(self, req: UserRequest) -> RoutingResult: start_time = time.perf_counter() # 1. 第一层：确定性规则分流 (MVP 降本核心) is_handled_by_rule, rule_resp = self._try_rule_routing(req.query) if is_handled_by_rule: elapsed_ms = (time.perf_counter() - start_time) * 1000 return RoutingResult( execution_type="RULE_ENGINE", response_data=rule_resp, token_cost_usd=0.0, # 0 成本 latency_ms=elapsed_ms ) # 2. 第二层：收窄 MVP 后的精简 LLM 调用 # 严格限制 Prompt 范围，只让 LLM 做 Intent Parsing (意图解析)，绝不让其直接生成代码 system_prompt = "你是一个意图解析器，将用户提问转为 JSON: {\"metric\": \"sales\", \"time\": \"Q3\"}" # 模拟 LLM 调用 llm_start = time.perf_counter() prompt_tokens = len(system_prompt + req.query) // 4 completion_text = "{\"metric\": \"sales\", \"time\": \"Q3\"}" completion_tokens = len(completion_text) // 4 # 计算本次 LLM 调用的硬成本 cost_usd = (prompt_tokens / 1000 * self.PROMPT_COST_PER_1K) + \ (completion_tokens / 1000 * self.COMPLETION_COST_PER_1K) elapsed_ms = (time.perf_counter() - start_time) * 1000 return RoutingResult( execution_type="LLM_AGENT", response_data="【LLM 解析后确定性执行】指标: 销售额, 时间: Q3", token_cost_usd=cost_usd, latency_ms=elapsed_ms )代码的关键逻辑在于：通过_try_rule_routing方法将非必要 LLM 请求拦截在零成本的确定性计算路径上。对需要大模型处理的请求，仅输出确定性的结构化 JSON，避免生成冗余文本。
4. 用业务样本验证 ROI 假设
在灰度范围内固定统计窗口，记录任务成功、人工接管、Token 费用、响应时间和单位任务收入。窗口长度应覆盖业务周期，不照搬固定天数。
重新统计的运行指标对比表格如下：
| 评估维度 | 采集来源 | 归因限制 |
|---|---|---|
| Token 费用与响应时间 | usage、Trace、当前价格表 | 固定任务构成、模型和重试策略 |
| 任务成功与人工接管 | 任务判定、工单记录 | 使用一致的成功定义与统计窗口 |
| 单位任务贡献毛利 | 可归因收入、调用及人工成本 | 不把渠道、季节等同期变化归给分流 |
表中费用、延迟和成功率应由同一统计窗口计算。用户活跃变化还会受功能、渠道与季节影响，不能仅凭时间相邻就归因于模型分流；ROI 只纳入能追溯的成本和收益。
5. 大模型产品化 ROI 治理规则总结
评估大模型产品是否值得继续投入，可以按下面三步做：
- 建立覆盖异常路径的评估集：区别演示环境的预设问法，同时覆盖并发、超时、无答案和人工接管。
- 确切切分 MVP 功能边界：将大模型限定在意图解析与非结构化文本转换场景，由确定性引擎承担业务计算。
- 建立 Token 成本与响应延时监控防线：在代码层面监控每次 LLM 调用的 ROI 指标，设置成本告警与熔断阈值。
最终是否保留某条模型链路，依据同一窗口内的任务质量、人工成本、调用费用和可归因收益决定。