适合业务的模型不会永远不变:更快的模型会出现,更低成本的版本会上线,语音交互也可能从附加功能变成核心入口。真正困难的不是接入某个模型,而是让团队以后更换模型时,不必重写智能体、工具调用和业务流程。
一个更稳妥的方向,是把模型选择、语音通道和质量评估从智能体业务逻辑中拆开。这样,新模型进入系统时只是增加一个候选项,而不是触发一次架构重建。
模型应该是可替换依赖,而不是业务逻辑的一部分
很多智能体原型会直接在业务代码里写入模型名称、服务地址、鉴权方式和提示词。例如,客服流程调用一个高质量模型,摘要流程又复制一套请求代码。短期看开发很快,长期却会形成几个问题:
- 更换模型时需要修改多个服务;
- 不同团队各自维护重试、超时和错误处理;
- 无法统一记录延迟、成本和任务成功率;
- 模型发生限流或故障时,没有清晰的降级路径;
- 语音、网页和后台任务难以复用同一套智能体能力。
更好的边界是让业务层只表达任务意图,例如 fast、quality 或 reasoning,由模型网关把意图映射到具体模型。模型版本和服务商配置属于运行时决策,不应散落在订单、客服或知识库代码中。
调用链可以保持为:
Web / Voice / Batch
|
v
Agent API -> Tool orchestration -> Model gateway -> Model A / Model B
| |
+------ business result <------+
|
v
Evaluation and telemetry
这里的关键不是增加更多抽象层,而是稳定边界:业务代码依赖统一输入输出,模型网关负责路由、超时、回退和观测。
一个可运行的模型路由网关
下面是一个可以直接改造的最小示例。它假设候选模型提供兼容 /chat/completions 的 HTTP 接口;实际接入托管模型服务时,需要按服务要求调整鉴权头、API 路径和请求字段。
创建 requirements.txt:
fastapi==0.115.0
uvicorn==0.30.6
httpx==0.27.2
创建 app.py:
import os
import time
from typing import Literal
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
MODELS = {
"fast": {
"endpoint": os.environ.get("FAST_MODEL_ENDPOINT", "http://localhost:8001"),
"api_key": os.environ.get("FAST_MODEL_API_KEY", "dev-key"),
"model": os.environ.get("FAST_MODEL_NAME", "fast-model"),
},
"quality": {
"endpoint": os.environ.get("QUALITY_MODEL_ENDPOINT", "http://localhost:8002"),
"api_key": os.environ.get("QUALITY_MODEL_API_KEY", "dev-key"),
"model": os.environ.get("QUALITY_MODEL_NAME", "quality-model"),
},
}
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: str
class AgentRequest(BaseModel):
task: Literal["realtime", "general", "complex"] = "general"
messages: list[Message]
def route(task: str) -> list[str]:
if task == "realtime":
return ["fast", "quality"]
if task == "complex":
return ["quality", "fast"]
return ["fast", "quality"]
async def call_model(alias: str, request: AgentRequest) -> dict:
config = MODELS[alias]
url = config["endpoint"].rstrip("/") + "/chat/completions"
payload = {
"model": config["model"],
"messages": [message.model_dump() for message in request.messages],
"temperature": 0.2,
}
headers = {
"Authorization": f"Bearer {config['api_key']}",
"Content-Type": "application/json",
}
started = time.perf_counter()
async with httpx.AsyncClient(timeout=20.0) as client:
response = await client.post(url, json=payload, headers=headers)
response.raise_for_status()
elapsed_ms = round((time.perf_counter() - started) * 1000)
return {
"model_alias": alias,
"model_name": config["model"],
"latency_ms": elapsed_ms,
"result": response.json(),
}
@app.post("/agent")
async def run_agent(request: AgentRequest):
errors = []
for alias in route(request.task):
try:
return await call_model(alias, request)
except (httpx.TimeoutException, httpx.HTTPError) as exc:
errors.append({"model_alias": alias, "error": str(exc)})
raise HTTPException(
status_code=503,
detail={"message": "All model candidates failed", "attempts": errors},
)
安装依赖并启动:
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
export FAST_MODEL_ENDPOINT="https://your-fast-model.example.com"
export FAST_MODEL_API_KEY="replace-me"
export FAST_MODEL_NAME="fast-model-version"
export QUALITY_MODEL_ENDPOINT="https://your-quality-model.example.com"
export QUALITY_MODEL_API_KEY="replace-me"
export QUALITY_MODEL_NAME="quality-model-version"
uvicorn app:app --host 0.0.0.0 --port 8080
发送一个请求:
curl -s http://localhost:8080/agent \
-H 'Content-Type: application/json' \
-d '{
"task": "complex",
"messages": [
{"role": "system", "content": "你是企业支持助手,回答必须简洁且可执行。"},
{"role": "user", "content": "请给出数据库变更上线前的检查清单。"}
]
}'
这个示例还不是完整生产网关,但已经隔离了三件容易变化的内容:模型名称、服务地址和路由规则。后续可加入并发限制、熔断器、流式输出、令牌统计,以及基于租户或数据敏感等级的路由策略。
语音智能体不只是给文本接口加一层转写
语音场景通常包含语音识别、智能体推理和语音合成三个阶段。如果按顺序等待每一步全部完成,用户会明显感到停顿。工程设计应围绕端到端延迟,而不是只看模型生成速度。
一条更实用的实时链路是:
麦克风音频
-> 流式语音识别
-> 部分文本与意图判断
-> 智能体调用模型和工具
-> 流式语音合成
-> 扬声器播放
模型路由可以在这里发挥作用:普通问候和简单查询优先走低延迟模型;涉及退款、合同或复杂诊断时,再切换到能力更强的模型。对话状态应由智能体层保存,而不是绑定到某个模型会话,否则模型切换会导致上下文丢失。
语音智能体还需要额外处理几类边界:
- 打断:用户开始说话时,应停止当前语音播放并取消不再需要的生成任务;
- 静音与噪声:不要把环境声音持续发送给昂贵的推理模型;
- 敏感操作确认:付款、删除和身份信息修改应要求明确确认;
- 工具调用等待:查询耗时较长时,可先播放简短进度提示;
- 隐私与留存:分别制定原始音频、转写文本和模型日志的保存策略。
因此,语音不是单独替换一个输入组件,而是会影响取消机制、状态管理、延迟预算和安全策略。
持续优化需要评测闭环,而不是凭感觉换模型
模型选择越丰富,团队越需要可重复的评估方法。每次上线新模型前,可以固定一组经过脱敏的代表性任务,比较以下指标:
- 任务成功率和人工评分;
- 首个文本或音频响应的时间;
- 完整响应延迟;
- 工具调用参数的正确率;
- 单次成功任务的估算成本;
- 超时、限流和回退比例;
- 安全策略触发与误拦截情况。
线上日志至少应记录 request_id、任务类型、模型别名、实际模型版本、延迟、工具调用结果、回退次数和最终业务结果。不要只记录“请求成功”,因为模型返回 HTTP 200 并不代表用户的问题已经解决。
更换模型时,建议采用渐进式流程:
- 用离线数据集建立旧模型基线;
- 对新模型运行相同评测,并检查失败案例;
- 先让新模型接收少量低风险流量;
- 同时监控质量、成本、延迟与安全指标;
- 达标后逐步扩大流量,并保留快速回退开关。
提示词也应版本化。模型更新后,旧提示词可能变得冗余,结构化输出或工具调用行为也可能变化。模型、提示词、工具定义和评测集应被视为一组共同发布的制品。
落地时先检查这六件事
在扩大智能体使用范围之前,可以用下面的清单检查架构是否真的支持持续演进:
- 业务服务是否只依赖稳定的智能体接口,而没有写死模型名称?
- 是否可以通过配置或流量规则切换模型,不必重新部署所有调用方?
- 模型失败时是否有超时、重试、回退和熔断策略?
- 文本与语音入口是否复用相同的权限、工具和对话状态?
- 是否能把模型响应映射到任务成功率,而不只是基础设施指标?
- 新模型是否经过离线评测、小流量验证和可逆发布?
模型选择增多本身不会自动提升业务效果。真正带来速度的是把变化限制在清晰的接口之后,再用评测数据决定何时升级。这样,团队采用更合适的模型时,推进的是产品能力,而不是重新施工底层架构。