AI 电子伴侣

创建时间: 2026-04-18最后更新: 2026-04-19作者: yangbo

1. 中间件：模型调用的「插拔层」

Hono 有中间件、Express 有中间件，AI SDK 也有中间件——只不过它挂在模型层，而不是 HTTP 层。

AI SDK 的模型中间件通过 wrapLanguageModel 挂载。它能在每次 LLM 调用前后插入逻辑：

请求改写：加日志、改参数、修改 prompt、注入 system
响应改写：过滤敏感词、剪裁长度、拼前缀
缓存：相同 prompt 直接返回上次的答
限流：按用户 / 按租户控制并发
重试：某个模型挂了自动重试
Fallback：检测到 error 自动换 Provider

这些能力是从 Demo 走向生产的关键。这一篇把最常用的几种中间件写清楚，以及怎么把它们组合起来。

2. wrapLanguageModel 基础

wrapLanguageModel 接收一个 model 和一个 middleware，返回新的 model。新 model 仍然是一个标准的 LanguageModelV2，可以到处塞。

wrap-basic.ts

01import { wrapLanguageModel, type LanguageModelV2Middleware } from 'ai'
02import { openai } from '@ai-sdk/openai'
03
04const loggingMiddleware: LanguageModelV2Middleware = {
05  wrapGenerate: async ({ doGenerate, params }) => {
06    console.log('[LLM] calling', params.prompt.length, 'chars')
07    const result = await doGenerate()
08    console.log('[LLM] finished, tokens:', result.usage)
09    return result
10  },
11
12  wrapStream: async ({ doStream, params }) => {
13    console.log('[LLM] streaming start')
14    return doStream()
15  },
16}
17
18const model = wrapLanguageModel({
19  model: openai('gpt-4o'),
20  middleware: loggingMiddleware,
21})
22
23// 使用：和普通 model 完全一样
24streamText({ model, prompt: '...' })

Middleware 有三个 hook：

hook	作用	何时调
`wrapGenerate`	包裹 `generateText` / `generateObject`	非流式调用
`wrapStream`	包裹 `streamText` / `streamObject`	流式调用
`transformParams`	只改参数不包裹执行	所有调用前，两个 hook 会先 `transformParams` 再进入自己

大多数中间件只要实现 wrapGenerate 和 wrapStream 就够了。

3. 缓存中间件：相同问答复用

LLM 调用是贵的。如果同一个 prompt + 参数反复出现（比如用户点「重新生成」），理论上可以命中缓存。

用 wrapLanguageModel + Cloudflare KV 实现缓存：

cache-middleware.ts

01import { type LanguageModelV2Middleware } from 'ai'
02import type { KVNamespace } from '@cloudflare/workers-types'
03
04export function cacheMiddleware(kv: KVNamespace, ttl = 3600): LanguageModelV2Middleware {
05  return {
06    wrapGenerate: async ({ doGenerate, params }) => {
07      const key = await hashKey(params)
08      const cached = await kv.get(key, 'json')
09      if (cached) {
10        console.log('[cache hit]', key)
11        return cached as any
12      }
13
14      const result = await doGenerate()
15      await kv.put(key, JSON.stringify(result), { expirationTtl: ttl })
16      return result
17    },
18
19    // 流式场景一般不缓存（太复杂），或者缓存文本后再重建流
20  }
21}
22
23async function hashKey(params: unknown): Promise<string> {
24  const json = JSON.stringify(params)
25  const hash = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(json))
26  const hex = [...new Uint8Array(hash)].map((b) => b.toString(16).padStart(2, '0')).join('')
27  return `llm:${hex.slice(0, 16)}`
28}

用起来：

use-cache.ts

1import { wrapLanguageModel } from 'ai'
2
3const cachedModel = wrapLanguageModel({
4  model: openai('gpt-4o-mini'),
5  middleware: cacheMiddleware(c.env.KV, 86400),
6})

适合用的场景：

分类 / 摘要类 pipeline（同样输入期望同样输出）
内部工具（翻译、提取）
产品里的示例或 Demo 查询

不适合的场景：

对话（prompt 每次都不同）
对「生成多样性」有要求的场景（用户点「再来一个」期望换一种）

4. 限流中间件：按用户 / 按租户控制并发

保护自己也保护上游（Provider）的 API 配额。基于 Cloudflare KV 的滑动窗口：

rate-limit-middleware.ts

