企业官网建设流程全解析

问题
模型名写死后，升级、下线、限流都会触发业务改动

核心
业务只依赖稳定路由ID，具体模型池交给平台维护

收益
模型能切换，结果仍可验证、可追溯、可核算

先看结论
智能路由最大的工程价值，不是“保证每次自动选到最聪明的模型”，而是把模型池、切换和故障策略从业务发布周期里拿走。业务侧仍然要对输入数据、结果契约、解析失败、证据校验和成本账本负责。

业务代码写死模型

Python/Java配置

调试直观、行为确定

换模型要改代码；模型下线/涨价会触发业务发版

业务维护任务-模型映射

客户端配置表

不同任务可用不同模型

映射表继续承载模型生命周期；故障切换逻辑分散

稳定路由ID + 平台模型池

平台侧

业务与具体模型解耦；策略可独立调整

必须加强可观测性和结构化结果校验

模型选择

模型池、优先级、效果/成本/平衡策略

定义业务能力边界，不把路由当成质量保证

故障处理

模型节点健康、平台级切换与降级

请求超时、失败分类、是否客户端重试

统一调用

稳定路由任务ID、OpenAI兼容入口

API Key、请求参数、Prompt版本

可观测性

调用统计、平台账单/用量

评论ID、批次ID、实际模型、原始响应、解析状态

业务正确性

不直接负责

Schema校验、证据核对、人工抽查、评测数据集

一个容易被夸大的结论
“用了智能路由”不等于“平台一定自动选到业务最优模型”，也不等于“必然更便宜、更快”。真正可以验证的是：路由任务ID稳定、平台模型池可以调整、实际请求能够继续执行；效果、费用和时延都要用自己的数据与账单单独测。

id

string

必须与输入评论ID一致，不能漏、重复或新增

sentiment

enum

仅允许 positive / negative / neutral

tags

string[]

去重后最多6个，禁止用逗号字符串冒充数组

evidence

string

必须能在原评论中定位到连续原文片段

summary_zh

string

仅差评触发；简短、可读，不把推断写成确定事实

SYSTEM_PROMPT = """你是商品评论分析器。只返回一个 JSON 对象，不要输出 Markdown。
字段：
- id: 原样返回输入 id
- sentiment: positive / negative / neutral 三选一
- tags: 1~6 个商品属性或问题主题
- evidence: 支撑情感判断的原文连续片段
若证据不足，不要编造。
"""

def build_messages(review_id: str, text: str) -> list[dict]:
return [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": json.dumps({"id": review_id, "text": text}, ensure_ascii=False)},
]

import json, os, time, random, urllib.request, urllib.error

ENDPOINT = "https://maas-api.lanyun.net/v1/chat/completions"
API_KEY = os.environ["LANYUN_API_KEY"]
REVIEW_ROUTE = os.environ["LANYUN_REVIEW_ROUTE"]

def call_route(messages, *, route_id=REVIEW_ROUTE, timeout=30):
body = {
"model": route_id,
"messages": messages,
"response_format": {"type": "json_object"},
"temperature": 0,
"max_tokens": 800,
"stream": False,
}
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
ENDPOINT, data=data, method="POST",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
)
started = time.perf_counter()
with urllib.request.urlopen(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode("utf-8"))
payload["_latency_ms"] = round((time.perf_counter() - started) * 1000, 1)
return payload

为什么不在调用函数里塞TASK_MODEL_MAP
一旦客户端重新出现“任务 → 具体模型”的映射，模型生命周期又回到了业务代码。应用可以按能力保存多个稳定路由ID，但不应该关心每个路由后面当前排了哪些具体模型。

import json, re
VALID_SENTIMENTS = {"positive", "negative", "neutral"}

def extract_json_object(text: str) -> dict:
text = text.strip()
text = re.sub(r"^```(?:json)?\\s*", "", text, flags=re.I)
text = re.sub(r"\\s*```$", "", text)
decoder = json.JSONDecoder()
for pos, ch in enumerate(text):
if ch != "{":
continue
try:
obj, _ = decoder.raw_decode(text[pos:])
if isinstance(obj, dict):
return obj
except json.JSONDecodeError:
pass
raise ValueError("找不到完整 JSON 对象")

def validate_result(obj: dict, review_id: str, review_text: str) -> dict:
if obj.get("id") != review_id:
raise ValueError("评论ID不一致")
sentiment = str(obj.get("sentiment", "")).strip().lower().strip("。.!！?？")
if sentiment not in VALID_SENTIMENTS:
raise ValueError("情感枚举非法")
tags = obj.get("tags")
if not isinstance(tags, list) or not all(isinstance(v, str) for v in tags):
raise ValueError("tags 必须是字符串数组")
evidence = str(obj.get("evidence", "")).strip()
if evidence and evidence not in review_text:
raise ValueError("evidence 不是原文连续片段")
return {
"id": review_id,
"sentiment": sentiment,
"tags": list(dict.fromkeys(v.strip() for v in tags if v.strip()))[:6],
"evidence": evidence,
}

