TS 中的 LLM 应用开发：AI SDK、流式响应、工具调用与类型安全

引言

把 LLM 接入 TypeScript 应用，难点往往不在「调用 API」，而在类型安全：模型的输出是自由文本，工具调用的参数是运行时才知道的 JSON，流式响应把一次完整结果拆成几十个增量片段。如何让这些不确定性在编译期就有约束，是 LLM 应用工程化的核心问题。

本文聚焦 TypeScript 中的 LLM 应用开发：从类型挑战讲起，覆盖 Vercel AI SDK 的核心抽象、流式响应、结构化输出、工具调用、多轮状态、多提供商路由、RAG 类型，最后给出错误处理与生产实践。

前置：运行时验证、zod、异步控制。

目录

• 1. LLM 应用的类型挑战
• 2. AI SDK 概览与核心抽象
• 3. 流式响应与增量解析
• 4. 结构化输出与 Schema
• 5. 工具调用与函数类型
• 6. 多轮对话与状态管理
• 7. 多提供商抽象与模型路由
• 8. RAG 与向量检索类型
• 9. 错误处理、重试与成本
• 10. 生产实践与可观测
• 延伸阅读

1. LLM 应用的类型挑战

1.1 三道不确定性

输出不确定：模型返回自由文本，形状无法用静态类型保证；工具参数不确定：模型生成的 JSON 参数需运行时校验；流式不确定：一次结果被拆成多个增量片段，顺序与边界需处理。

1.2 类型安全的落点

输入侧用类型约束 prompt 变量与消息结构；输出侧用 schema 约束结构化输出、parse 后收窄类型；工具侧用 schema 声明参数、执行前校验。

1.3 核心原则

LLM 的输出本质是外部输入，应像对待 HTTP 响应一样对待它：声明期望的 schema，运行时校验，失败即重试或降级。类型只是声明，校验才是保障。

一句话总结：LLM 的输出是外部输入，必须用 schema 在运行时校验——类型负责声明期望，校验负责保证事实。

2. AI SDK 概览与核心抽象

2.1 核心函数

generateText 一次性生成完整文本；generateObject 按 schema 生成结构化对象（类型安全）；streamText 流式生成文本；streamObject 流式生成结构化对象；tool 声明一个可被模型调用的工具（含参数 schema）。

2.2 最小示例

import { generateText } from "ai"
import { anthropic } from "@ai-sdk/anthropic"

const { text } = await generateText({
  model: anthropic("claude-sonnet-4-5"),
  prompt: "用一句话解释事件循环",
})
console.log(text)

2.3 类型从何而来

AI SDK 的关键设计是把 schema 作为类型的来源：传入 zod schema，返回值的类型自动从 schema 推断，无需手写泛型参数。

import { generateObject } from "ai"
import { z } from "zod"

const schema = z.object({
  title: z.string(),
  tags: z.array(z.string()),
  score: z.number().min(0).max(1),
})

const { object } = await generateObject({ model, schema, prompt })
// object: { title: string; tags: string[]; score: number }

一句话总结：AI SDK 把 zod schema 当作类型的唯一来源——schema 一变，返回值的类型自动跟着变，校验与类型永不失配。

3. 流式响应与增量解析

3.1 为什么流式

LLM 生成是逐 token 的，非流式要等全部完成才返回，首字节延迟高。流式让用户尽快看到内容，是聊天类体验的基础。

3.2 服务端流式

import { streamText } from "ai"

export async function POST(req: Request) {
  const { messages } = await req.json()
  const result = streamText({ model, messages })
  return result.toDataStreamResponse()
}

3.3 客户端消费

import { useChat } from "ai/react"

export function Chat() {
  const { messages, input, handleInputChange, handleSubmit } = useChat()
  return (
    <form onSubmit={handleSubmit}>
      <input value={input} onChange={handleInputChange} />
      {messages.map((m) => <p key={m.id}>{m.role}: {m.content}</p>)}
    </form>
  )
}

3.4 增量处理的坑

不能假设每块都是完整 JSON，要用 SDK 的解析而非手写 split；网络中断要能续传或重试，记录已收到的片段；前端渲染要节流，每 token 都 setState 会卡；流结束信号要显式处理，在 onFinish 里落库。

一句话总结：流式让首字节更快，但增量边界与中断恢复要专门处理——用 SDK 的流解析而非手写 split，前端渲染记得节流。

4. 结构化输出与 Schema

4.1 从文本到对象

