AI 接口的参数校验设计
要点
- AI 接口的请求结构比普通 REST 接口复杂得多：嵌套数组、动态内容类型、模型参数范围约束交织在一起，用单个
z.object()难以维护 messages数组中每条消息的合法字段取决于role的值，z.discriminatedUnion比z.union更精确- Token 预估和内容安全属于「业务级校验」，和 Zod 做的「结构级校验」是两层不同的事，不宜混在一起
- 不同模型提供商的参数范围和字段名存在差异，校验层需要把差异收口到统一的内部 schema，在适配层做转换
- 结构化输出可以用同一个 schema 同时约束请求端和校验响应端
- AI 接口更适合按 token 消耗而非请求次数做频率限制
- 这是「请求验证与类型安全」系列的最后一篇，会把前面学到的能力整合到一个完整示例里
1. AI 接口为什么需要单独的校验策略
前面 11 篇覆盖了 Zod schema、zValidator、DTO 分层、错误消息规范化等能力。这些对普通 REST 接口够用了，但 AI 接口的请求结构有几层特殊复杂度：
- 嵌套数组里的异构对象：
messages数组中每条消息的字段取决于role的值 - 数值参数的范围约束：
temperature在 0–2 之间，top_p在 0–1 之间，不同模型范围还不同 - 条件必填字段：
role为tool时必须提供tool_call_id，assistant时可能有tool_calls - 动态结构：
tools内部的parameters是开放的 JSON Schema，结构不固定 - 流式与非流式：同一个接口根据
stream字段返回完全不同的响应格式
这些复杂度叠加在一起，按职责逐层拆分比一锅端更可控。
2. 聊天消息数组的校验
messages 数组的难点在于每条消息的合法字段取决于 role。Zod 的 z.discriminatedUnion 先按判别字段分流，再对匹配分支做精确校验：
// src/schemas/chat-messages.ts
import { z } from 'zod'
const systemMessageSchema = z.object({
role: z.literal('system'),
content: z.string().min(1),
name: z.string().max(64).optional(),
})
const userMessageSchema = z.object({
role: z.literal('user'),
content: z.union([
z.string().min(1),
z.array(z.object({
type: z.enum(['text', 'image_url']),
text: z.string().optional(),
image_url: z.object({
url: z.string().url(),
detail: z.enum(['low', 'high', 'auto']).optional(),
}).optional(),
})),
]),
})
const assistantMessageSchema = z.object({
role: z.literal('assistant'),
content: z.string().nullable().optional(),
tool_calls: z.array(z.object({
id: z.string().min(1),
type: z.literal('function'),
function: z.object({
name: z.string().min(1),
arguments: z.string().min(1),
}),
})).optional(),
})
const toolMessageSchema = z.object({
role: z.literal('tool'),
content: z.string().min(1),
tool_call_id: z.string().min(1),
})
const chatMessageSchema = z.discriminatedUnion('role', [
systemMessageSchema, userMessageSchema,
assistantMessageSchema, toolMessageSchema,
])z.discriminatedUnion 相比 z.union 有两个优势：校验性能更好（不需要逐个分支尝试），错误信息更精确（会指出哪个 role 分支不匹配）。
跨消息的约束需要用 .refine() 补充——schema 管字段结构，refine 管上下文规则：
// src/schemas/chat-messages.ts
const messagesArraySchema = z.array(chatMessageSchema).min(1).max(200)
.refine((msgs) => msgs.some((m) => m.role === 'user'),
{ message: '消息列表中必须至少包含一条 user 消息' })
.refine((msgs) => ['system', 'user'].includes(msgs[0].role),
{ message: '第一条消息必须是 system 或 user' })3. 模型参数范围与多模型路由
不同模型的参数范围不一样。硬编码某个模型的范围会在切换模型时出问题。用模型注册表 + 通用 schema 的方式可以兼顾统一校验和差异化约束：
// src/config/model-registry.ts
interface ModelSpec { maxTokensLimit: number; temperatureRange: [number, number] }
const MODEL_REGISTRY: Record<string, ModelSpec> = {
'gpt-4o': { maxTokensLimit: 16384, temperatureRange: [0, 2] },
'gpt-4o-mini': { maxTokensLimit: 16384, temperatureRange: [0, 2] },
'claude-3-5-sonnet': { maxTokensLimit: 8192, temperatureRange: [0, 1] },
}
// src/schemas/model-params.ts
import { z } from 'zod'
const commonModelParamsSchema = z.object({
temperature: z.number().min(0).max(2).default(0.7),
top_p: z.number().min(0).max(1).default(1),
max_tokens: z.number().int().positive().max(128000).optional(),
presence_penalty: z.number().min(-2).max(2).default(0),
frequency_penalty: z.number().min(-2).max(2).default(0),
stop: z.union([z.string(), z.array(z.string()).max(4)]).optional(),
})
// 针对特定模型动态生成更精确的范围约束
function createModelParamsSchema(model: string) {
const spec = MODEL_REGISTRY[model]
if (!spec) return commonModelParamsSchema
return commonModelParamsSchema.extend({
temperature: z.number().min(spec.temperatureRange[0]).max(spec.temperatureRange[1]).default(0.7),
max_tokens: z.number().int().positive().max(spec.maxTokensLimit).optional(),
})
}后端对接多个模型提供商时，不同提供商的参数名可能不同。在接口层使用统一的内部 schema，在适配层做格式转换：
// src/adapters/model-parameter-adapter.ts
export function adaptToProvider(params: {
model: string; temperature: number; maxOutputTokens?: number
topP: number; provider: 'openai' | 'anthropic' | 'other'
}): Record<string, unknown> {
const base = { model: params.model, temperature: params.temperature, top_p: params.topP }
// openai/anthropic 用 max_tokens，其他提供商可能用 max_output_tokens
const tokenKey = params.provider === 'other' ? 'max_output_tokens' : 'max_tokens'
return { ...base, [tokenKey]: params.maxOutputTokens }
}请求校验逻辑不需要按 provider 分支，只需要在最终调用模型 API 之前做一次参数翻译。
4. Tool / Function Calling 的参数校验
Tool calling 的校验分两个阶段：请求前校验 tool 定义格式，响应后校验模型返回的 tool_calls。
请求侧，tool 定义中的 parameters 是一段 JSON Schema，形成递归结构，需要 z.lazy 处理：
// src/schemas/tool-definition.ts
import { z } from 'zod'
// parameters 是递归的 JSON Schema 结构，需要 z.lazy
const jsonSchemaSubsetSchema: z.ZodType<any> = z.lazy(() =>
z.object({
type: z.enum(['object', 'array', 'string', 'number', 'boolean', 'null']),
description: z.string().optional(),
properties: z.record(jsonSchemaSubsetSchema).optional(),
items: jsonSchemaSubsetSchema.optional(),
required: z.array(z.string()).optional(),
}).passthrough()
)
export const toolDefinitionSchema = z.object({
type: z.literal('function'),
function: z.object({
name: z.string().min(1).max(64).regex(/^[a-zA-Z0-9_-]+$/, '函数名只能包含字母、数字、下划线和连字符'),
description: z.string().min(1).max(1024),
parameters: jsonSchemaSubsetSchema.optional(),
}),
})
export const toolsArraySchema = z.array(toolDefinitionSchema).min(1).max(128)
.refine((tools) => new Set(tools.map((t) => t.function.name)).size === tools.length,
{ message: 'tool 名称不能重复' })响应侧，校验模型返回的 tool_calls 是否引用了已注册的 tool：
// src/validators/tool-call-validator.ts
export function validateToolCallResponse(
toolCalls: Array<{ id: string; function: { name: string; arguments: string } }>,
registeredTools: Array<{ function: { name: string } }>
) {
const toolMap = new Set(registeredTools.map((t) => t.function.name))
const errors: string[] = []
for (const call of toolCalls) {
if (!toolMap.has(call.function.name)) { errors.push(`调用了未注册的 tool: ${call.function.name}`); continue }
try { JSON.parse(call.function.arguments) }
catch { errors.push(`tool ${call.function.name} 的 arguments 不是合法 JSON`) }
}
return { valid: errors.length === 0, errors }
}5. 流式请求与 SSE 事件校验
流式接口的请求体校验和普通接口相同——stream 只是一个布尔字段。区别在响应端，需要逐块校验 SSE 事件格式：
// src/utils/sse-validator.ts
import { z } from 'zod'
const sseDataSchema = z.object({
choices: z.array(z.object({
index: z.number().int().nonnegative(),
delta: z.object({ content: z.string().optional(), tool_calls: z.array(z.any()).optional() }),
finish_reason: z.string().nullable().optional(),
})),
usage: z.object({ prompt_tokens: z.number(), completion_tokens: z.number(), total_tokens: z.number() }).optional(),
})
export function validateSSEChunk(rawData: string) {
if (rawData === '[DONE]') return { done: true, valid: true }
try {
const result = sseDataSchema.safeParse(JSON.parse(rawData))
if (!result.success) {
console.warn('[SSE] 格式不合规的 chunk:', result.error.issues)
return { done: false, valid: false }
}
return { done: false, valid: true, data: result.data }
} catch { return { done: false, valid: false } }
}流式场景下不适合直接拒绝请求。更稳妥的做法是记录异常 chunk、尝试跳过，或在错误严重时关闭连接。
6. 内容安全、Token 预估与成本限流
这三项属于「业务级校验」，和 Zod 做的「结构级校验」是不同层次的事。
Token 预估
在发送给模型之前做 token 预估，避免超出上下文窗口：
// src/utils/token-estimation.ts
// 粗略估算：中文约 1.5 字/token，英文约 4 字符/token；精确计数需要用 tiktoken
export function estimateTokenCount(
messages: Array<{ role: string; content: string | unknown[] }>
): number {
let total = 0
for (const msg of messages) {
total += 4
const content = typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content)
const cjk = (content.match(/[一-鿿]/g) || []).length
total += Math.ceil(cjk * 1.5) + Math.ceil((content.length - cjk) * 0.25)
}
return total + 2
}粗略估算适合前置拦截，精确计数场景（比如计费）应该使用 tiktoken。
内容安全校验
Prompt 注入检测通常放在独立的中间件层：
// src/middleware/content-safety.ts
import type { Context } from 'hono'
const INJECTION_PATTERNS = [
/忽略(?:上面|以上|前面)(?:的)?(?:所有)?指令/,
/ignore (?:all )?(?:previous|above) instructions/i,
/system\s*:\s*you are now/i,
]
export async function contentSafetyMiddleware(c: Context, next: () => Promise<void>) {
const body = await c.req.json()
const messages = (body.messages ?? []) as Array<{ role: string; content: string }>
for (const msg of messages) {
if (msg.role !== 'user') continue
if (INJECTION_PATTERNS.some((p) => p.test(msg.content))) {
return c.json({ code: 'CONTENT_SAFETY_REJECTED', message: '消息内容触发了安全规则' }, 400)
}
}
await next()
}正则匹配只是第一道防线。实际项目中通常还需要接入外部内容审核服务，处理更隐蔽的注入手法。
成本感知的频率限制
普通 API 按请求次数限流。AI 接口更适合按 token 消耗限流——一个带 10 万 token 上下文的请求和一个 100 token 的请求，消耗的资源差距很大：
// src/middleware/ai-rate-limiter.ts
import type { Context } from 'hono'
const TOKEN_QUOTA_PER_MINUTE = 100_000
export async function tokenBasedRateLimiter(c: Context, next: () => Promise<void>) {
const estimatedTokens: number = c.get('estimatedTokens') ?? 0
const consumed = await getConsumedTokens(c.req.header('x-api-key') ?? 'anonymous')
if (consumed + estimatedTokens > TOKEN_QUOTA_PER_MINUTE) {
return c.json({
code: 'RATE_LIMIT_EXCEEDED',
message: `token 配额不足：已使用 ${consumed}，本次预估 ${estimatedTokens}`,
retry_after: 60,
}, 429)
}
await next()
}estimatedTokens 来自 token 预估中间件，让配额计算基于实际资源消耗而非请求次数。
7. RAG 参数与结构化输出
RAG（检索增强生成）接口在聊天参数之上多了一组检索相关字段：
// src/schemas/rag-params.ts
import { z } from 'zod'
export const ragParamsSchema = z.object({
embedding_model: z.string().min(1),
embedding_dimensions: z.number().int().positive()
.refine((d) => [256, 384, 512, 768, 1024, 1536].includes(d), { message: 'embedding 维度必须是标准值' }),
similarity_threshold: z.number().min(0).max(1).default(0.7),
top_k: z.number().int().min(1).max(20).default(5),
diversity_threshold: z.number().min(0).max(1).optional(),
context_max_tokens: z.number().int().positive().max(32000).optional(),
})embedding_dimensions 用 refine 限制为标准值，因为大多数 embedding 模型只支持固定的维度输出。
要求模型以 JSON 格式返回时，响应端的结构也需要校验。schema 定义一次，请求端和响应端都能用：
// src/schemas/structured-output.ts
import { z } from 'zod'
export const sentimentResultSchema = z.object({
sentiment: z.enum(['positive', 'negative', 'neutral']),
confidence: z.number().min(0).max(1),
reasoning: z.string().max(500),
keywords: z.array(z.string()).max(10),
})
export function validateStructuredOutput<T extends z.ZodType>(raw: string, schema: T) {
try {
const result = schema.safeParse(JSON.parse(raw))
if (result.success) return { success: true as const, data: result.data }
return { success: false as const, error: `输出结构不符合预期: ${result.error.issues.map((i) => i.message).join(', ')}` }
} catch { return { success: false as const, error: '模型返回的内容不是合法 JSON' } }
}用同一个 schema 同时约束请求（告诉模型输出格式）和校验响应（确保模型遵守格式），是 Zod 在 AI 接口里比较有价值的用法之一。
8. 完整示例：AI Chat API 校验链路
把前面的能力整合到一起，看一个完整的校验链路：
// src/routes/chat.ts
import { Hono } from 'hono'
import { z } from 'zod'
import { zValidator } from '@hono/zod-validator'
import { messagesArraySchema } from '../schemas/chat-messages'
import { toolsArraySchema } from '../schemas/tool-definition'
import { ragParamsSchema } from '../schemas/rag-params'
import { estimateTokenCount } from '../utils/token-estimation'
import { contentSafetyMiddleware } from '../middleware/content-safety'
import { tokenBasedRateLimiter } from '../middleware/ai-rate-limiter'
const app = new Hono()
const chatRequestSchema = z.object({
model: z.string().min(1),
messages: messagesArraySchema,
temperature: z.number().min(0).max(2).default(0.7),
top_p: z.number().min(0).max(1).default(1),
max_tokens: z.number().int().positive().optional(),
tools: toolsArraySchema.optional(),
tool_choice: z.union([
z.literal('auto'), z.literal('none'), z.literal('required'),
z.object({ type: z.literal('function'), function: z.object({ name: z.string().min(1) }) }),
]).optional(),
stream: z.boolean().default(false),
response_format: z.object({ type: z.enum(['text', 'json_object']) }).optional(),
rag: ragParamsSchema.optional(),
seed: z.number().int().optional(),
})
app.post('/api/chat',
// 第一层：结构级校验（Zod）
zValidator('json', chatRequestSchema, (result, c) => {
if (!result.success) {
return c.json({ code: 'VALIDATION_ERROR', errors: result.error.flatten().fieldErrors }, 400)
}
}),
// 第二层：内容安全校验
contentSafetyMiddleware,
// 第三层：Token 预估 + 成本限流
async (c, next) => {
const body = await c.req.json()
const estimated = estimateTokenCount(body.messages)
if (estimated > 120000) {
return c.json({ code: 'TOKEN_LIMIT_EXCEEDED', message: `预估 token 数 ${estimated} 超出安全范围` }, 400)
}
c.set('estimatedTokens', estimated)
await next()
},
tokenBasedRateLimiter,
// 所有校验通过，进入业务逻辑
async (c) => {
const request = c.req.valid('json')
// 调用模型、处理流式响应...
return c.json({ status: 'ok' })
}
)
export default app四层校验的分工：
- 结构级（Zod）：请求体格式、字段类型、值范围
- 安全级（中间件）：prompt 注入检测、内容过滤
- 资源级（中间件）：token 预估、基于消耗的限流
- 业务级（处理函数内部）：模型选择、适配层转换、结构化输出校验
每一层只关心自己负责的校验类型，出了问题能快速定位是哪一层拦截的。
延伸阅读
- 01-为什么需要请求验证 — Zod + zValidator 的基础用法和校验原理
- 06-Header 校验 — 请求头校验在 API Key 验证场景中的应用
- 07-嵌套对象与数组校验 —
messages数组校验依赖的嵌套校验能力 - 08-错误消息规范化 — AI 接口四层校验各自的错误格式设计
- Zod discriminatedUnion 文档 —
z.discriminatedUnion的完整用法 - OpenAI API 参考 — Chat Completions 参数规格
总结
AI 接口的校验比传统 REST 接口多出了几个维度。把这几层校验理清楚，能避免很多上线后才暴露的问题。
整个系列从第一篇到现在，覆盖的能力可以归纳为以下几层：
- 基础能力（01–03）：Zod schema 定义、zValidator 中间件、不同位置的数据校验（body / query / param / header）
- 结构设计（04–07）：DTO 与 Schema 分层、嵌套对象和数组校验、动态 schema 组合
- 错误处理（08–09）：错误消息规范化、自定义错误格式
- 类型安全（10–11）：类型推导链路、输入输出类型共享
- 场景整合（12，本文）：AI 接口里把前面所有能力按职责分层组装到一起
核心思路始终是同一条：把校验规则从业务代码中剥离出来，用 schema 声明式地定义，让类型推导和运行时校验共用同一份规则。 无论面对的是普通 REST 接口还是复杂的 AI 聊天请求，这条思路都适用。
至此，「请求验证与类型安全」系列全部结束。下一篇开始进入「Hono RPC」系列——看看怎样用 Hono 的 RPC 特性把后端路由直接变成前端的类型安全调用，schema 即类型这条路径会从后端一路延伸到端到端。