01import { type LanguageModelV2Middleware } from 'ai'
02import type { KVNamespace } from '@cloudflare/workers-types'
03
04interface RateLimitConfig {
05  kv: KVNamespace
06  keyFn: () => string       // 怎么识别调用方（用户 ID、租户 ID、IP）
07  limit: number             // 时间窗口内的最大调用次数
08  windowSec: number         // 窗口秒数
09}
10
11export function rateLimitMiddleware(cfg: RateLimitConfig): LanguageModelV2Middleware {
12  const check = async () => {
13    const k = `rate:${cfg.keyFn()}`
14    const now = Math.floor(Date.now() / 1000)
15    const windowStart = now - cfg.windowSec
16
17    // 读上一个窗口内的请求时间戳列表
18    const raw = await cfg.kv.get(k)
19    const list: number[] = raw ? JSON.parse(raw) : []
20    const fresh = list.filter((t) => t > windowStart)
21
22    if (fresh.length >= cfg.limit) {
23      throw new Error(`rate limit exceeded: ${fresh.length}/${cfg.limit} per ${cfg.windowSec}s`)
24    }
25
26    fresh.push(now)
27    await cfg.kv.put(k, JSON.stringify(fresh), { expirationTtl: cfg.windowSec * 2 })
28  }
29
30  return {
31    wrapGenerate: async ({ doGenerate }) => {
32      await check()
33      return doGenerate()
34    },
35    wrapStream: async ({ doStream }) => {
36      await check()
37      return doStream()
38    },
39  }
40}

使用：

use-rate-limit.ts

1const model = wrapLanguageModel({
2  model: openai('gpt-4o'),
3  middleware: rateLimitMiddleware({
4    kv: c.env.KV,
5    keyFn: () => sessionId,
6    limit: 30,           // 每分钟 30 次
7    windowSec: 60,
8  }),
9})

生产上几条经验：

限流策略和用户等级挂钩（免费用户 30/min、会员 300/min）
超限后给出友好提示（返回 429 + 说明）
全局限流 + 用户限流两层保护，前者防止单用户打爆

5. Fallback 与 Retry 中间件

5.1 Fallback：一家挂了换下一家

场景是 OpenAI 突然 500 或 429，你希望自动切到 Anthropic，不让用户感知。

AI SDK 官方推荐的做法就是组合多个 Provider：

fallback-middleware.ts

01import { type LanguageModelV2Middleware, APICallError } from 'ai'
02import { openai } from '@ai-sdk/openai'
03import { anthropic } from '@ai-sdk/anthropic'
04
05export function fallbackMiddleware(fallback: LanguageModelV2): LanguageModelV2Middleware {
06  return {
07    wrapGenerate: async ({ doGenerate, params, model }) => {
08      try {
09        return await doGenerate()
10      } catch (err) {
11        if (shouldFallback(err)) {
12          console.warn('[fallback] primary failed, retry on fallback', err)
13          // 用 fallback model 跑一次
14          return await fallback.doGenerate(params)
15        }
16        throw err
17      }
18    },
19
20    wrapStream: async ({ doStream, params, model }) => {
21      try {
22        return await doStream()
23      } catch (err) {
24        if (shouldFallback(err)) {
25          return await fallback.doStream(params)
26        }
27        throw err
28      }
29    },
30  }
31}
32
33function shouldFallback(err: unknown): boolean {
34  if (APICallError.isInstance(err)) {
35    // 5xx 或 429 或网络错误才 fallback
36    return err.isRetryable || (err.statusCode ?? 0) >= 500
37  }
38  return false
39}

使用：

use-fallback.ts

1const model = wrapLanguageModel({
2  model: openai('gpt-4o'),
3  middleware: fallbackMiddleware(anthropic('claude-opus-4-6')),
4})

几条经验：

Fallback 不应该级联超过三层，层数多了太慢
Fallback 的模型能力要接近主 model，避免回复质量差太多
要打埋点：每次 fallback 发生要记录，用来监控主 Provider 稳定度
流式下的 Fallback 有坑：如果主 model 已经 emit 了前半段再挂，切 fallback 会重新开始，前半段就白流了。要么接受「重来」，要么检测到进入 stream 就不 fallback

5.2 Retry：短暂失败自动重试

和 Fallback 的区别：retry 是同一个模型重试，fallback 是换模型。

retry-middleware.ts

01export function retryMiddleware(maxRetries = 2, baseDelay = 500): LanguageModelV2Middleware {
02  const doWithRetry = async <T>(fn: () => Promise<T>, retries = maxRetries): Promise<T> => {
03    try {
04      return await fn()
05    } catch (err) {
06      if (retries > 0 && isRetryable(err)) {
07        const delay = baseDelay * Math.pow(2, maxRetries - retries)
08        await sleep(delay + Math.random() * 200) // jitter
09        return doWithRetry(fn, retries - 1)
10      }
11      throw err
12    }
13  }
14
15  return {
16    wrapGenerate: ({ doGenerate }) => doWithRetry(() => doGenerate()),
17    // 流式重试也类似，但要注意部分 emit 的问题
18  }
19}

