AI 项目中的中间件实践

要点

  • AI 项目的中间件需要处理 token 计量、流式响应转发、prompt 缓存、多租户鉴权等特有场景
  • token 计量中间件在请求前后统计输入/输出 token 数，用于计费和限速
  • 流式响应（SSE）需要特殊的中间件处理，不能简单套用同步模式
  • prompt 缓存可以避免重复计算相似的请求，降低 API 调用成本
  • 多租户场景下，中间件需要根据租户配置动态调整限流、配额、模型选择
  • AI 中间件的设计需要考虑成本、延迟、可观测性三个维度

1. Token 计量中间件

AI API 通常按 token 数计费。中间件可以在请求前后统计 token 用量：

// src/middleware/token-counter.ts
import { createMiddleware } from 'hono/factory'
 
type TokenUsage = {
  promptTokens: number
  completionTokens: number
  totalTokens: number
}
 
export const tokenCounter = createMiddleware<{
  Variables: { tokenUsage: TokenUsage }
}>(async (c, next) => {
  await next()
 
  // 从响应里读取 token 用量（假设路由已经写入 context）
  const usage = c.get('tokenUsage') as TokenUsage | undefined
 
  if (usage) {
    // 记录到数据库或日志
    console.log(JSON.stringify({
      time: new Date().toISOString(),
      userId: c.get('userId'),
      model: c.get('model'),
      ...usage,
    }))
 
    // 更新用户的 token 配额
    await updateTokenQuota(c.get('userId'), usage.totalTokens)
  }
})
// src/routes/chat.ts
import { Hono } from 'hono'
import { openai } from '../lib/openai'
 
const chat = new Hono()
 
chat.post('/completions', async (c) => {
  const { messages, model } = await c.req.json()
 
  const response = await openai.chat.completions.create({
    model,
    messages,
  })
 
  // 把 token 用量写入 context
  c.set('tokenUsage', {
    promptTokens: response.usage.prompt_tokens,
    completionTokens: response.usage.completion_tokens,
    totalTokens: response.usage.total_tokens,
  })
 
  c.set('model', model)
 
  return c.json(response)
})
 
export default chat

token 计量中间件的关键是「不阻塞请求」——统计和配额更新可以异步执行，不影响响应速度。

2. 流式响应转发

AI API 常用 SSE（Server-Sent Events）返回流式响应。中间件需要特殊处理流式数据：

// src/middleware/stream-logger.ts
import { createMiddleware } from 'hono/factory'
 
export const streamLogger = createMiddleware(async (c, next) => {
  const isStream = c.req.header('Accept')?.includes('text/event-stream')
 
  if (!isStream) {
    await next()
    return
  }
 
  // 记录流式请求的开始
  console.log(`Stream started: ${c.req.method} ${c.req.path}`)
  const start = Date.now()
 
  await next()
 
  // 流式响应的处理
  const duration = Date.now() - start
  console.log(`Stream ended: ${duration}ms`)
})

流式响应的中间件不能在 await next() 之后读取响应体（因为响应是流式的，已经发送给客户端）。如果需要修改流式内容（例如过滤敏感信息），需要包装 c.res：

// src/middleware/stream-filter.ts
import { createMiddleware } from 'hono/factory'
 
export const streamFilter = createMiddleware(async (c, next) => {
  await next()
 
  // 检查是否是流式响应
  if (!c.res.body) return
 
  const contentType = c.res.headers.get('Content-Type') ?? ''
  if (!contentType.includes('text/event-stream')) return
 
  // 包装原始流，过滤内容
  const originalStream = c.res.body
  const filteredStream = new ReadableStream({
    async start(controller) {
      const reader = originalStream.getReader()
      const decoder = new TextDecoder()
      const encoder = new TextEncoder()
 
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
 
        const chunk = decoder.decode(value)
        // 过滤敏感信息
        const filtered = chunk.replace(/password:\s*\S+/g, 'password: [FILTERED]')
        controller.enqueue(encoder.encode(filtered))
      }
 
      controller.close()
    },
  })
 
  c.res = new Response(filteredStream, {
    headers: c.res.headers,
    status: c.res.status,
  })
})

流式中间件的性能开销需要关注——每个 chunk 都要解码、处理、重新编码。对于高吞吐场景，可能需要在 CDN 或网关层处理。

3. Prompt 缓存

AI API 调用成本高，相似的 prompt 可以复用之前的结果：

// src/middleware/prompt-cache.ts
import { createMiddleware } from 'hono/factory'
import { createHash } from 'crypto'
 
type CacheStore = {
  get(key: string): Promise<string | null>
  set(key: string, value: string, ttl: number): Promise<void>
}
 