自由文本难以程序化消费，结构化输出让模型直接产出符合 schema 的 JSON。

const { object } = await generateObject({
  model,
  schema: z.object({
    sentiment: z.enum(["positive", "neutral", "negative"]),
    confidence: z.number(),
    reasons: z.array(z.string()).max(3),
  }),
  prompt: `分析这条评论的情绪：${comment}`,
})

4.2 schema 设计要点

用 enum 而非自由字符串以收窄取值；用 .max()/.min() 限制数组与数值范围；避免过深嵌套（模型容易漏字段）；字段名与描述要清晰，它们等于给模型的提示。

4.3 校验失败的处理

try {
  const { object } = await generateObject({ model, schema, prompt })
  return object
} catch (err) {
  // 模型可能返回不符合 schema 的内容
  logger.warn({ err }, "structured output invalid")
  return fallback
}

4.4 与运行时验证的关系

generateObject 内部会用 schema 校验，但跨进程或落库后的数据仍要再校验一次——schema 是契约，不因来源可信就免检。

一句话总结：结构化输出用 schema 约束模型产出，enum 与范围限制能显著提升稳定性——校验失败要降级，且落库/跨进程后仍需复检。

5. 工具调用与函数类型

5.1 声明工具

工具让模型能「调用你的函数」，参数由 schema 声明，执行前自动校验：

import { tool } from "ai"
import { z } from "zod"

const weather = tool({
  description: "查询指定城市的当前天气",
  parameters: z.object({
    city: z.string().describe("城市名，如 Beijing"),
    unit: z.enum(["celsius", "fahrenheit"]).default("celsius"),
  }),
  execute: async ({ city, unit }) => {
    return { city, temp: 22, unit }
  },
})

5.2 在生成中启用

const result = await generateText({
  model,
  tools: { weather },
  prompt: "北京现在多少度？",
})

SDK 会：把工具 schema 转成模型可理解的描述 → 模型返回工具调用 → 执行 execute → 把结果回灌给模型 → 生成最终答案。

5.3 多步工具调用

generateText({ model, tools: { weather, search }, maxSteps: 5, prompt }) 中的 maxSteps 允许多轮「调用-回灌」循环，模型可先查天气再搜索新闻，直到给出最终答案。

5.4 安全要点

工具参数必须校验（schema 自动做，但 execute 内部仍需防注入）；危险操作（写库、发邮件）要人工确认或权限门控；工具描述要精确，否则模型会误用或漏用；限制 maxSteps，避免无限循环。

一句话总结：工具把「模型输出」变成「带 schema 校验的函数调用」——用 maxSteps 控制多轮，危险操作必须加确认与权限门控。

6. 多轮对话与状态管理

6.1 消息类型

type Role = "system" | "user" | "assistant" | "tool"

interface Message {
  id: string
  role: Role
  content: string
  toolInvocations?: ToolInvocation[]
}

6.2 状态存哪里

客户端内存刷新即丢，适合演示；本地存储按会话 ID 存，适合单设备；服务端 DB 跨设备同步且可审计，适合生产。

6.3 服务端会话示例

async function chat(sessionId: string, userInput: string) {
  const history = await db.messages.findMany({ where: { sessionId } })
  const messages = [...history.map(toModelMessage), { role: "user", content: userInput }]
  const result = streamText({ model, messages })
  await db.messages.create({ data: { sessionId, role: "user", content: userInput } })
  return result.toDataStreamResponse({ onFinish: async ({ text }) =>
    db.messages.create({ data: { sessionId, role: "assistant", content: text } }) })
}

6.4 上下文窗口管理

超长对话要截断或摘要，避免超出上下文窗口；system prompt 尽量稳定，利于提供商侧缓存；工具结果通常很大，落库但不必全量回灌。

一句话总结：多轮对话的关键是「消息历史持久化 + 上下文窗口管理」——生产用服务端 DB 存储便于审计，超长历史要截断或摘要。

7. 多提供商抽象与模型路由

7.1 统一接口

AI SDK 用统一的 model 参数抽象不同提供商，切换只改一行：

import { anthropic } from "@ai-sdk/anthropic"
import { openai } from "@ai-sdk/openai"
import { google } from "@ai-sdk/google"

const models = {
  fast: openai("gpt-4o-mini"),
  balanced: anthropic("claude-sonnet-4-5"),
  reasoning: anthropic("claude-opus-4-1"),
}

7.2 按任务路由