一般只对非流式调用做 retry，流式调用让上层业务自己决定。

6. 组合多个中间件

middleware 参数支持数组，按顺序组合：

compose.ts

01const model = wrapLanguageModel({
02  model: openai('gpt-4o'),
03  middleware: [
04    loggingMiddleware,
05    cacheMiddleware(kv),
06    rateLimitMiddleware({ kv, keyFn: () => userId, limit: 30, windowSec: 60 }),
07    retryMiddleware(2),
08    fallbackMiddleware(anthropic('claude-opus-4-6')),
09  ],
10})

执行顺序是：数组越靠前，越靠外层。上面这个栈的执行顺序是：

order.txt

1request → logging → cache (hit? return) → rate limit → retry → fallback → primary model

不同场景的组合经验：

后端内部脚本：cache + retry + logging
用户面向的生产：rate limit + retry + fallback + logging（不 cache）
开发阶段：logging + 让错误透传

defaultSettingsMiddleware

有一种特殊的中间件，只改默认参数，不改行为：

default-settings.ts

01import { defaultSettingsMiddleware } from 'ai'
02
03const model = wrapLanguageModel({
04  model: openai('gpt-4o'),
05  middleware: defaultSettingsMiddleware({
06    settings: {
07      temperature: 0.3,
08      maxOutputTokens: 1000,
09      frequencyPenalty: 0.2,
10    },
11  }),
12})

业务层调用时不传这些参数，model 自带；但业务层显式传了也会覆盖。适合「全站默认低 temperature」「某个路由默认短输出」这类统一默认值的场景。

7. 真实世界的中间件栈

AI 伴侣项目的模型栈大概长这样：

real-world.ts

01import { wrapLanguageModel } from 'ai'
02import { createWorkersAI } from 'workers-ai-provider'
03import { openai } from '@ai-sdk/openai'
04
05export function buildCompanionModel(env: Env, sessionId: string, userId: string) {
06  const workersai = createWorkersAI({ binding: env.AI })
07  const primaryRaw = workersai('@cf/meta/llama-3.3-70b-instruct-fp8-fast')
08  const fallbackRaw = openai('gpt-4o-mini')
09
10  return wrapLanguageModel({
11    model: primaryRaw,
12    middleware: [
13      telemetryMiddleware({ sessionId, userId }),       // 先埋点，拿到总量
14      rateLimitMiddleware({
15        kv: env.KV,
16        keyFn: () => userId,
17        limit: 120,
18        windowSec: 60,
19      }),
20      retryMiddleware(2),
21      fallbackMiddleware(fallbackRaw),
22      defaultSettingsMiddleware({
23        settings: { temperature: 0.7 },
24      }),
25    ],
26  })
27}

调用方只需要：

usage.ts

1const model = buildCompanionModel(c.env, sessionId, userId)
2const result = streamText({ model, messages, tools })
3return result.toUIMessageStreamResponse()

中间件栈在业务代码里是不可见的，但每次调用都有限流、重试、Fallback、埋点的保护。

8. 常见坑

Middleware 顺序错导致 cache 不生效

数组靠前的是外层。cache 应该尽量靠外（尽早判断缓存命中），rate limit 也应该放在 cache 之后（命中缓存不占额度）。

流式 Fallback 的半截流

上面提过，流失败了切 fallback 会从头重来，用户看到的是「文字闪一下然后重新写」。解决办法：只对前几个 chunk 做 fallback（超过就算了），或者业务层做「生成失败继续，之前的文字保留」。

Cache key 没考虑 tools

有 tools 的调用，相同 prompt 但 tools 不同，结果差很多。cache key 要把 params.tools 的 schema 指纹也考虑进去。

Middleware 吞错误

中间件里用 try/catch 时要小心。如果你吞了错误返回一个假结果，上层完全不知情。除非是 Fallback 这种明确的降级，否则一定要让错误抛出来。

9. 小结

wrapLanguageModel 给 model 挂上中间件，新 model 仍然是标准 LanguageModelV2
核心三个 hook：wrapGenerate / wrapStream / transformParams
常用中间件：Logging / Cache / RateLimit / Retry / Fallback / DefaultSettings
数组组合时越靠前越外层，顺序设计很关键
生产栈常见组合：telemetry → rate limit → retry → fallback → defaults
流式下的 Fallback 要想清楚「半截流重来」的接受度
中间件不要吞错误，除非是明确的降级

下一篇进入后端三连的最后一篇——可观测性。Telemetry / OpenTelemetry / Langfuse 怎么接，怎么在生产追踪一次 chat 的完整链路。