export function promptCache(store: CacheStore, ttl = 3600) {
  return createMiddleware(async (c, next) => {
    // 只对 POST 请求生效
    if (c.req.method !== 'POST') {
      await next()
      return
    }
 
    const body = await c.req.clone().json()
    const { messages, model } = body
 
    // 计算 prompt 的 hash 作为缓存 key
    const promptHash = createHash('sha256')
      .update(JSON.stringify({ messages, model }))
      .digest('hex')
 
    const cacheKey = `prompt:${promptHash}`
 
    // 检查缓存
    const cached = await store.get(cacheKey)
    if (cached) {
      console.log(`Prompt cache hit: ${cacheKey}`)
      return c.json(JSON.parse(cached))
    }
 
    // 缓存未命中，执行请求
    await next()
 
    // 缓存响应
    if (c.res.status === 200) {
      const responseBody = await c.res.clone().json()
      await store.set(cacheKey, JSON.stringify(responseBody), ttl)
    }
  })
}
// src/index.ts
import { Hono } from 'hono'
import { promptCache } from './middleware/prompt-cache'
import { redisStore } from './lib/redis'
 
const app = new Hono()
 
// prompt 缓存中间件
app.use('/api/chat/*', promptCache(redisStore, 3600))
 
app.post('/api/chat/completions', async (c) => {
  // 调用 AI API
})
 
export default app

Prompt 缓存的注意事项：

  1. 温度参数：temperature > 0 时响应不幂等，缓存可能不合适
  2. 缓存失效：prompt 微小变化（例如时间戳）会导致缓存未命中，可以用模糊匹配
  3. 缓存粒度：可以缓存完整响应，也可以缓存部分结果（例如 embedding）

4. 多租户鉴权与配额

AI 项目通常是 SaaS 模式，不同租户有不同的配额和模型权限：

// src/middleware/tenant-auth.ts
import { createMiddleware } from 'hono/factory'
 
type Tenant = {
  id: string
  name: string
  quota: {
    maxTokensPerMinute: number
    allowedModels: string[]
  }
}
 
export const tenantAuth = createMiddleware<{
  Variables: { tenant: Tenant }
}>(async (c, next) => {
  const apiKey = c.req.header('X-API-Key')
 
  if (!apiKey) {
    return c.json({ error: 'Missing API key' }, 401)
  }
 
  // 查询租户信息
  const tenant = await db.tenant.findUnique({
    where: { apiKey },
  })
 
  if (!tenant) {
    return c.json({ error: 'Invalid API key' }, 401)
  }
 
  c.set('tenant', tenant)
  await next()
})
// src/middleware/model-guard.ts
import { createMiddleware } from 'hono/factory'
 
export const modelGuard = createMiddleware(async (c, next) => {
  const tenant = c.get('tenant') as Tenant
  const body = await c.req.clone().json()
  const { model } = body
 
  if (!tenant.quota.allowedModels.includes(model)) {
    return c.json({
      error: `Model ${model} not allowed for your plan`,
    }, 403)
  }
 
  await next()
})
// src/middleware/token-rate-limit.ts
import { createMiddleware } from 'hono/factory'
 
export const tokenRateLimit = createMiddleware(async (c, next) => {
  const tenant = c.get('tenant') as Tenant
  const currentUsage = await getTenantTokenUsage(tenant.id)
 
  if (currentUsage > tenant.quota.maxTokensPerMinute) {
    return c.json({
      error: 'Rate limit exceeded',
      retryAfter: 60,
    }, 429)
  }
 
  await next()
})
// src/index.ts
import { Hono } from 'hono'
import { tenantAuth } from './middleware/tenant-auth'
import { modelGuard } from './middleware/model-guard'
import { tokenRateLimit } from './middleware/token-rate-limit'
 
const app = new Hono()
 
// 租户鉴权 → 模型权限检查 → token 限速
app.use('/api/*', tenantAuth)
app.use('/api/chat/*', modelGuard)
app.use('/api/chat/*', tokenRateLimit)
 
export default app

多租户中间件的关键是「配置驱动」——租户的配额、权限、模型列表都存在数据库里，中间件只负责查询和执行策略。

5. 请求重试与降级

AI API 调用可能失败（超时、限流、服务端错误）。中间件可以实现重试和降级逻辑：

// src/middleware/retry.ts
import { createMiddleware } from 'hono/factory'
 
export function retry(maxRetries = 3, baseDelay = 1000) {
  return createMiddleware(async (c, next) => {
    let lastError: Error | null = null
 
    for (let i = 0; i < maxRetries; i++) {
      try {
        await next()
        return
      } catch (err) {
        lastError = err as Error
 
        // 只对可重试的错误重试（5xx、超时）
        if (!isRetryable(err)) {
          throw err
        }
 
        // 指数退避
        const delay = baseDelay * Math.pow(2, i)
        await new Promise((resolve) => setTimeout(resolve, delay))
      }
    }
 
    throw lastError
  })
}
 