RETRYABLE_HTTP = {429, 502, 503, 504}

def call_with_retry(messages, max_attempts=2):
for attempt in range(1, max_attempts + 1):
try:
return call_route(messages)
except urllib.error.HTTPError as exc:
if exc.code not in RETRYABLE_HTTP or attempt == max_attempts:
raise
retry_after = exc.headers.get("Retry-After")
delay = float(retry_after) if retry_after else min(8.0, 0.8 * (2 ** (attempt - 1)) + random.random())
time.sleep(delay)
except (TimeoutError, urllib.error.URLError):
if attempt == max_attempts:
raise
time.sleep(min(8.0, 0.8 * (2 ** (attempt - 1)) + random.random()))

HTTP 429 / 502 / 503 / 504

少量、有退避

优先读取 Retry-After；限制最大尝试次数

超时/临时网络错误

少量

记录原请求；避免整批同步重试

JSON截断

通常不直接重试

保存原始响应；可进入单独失败队列

字段类型非法

不自动重试

视为模型/Prompt兼容问题，先分析失败模式

evidence 不在原文

不自动重试

标记语义失败，不能自动篡改证据

route_id

确认调用的能力入口，支持按路由统计

actual_model

判断质量/时延变化是否与实际模型有关；若响应可提供则必须记录

prompt_hash

避免“同一实验”实际上使用了不同Prompt

input_hash

确认数据没有在实验间悄悄变化

usage

Token成本分析的基础，但最终费用仍以平台账单为准

raw_response

解析器升级后可重放；也能还原失败现场

parse_status

区分HTTP成功、JSON成功和业务语义成功

不要过度解读
这项测试能证明“模型池配置变化可以通过同一路由ID影响真实请求”，不能单独证明“新模型更优”“路由更快”或“自动故障切换已验证”。超时/限流降级要单独做故障注入或压力测试。

请求完成率

成功收到可处理响应的请求数 / 总请求数

不能代表模型理解正确

结构化有效率

通过JSON + Schema校验的记录数 / 总记录数

不能代表情感标签正确

情感标签一致率

与人工/公开标签一致的记录数 / 可评记录数

不能代表摘要忠实

证据精确匹配率

evidence 为对应原文连续子串的比例

不能证明证据足以支撑判断

摘要人工通过率

人工抽检忠实、无夸大、可读

不能由自动标签指标替代

直接 json.loads

5,758 / 10,000

57.58%

代码围栏、前后说明都会直接失败

防御性解析 + Schema校验

8,967 / 10,000

89.67%

接受可归一格式；528条截断JSON、505条语义非法输出明确拒绝

为什么89.67%不是越高越好
解析器的目标不是“把所有输出都修成功”。如果为了提高通过率而自动补齐截断JSON、把字符串tags强行切成数组、替模型重写evidence，就会把模型错误隐藏成应用成功。可靠系统宁可保留明确失败，也不要制造伪成功。

1

把路由当“质量保证”

路由解决调度，不替代业务评测。

2

重新在客户端维护模型名

只保存稳定路由ID；具体模型池放平台。

3

一个路由承担所有任务

按结构化抽取、摘要等能力边界拆路由。

4

把 HTTP 200 当成功

还要过 JSON、Schema、证据和业务校验。

5

解析失败就自动“修答案”

保留原始响应，失败队列单独重跑。

6

盲目高并发

从小并发开始，根据429与P95时延调节。

7

无限自动重试

生成请求可能重复计费；设置上限和退避。

8

不记录实际模型

质量波动时无法解释路由变化。

9

本地估算当最终账单

本地仅用于预算；最终费用以平台结算为准。

10

公开老数据当生产代表

公开数据只验流程；必须补业务真实分布。

11

摘要当事实

摘要可能含合理推断；核心结论必须回到原文证据。

12

API Key写进脚本

使用环境变量/密钥管理，日志严禁打印密钥。

review-router-app/
├── app/
│ ├── client.py # 蓝耘路由客户端
│ ├── prompts.py # Prompt版本化
│ ├── parser.py # JSON与Schema校验
│ ├── batch.py # 有界并发、失败队列
│ └── metrics.py # 时延、Token、失败率
├── data/
│ ├── input.jsonl # 固定输入与哈希
│ └── eval.jsonl # 人工/公开评测标签
├── outputs/
│ └── <batch_id>/ # 原始响应、结果、错误、账本
├── tests/
│ ├── test_parser.py
│ └── fixtures/
└── .env.example # 只写变量名，不放真实Key

API Key安全

源码、日志、报告中均不出现真实密钥

路由切换

同一路由ID下，平台策略变化可通过小样本验证；应用无需发版

结构化有效率

按业务数据统计；失败必须可追溯到原始响应

P95时延

按路由和实际模型拆分观察，不能只看均值

429/5xx比例

持续监控；达到阈值主动降并发或暂停批次

预算

批次开始前估算上限，运行中按usage滚动累计，最终对账平台账单

可重放性

输入、Prompt版本、请求参数、原始响应均可定位