模型接入与输出验证
统一请求契约,保留供应商能力差异,把格式、业务约束和事实依据分别检查。
接口统一到应用需要的边界
应用通常需要生成文本、结构化对象、工具调用、嵌入和流式事件。官方 SDK 提供相应供应商的接口与类型;统一 SDK 把若干接口映射成公共形态。一个兼容 Chat Completions 的端点是否支持 Responses、实时音频、文件、严格 schema、缓存和服务端会话,要逐项核实。修改 base URL 只能改变请求目的地。
可以在应用内部声明 generate、stream、embed 三种能力,并在适配器中保存厂商原始 request ID、终止原因和用量。统一结果仍需允许表达“该能力不支持”、拒绝、截断和中断;把这些情形都塞进空字符串会让调用方无法处理。
TypeScript 的 AI SDK 提供跨 provider 的 Core 与 UI 接口;官方 SDK 则适合直接使用某家专属能力。两条路线都需要应用自身的超时、费用预算和业务授权。AI SDK 官方概览 描述的是工具包能力,具体 provider 的支持范围需要查看相应文档。
让 schema 成为应用的一份契约
假设应用从文本中提取待办事项,每项包括标题、负责人和来源段落 ID。schema 可以约束字段与类型,应用还要确认负责人属于团队、来源段落可访问且确实支持待办。未提供负责人时返回 null,比凭空补一个名字更便于后续处理。
下面是 TypeScript + Zod 的应用层验证例子。raw 是模型接口最终完成后的原始结果,allowedSourceIds 来自本次经过权限过滤的原文;示例没有模型网络调用。
import { z } from 'zod';
const TaskSchema = z.object({
title: z.string().min(1),
ownerId: z.string().nullable(),
sourceId: z.string(),
}).strict();
const TasksSchema = z.array(TaskSchema).max(50);
export function parseTasks(
raw: unknown,
allowedSourceIds: ReadonlySet<string>,
memberIds: ReadonlySet<string>,
) {
const tasks = TasksSchema.parse(raw);
for (const task of tasks) {
if (!allowedSourceIds.has(task.sourceId)) {
throw new Error('引用不在本次可用资料中');
}
if (task.ownerId !== null && !memberIds.has(task.ownerId)) {
throw new Error('负责人不在允许成员范围内');
}
}
return tasks;
}这段代码核实了结构、来源 ID 和成员 ID,尚未核实“该段文字真的表达了该待办”。事实核查需要回看原文或人工审阅。服务端约束解码可以减少格式错误,Zod/Pydantic 验证可以发现已写入规则的业务错误,两者都没有覆盖所有事实的证明能力。Python 中可用 Pydantic 定义对应模型和自定义 validator;注意其默认类型转换与 strict 模式的区别。Pydantic 文档 提供类型与验证机制。
TypeScript:官方 Responses 接口
以下示例抽取一项待办,使用 Zod 定义输出,再把拒绝和未完成状态独立处理。LLM_MODEL 必须指向支持相应结构化输出的模型;依赖是 openai 与 zod,SDK 版本需支持 responses.parse 和 zodTextFormat。
import OpenAI from 'openai';
import { zodTextFormat } from 'openai/helpers/zod';
import { z } from 'zod';
const Task = z.object({
title: z.string(),
dueDate: z.string().nullable(),
});
const model = process.env.LLM_MODEL;
if (!model) throw new Error('LLM_MODEL is required');
const client = new OpenAI(); // 从服务端 OPENAI_API_KEY 读取凭据
export async function extractTask(text: string) {
const response = await client.responses.parse({
model,
input: [
{ role: 'system', content: '提取待办;没有日期时使用 null。' },
{ role: 'user', content: text },
],
text: { format: zodTextFormat(Task, 'task') },
max_output_tokens: 256,
});
const refused = response.output.some(
(item) => item.type === 'message'
&& item.content.some((part) => part.type === 'refusal'),
);
if (refused) return { status: 'refused' as const };
if (response.status !== 'completed' || !response.output_parsed) {
return { status: 'incomplete' as const, responseId: response.id };
}
return {
status: 'ok' as const,
task: response.output_parsed,
responseId: response.id,
usage: response.usage,
};
}本轮核对了官方指南中的两种语言 parse 接口及拒绝处理。严格 schema 只覆盖供应商支持的子集;拒绝或中断需另行判断。日期合法、日期确实来自输入,以及该用户是否可以创建任务,应由应用进一步验证。OpenAI 结构化输出指南。
Python:相同契约的客户端
Pydantic 可以维护字段结构与应用验证,依赖为 openai 和 pydantic。无日期用 None;日期推断的接受规则由应用决定。
import os
from openai import OpenAI
from pydantic import BaseModel
class Task(BaseModel):
title: str
due_date: str | None
client = OpenAI()
response = client.responses.parse(
model=os.environ['LLM_MODEL'],
input=[
{'role': 'system', 'content': '提取待办;没有日期时使用 null。'},
{'role': 'user', 'content': '周五检查发布说明。'},
],
text_format=Task,
max_output_tokens=256,
)
refused = any(
getattr(part, 'type', None) == 'refusal'
for item in response.output
for part in getattr(item, 'content', [])
)
if refused:
result = {'status': 'refused'}
elif response.status != 'completed' or response.output_parsed is None:
result = {'status': 'incomplete', 'response_id': response.id}
else:
result = {'status': 'ok', 'task': response.output_parsed.model_dump()}异步服务使用对应异步客户端,复用连接池,并在服务退出时关闭。将 SDK 网络重试、业务重试和用户点击重试分别计入一次运行的总预算;网络超时可能发生在服务端已经处理请求之后。
多供应商接入
AI SDK 可将 generateText/streamText 的 output 配置与 Zod、Valibot 或 JSON Schema 结合。以下接线示意使用统一网关模型标识;使用厂商 provider 包时,改为该 provider 的模型实例。
import { generateText, Output } from 'ai';
import { z } from 'zod';
const model = process.env.AI_MODEL;
if (!model) throw new Error('AI_MODEL is required');
const result = await generateText({
model,
prompt: '把下面资料归为说明、问题或请求:请补充部署步骤。',
output: Output.object({
schema: z.object({ category: z.enum(['说明', '问题', '请求']) }),
}),
});
const classification = result.output;本轮官方页面确认:部分对象流尚未完整,无法按完整 schema 验证;完整数组元素流与部分数组流具有不同验证边界。流错误还需由错误回调或流消费者接收。AI SDK 结构化数据文档。
LiteLLM 可以作为应用库或 Proxy。采用 Proxy 时,应用保存网关凭据,网关管理厂商凭据、模型映射、预算和路由。应用层验证保留在网关之后。instructor 可为 Pydantic 校验失败反馈错误并重试;自定义 validator 只验证写出的规则,无法自动证明任意语义正确。原归档“唯一语义兜底”的断言缺少依据,不作为选型前提。LiteLLM Proxy、instructor。
国内供应商与兼容接口
归档核对过 DeepSeek、百炼 Qwen、Z.AI GLM 和 Moonshot Kimi 的 OpenAI 兼容接入。共同操作是配置 base_url/baseURL、供应商 key 和模型名;Qwen 端点与区域相关,模型、key 和数据所在地必须匹配。Chat Completions 的兼容性需与 Responses、strict schema、工具调用、计费字段分别确认。模型重命名或端点迁移进入配置更新,避免散落在业务函数中。DeepSeek 接入说明、百炼兼容接口、Z.AI 接口、Moonshot 文档。
Claude 和 Gemini 官方 SDK 的消息、结构化输出和自动工具执行字段各有约定。厂商提供的工具 runner 可省去循环样板,权限、预算和外部效果的幂等性仍由宿主负责。归档中 Google 两代 API 的迁移判断、Anthropic beta runner 的具体停止条件应按采用版本复查,不将未经本轮核对的状态写为固定默认。Claude SDK 文档、Google GenAI SDK。
重试、错误与预算一起设计
请求失败分成连接失败、限流、供应商错误、拒绝、输出截断、解析失败、业务不一致和无证据。限流可在总截止时间内退避;截断可以调整输出预算或任务拆分;解析错误可带具体错误重试一次;没有证据时应保留未知状态。把所有失败都重新提交相同请求会重复收费,也会掩盖真正的失败位置。
读取操作的网络重试和工具写入的业务重试分别控制。模型生成了一次写工具调用,应用须先创建动作身份并检查是否已有回执,随后才执行。供应商 SDK 自带网络重试时,要把它计入应用重试预算,防止两层重试相乘。
用量来自供应商的完成事件或计费记录,预算预估来自本地 tokenizer。它们可能因隐藏提示、工具定义、多模态编码和缓存口径不同而有差异。记录每次尝试的用量与路由,最终成功结果不能抹掉前面失败调用的费用。
提示缓存作为条件能力接入
把稳定的指令、工具定义和文档前缀安排在前面,有利于部分服务复用提示处理。但是否自动命中、最小长度、TTL、缓存创建费、租户隔离与模型支持都由服务决定。服务端缓存不等于应用已保存会话,也不赋予访问资料的权限。
能力表至少区分支持结构化输出、schema 子集、工具并行、流式工具增量、提示缓存、用量终态、取消语义及数据留存。锁定 SDK 与模型后根据实际文档和用例填表。变更模型、网关或 adapter 后复查这些行为,避免沿用旧接口假设。
最后更新于