分类任务用便宜模型（models.fast）、摘要用 models.balanced、复杂推理才用 models.reasoning——用 switch (task) 决定返回哪个模型，把成本花在真正需要的地方。

7.3 降级与容灾

async function withFallback(fn: (m: LanguageModel) => Promise<unknown>) {
  try { return await fn(models.balanced) }
  catch (err) {
    logger.warn({ err }, "primary model failed, falling back")
    return await fn(models.fast)
  }
}

7.4 抽象的边界

能统一的是 generate/stream 的接口、消息结构与工具调用；难统一的是特定提供商的独有能力（缓存、批处理、推理控制）——这些用可选参数或适配层隔离。

一句话总结：多提供商抽象让模型可路由、可降级——按任务复杂度选模型，主模型失败自动 fallback，提供商独有能力用适配层隔离。

8. RAG 与向量检索类型

8.1 RAG 的流程

离线侧：文档 → 分块 → 向量化 → 存入向量库；在线侧：问题 → 向量化 → 相似检索 → 拼进 prompt → 生成答案。

8.2 类型化检索结果

interface Chunk {
  id: string
  text: string
  source: string
  score: number   // 相似度
}

async function retrieve(query: string, topK = 5): Promise<Chunk[]> {
  const embedding = await embed(query)
  const rows = await vectorStore.search(embedding, topK)
  return rows.map(toChunk)
}

8.3 拼装 prompt

检索到 chunks 后，用 chunks.map((c, i) => "[${i + 1}] ${c.text}").join("\n\n") 拼成带编号的上下文，再把它与问题一起塞进 prompt，并明确要求「仅依据资料回答、标注来源编号」。

8.4 常见坑

分块过大稀释相关度、过小丢上下文；不重排时仅按向量相似度可能把噪声排在前面，应加 rerank；不返回来源会让用户无法验证；忽略 token 预算则检索结果拼太长会挤占回答空间。

一句话总结：RAG 用向量检索把相关资料拼进 prompt——分块大小、重排、来源引用与 token 预算是最容易踩的四类坑。

9. 错误处理、重试与成本

9.1 常见错误分类

限流（429）用指数退避重试；超时则重试或缩短请求；上下文超限就截断历史或摘要；schema 不符需重试并加强提示；内容过滤则降级或转人工。

9.2 退避重试

async function withRetry<T>(fn: () => Promise<T>, max = 3): Promise<T> {
  for (let i = 0, delay = 500; ; i++, delay *= 2) {
    try { return await fn() } catch (err) {
      if (i >= max - 1) throw err
      await new Promise((r) => setTimeout(r, delay))   // 指数退避
    }
  }
}

9.3 成本控制

按任务选模型（分类/抽取用便宜模型，推理才用强模型）；缓存相同 prompt 的结果（注意 system 稳定性）；用 maxTokens 限制输出；精简 prompt，上下文只放必要内容；统计每次调用的 input/output token。

9.4 用量与成本记录

generateText 返回的 result.usage 含 promptTokens/completionTokens/totalTokens，result.response.modelId 记录实际模型——把两者连同延迟写进日志，即可按模型与任务分析成本。

一句话总结：LLM 调用要按错误类型分别处理（限流退避、超限截断、schema 重试）——成本控制靠「按任务选模型 + 缓存 + 限制输出 + 记录用量」。

10. 生产实践与可观测

10.1 落地清单

所有结构化输出走 schema 校验、失败降级；工具调用加权限门控与 maxSteps 限制；消息历史持久化、超长做截断或摘要；多提供商路由加失败降级；记录每次调用的 token 用量、延迟与模型。

10.2 可观测指标

10.3 安全与合规

用户输入与检索内容都可能含提示注入，要做隔离与清洗；不把敏感数据拼进 prompt，或先脱敏；输出内容按场景做过滤；保留审计日志，满足可追溯要求。

10.4 与追踪集成

把每次 LLM 调用当作一个 Span，记录模型、token 用量与工具调用，便于在链路中定位慢调用与异常——这与服务的分布式追踪天然衔接。

一句话总结：LLM 应用的生产化 = schema 校验 + 工具门控 + 状态持久化 + 多模型路由 + 用量可观测——把每次调用当 Span 埋点，注入防护与审计不可省。

延伸阅读

• zod schema 与运行时校验
• 运行时验证与类型安全
• 异步控制与流式处理
• Node 后端的服务化实践
• 事件流与增量数据处理
• TypeScript 专题 — TypeScript 专题