function isRetryable(err: unknown): boolean {
  if (err instanceof Error) {
    // 超时错误
    if (err.name === 'AbortError') return true
    // 服务端错误（需要路由抛出特定错误）
    if (err.message.includes('503') || err.message.includes('504')) return true
  }
  return false
}

降级逻辑可以在重试失败后切换到备用模型：

// src/middleware/model-fallback.ts
import { createMiddleware } from 'hono/factory'
 
export function modelFallback(primaryModel: string, fallbackModel: string) {
  return createMiddleware(async (c, next) => {
    const body = await c.req.clone().json()
 
    try {
      // 尝试主模型
      c.set('model', primaryModel)
      await next()
    } catch (err) {
      if (isRetryable(err)) {
        // 降级到备用模型
        console.log(`Fallback from ${primaryModel} to ${fallbackModel}`)
        c.set('model', fallbackModel)
        await next()
      } else {
        throw err
      }
    }
  })
}

6. 可观测性

AI 项目的可观测性需要关注：

  1. 延迟：AI API 调用通常较慢（秒级），需要监控 P50/P95/P99
  2. 成本：token 用量、API 调用次数
  3. 错误率：超时、限流、模型错误
  4. 缓存命中率：prompt 缓存的效果
// src/middleware/metrics.ts
import { createMiddleware } from 'hono/factory'
 
export const metrics = createMiddleware(async (c, next) => {
  const start = Date.now()
  const requestId = c.get('requestId') as string
 
  await next()
 
  const duration = Date.now() - start
  const status = c.res.status
  const model = c.get('model') as string | undefined
  const tokenUsage = c.get('tokenUsage') as { totalTokens: number } | undefined
 
  // 上报指标
  reportMetrics({
    requestId,
    method: c.req.method,
    path: c.req.path,
    status,
    duration,
    model,
    tokens: tokenUsage?.totalTokens ?? 0,
  })
})

结合 Prometheus、Grafana 或云监控服务，可以建立 AI API 的监控面板。

7. 完整的 AI 中间件栈

把这一节的中间件组合起来，一个完整的 AI 项目中间件栈：

// src/index.ts
import { Hono } from 'hono'
import { logger } from 'hono/logger'
import { requestId } from 'hono/request-id'
import { cors } from 'hono/cors'
import { tenantAuth } from './middleware/tenant-auth'
import { modelGuard } from './middleware/model-guard'
import { tokenRateLimit } from './middleware/token-rate-limit'
import { promptCache } from './middleware/prompt-cache'
import { retry } from './middleware/retry'
import { tokenCounter } from './middleware/token-counter'
import { metrics } from './middleware/metrics'
 
const app = new Hono()
 
// 公共中间件
app.use('*', logger())
app.use('*', requestId())
app.use('*', cors())
app.use('*', metrics)
 
// AI 专用中间件
app.use('/api/*', tenantAuth)
app.use('/api/chat/*', modelGuard)
app.use('/api/chat/*', tokenRateLimit)
app.use('/api/chat/*', promptCache(redisStore, 3600))
app.use('/api/chat/*', retry(3, 1000))
app.use('/api/chat/*', tokenCounter)
 
export default app

执行顺序：

  1. 日志、请求 ID、CORS、指标（公共）
  2. 租户鉴权（识别租户）
  3. 模型权限检查（检查租户是否有权使用该模型）
  4. Token 限速（检查租户是否超出配额）
  5. Prompt 缓存（检查是否有缓存结果）
  6. 重试（调用 AI API，失败时重试）
  7. Token 计量（统计用量，更新配额）

每个中间件只做一件事，通过组合实现完整的 AI 请求处理链路。

总结

AI 项目的中间件需要处理 token 计量、流式响应、prompt 缓存、多租户鉴权、重试降级、可观测性等特有场景。

这一节涉及到的几个中间件：

  1. Token 计量：统计输入/输出 token 数，用于计费和限速
  2. 流式响应转发：处理 SSE 流式数据，过滤敏感信息
  3. Prompt 缓存：复用相似 prompt 的结果，降低 API 调用成本
  4. 多租户鉴权：根据租户配置动态调整限流、配额、模型选择
  5. 请求重试与降级：失败时重试或切换到备用模型
  6. 可观测性：监控延迟、成本、错误率、缓存命中率

AI 中间件的设计需要考虑成本、延迟、可观测性三个维度。与传统的 CRUD 中间件相比，AI 中间件更关注资源消耗和成本控制。

中间件系列到这里就结束了。从基本概念到内置中间件，再到自定义开发和 AI 项目实践，这十七篇覆盖了 Hono 中间件体系的主要用法。接下来进入 06-Validation & Type Safety，会展开数据校验和类型安全的主题。