跳到主要内容

前端接入大模型 API

问题​

前端如何接入 OpenAI / Anthropic 等大模型 API？如何处理流式响应、Function Calling、结构化输出和 Token 计费？

答案​

接入大模型 API 是 AI 应用开发的基础能力。本文以 OpenAI 和 Anthropic 两大主流 Provider 为例，系统讲解从 API 调用、流式解析、Function Calling、结构化输出到成本控制的完整工程实践。

安全警告

永远不要把 API Key 放在前端代码中！ 前端打包产物完全透明，Key 泄露将导致巨额费用。必须通过后端/BFF 层转发请求。本文的直接调用示例仅用于理解 API 格式，生产环境必须经过后端代理。

一、API 请求格式对比​

理解两大 Provider 的 API 设计差异是接入的第一步。核心差异在于 消息结构 和 流式事件格式。

OpenAI Chat Completions API​

types/openai.ts
// OpenAI 请求格式
interface OpenAIChatRequest {
  model: string;                    // "gpt-4o" | "gpt-4o-mini" | "o3" | "o4-mini"
  messages: OpenAIMessage[];
  temperature?: number;             // 0-2，默认 1
  top_p?: number;                   // 核采样，与 temperature 二选一
  max_tokens?: number;              // 最大输出 token 数（可选，有默认值）
  stream?: boolean;                 // 是否流式返回
  tools?: OpenAITool[];             // Function Calling 工具定义
  tool_choice?: 'auto' | 'none' | 'required' | { type: 'function'; function: { name: string } };
  response_format?: ResponseFormat; // 结构化输出
  stop?: string[];                  // 停止序列
  seed?: number;                    // 可复现输出
  n?: number;                       // 生成几个候选回复
  presence_penalty?: number;        // -2 到 2，降低重复话题
  frequency_penalty?: number;       // -2 到 2，降低重复 token
}

// OpenAI 消息：system prompt 在 messages 数组中
interface OpenAIMessage {
  role: 'system' | 'user' | 'assistant' | 'tool';
  content: string | ContentPart[];  // 文本或多模态内容
  name?: string;                    // 可选的发言者名称
  tool_calls?: ToolCall[];          // assistant 消息中的工具调用
  tool_call_id?: string;            // tool 消息需要关联到具体的调用
}

type ContentPart =
  | { type: 'text'; text: string }
  | { type: 'image_url'; image_url: { url: string; detail?: 'low' | 'high' | 'auto' } };

interface ToolCall {
  id: string;                       // "call_xxx" 格式的唯一 ID
  type: 'function';
  function: { name: string; arguments: string }; // arguments 是 JSON 字符串
}

Anthropic Messages API​

types/anthropic.ts
// Anthropic 请求格式
interface AnthropicRequest {
  model: string;                    // "claude-sonnet-4-20250514" | "claude-opus-4-20250514"
  max_tokens: number;               // 必填！Anthropic 没有默认值
  system?: string | SystemContent[]; // system prompt 是独立字段，不在 messages 中
  messages: AnthropicMessage[];
  stream?: boolean;
  tools?: AnthropicTool[];
  tool_choice?: { type: 'auto' | 'any' | 'tool'; name?: string };
  temperature?: number;             // 0-1，默认 1
  top_p?: number;
  top_k?: number;                   // Anthropic 独有，限制采样词表大小
  thinking?: {                      // 扩展思考（Extended Thinking）
    type: 'enabled';
    budget_tokens: number;          // 思考预算
  };
  metadata?: { user_id?: string };  // 用于滥用检测
}

// Anthropic 消息：只有 user 和 assistant 两种角色
interface AnthropicMessage {
  role: 'user' | 'assistant';
  content: string | ContentBlock[];
}

// Anthropic 的内容是块级结构（ContentBlock），这是与 OpenAI 最大的架构差异
type ContentBlock =
  | { type: 'text'; text: string }
  | { type: 'image'; source: { type: 'base64'; media_type: string; data: string } }
  | { type: 'tool_use'; id: string; name: string; input: Record<string, unknown> }
  | { type: 'tool_result'; tool_use_id: string; content: string | ContentBlock[] }
  | { type: 'thinking'; thinking: string }  // 扩展思考内容
  | { type: 'redacted_thinking'; data: string }; // 加密的思考（安全场景）

核心差异对比​

特性OpenAIAnthropic
System Prompt放在 messages 中（role: 'system'）独立 system 字段
Max Tokens可选（有默认值）必填（无默认值）
消息角色system / user / assistant / tooluser / assistant（tool_result 在 user 中）
内容结构content: string | ContentPart[]content: string | ContentBlock[]
流式格式data: {...}\n\n（纯 SSE）event: xxx\ndata: {...}\n\n（事件驱动 SSE）
工具调用assistant 的 tool_calls 字段assistant 的 tool_use ContentBlock
工具结果独立的 tool 角色消息user 消息中的 tool_result block
推理模型切换独立模型（o3、o4-mini）同一模型开启 thinking 参数
认证方式Authorization: Bearer sk-xxxx-api-key: sk-ant-xxx + anthropic-version
Temperature 范围0-20-1
为什么 Anthropic 的 system prompt 是独立字段？

OpenAI 把 system 混在 messages 数组中，导致一些模糊性（比如多个 system 消息时的优先级）。Anthropic 将 system 独立出来，语义更清晰：system 是给模型的「元指令」，不属于对话历史的一部分。这在多轮对话中不会被上下文压缩掉。实际工程中，这个设计更加合理。

二、后端代理层（BFF）​

生产环境必须通过后端转发 LLM 请求。后端代理层承担 鉴权、限流、计费、日志、内容过滤 等职责。

app/api/chat/route.ts
import { auth } from '@/lib/auth';
import { rateLimit } from '@/lib/rate-limit';
import { tokenBudget } from '@/lib/token-budget';
import { contentFilter } from '@/lib/content-filter';

export async function POST(request: Request): Promise<Response> {
  // 1. 用户认证
  const session = await auth();
  if (!session?.user) {
    return new Response('Unauthorized', { status: 401 });
  }

  const { messages, model, provider = 'openai' } = await request.json();

  // 2. 输入内容过滤（防 Prompt 注入）
  const filterResult = await contentFilter.check(messages);
  if (filterResult.blocked) {
    return Response.json({ error: filterResult.reason }, { status: 400 });
  }

  // 3. 频率限制
  const { success, remaining } = await rateLimit.check(session.user.id, {
    maxRequests: 60,
    window: '1m',
  });
  if (!success) {
    return new Response('Rate limit exceeded', {
      status: 429,
      headers: { 'X-RateLimit-Remaining': String(remaining) },
    });
  }

  // 4. Token 预算检查
  const budget = await tokenBudget.check(session.user.id);
  if (budget.exceeded) {
    return Response.json({
      error: `本月 Token 用量已达上限（${budget.used}/${budget.limit}）`,
    }, { status: 402 });
  }

  // 5. 构建 LLM 请求
  const apiConfig = getProviderConfig(provider);
  const requestBody = buildLLMRequest(provider, { messages, model });

  // 6. 转发请求到 LLM API
  const llmResponse = await fetch(apiConfig.baseUrl, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      ...apiConfig.authHeaders,
    },
    body: JSON.stringify(requestBody),
  });

  if (!llmResponse.ok) {
    const error = await llmResponse.text();
    console.error(`LLM API error [${provider}]:`, llmResponse.status, error);
    return new Response(error, { status: llmResponse.status });
  }

  // 7. 流式转发给前端（透传 LLM 的 ReadableStream）
  // 同时在后端异步统计 Token 用量
  const [streamForClient, streamForMetrics] = llmResponse.body!.tee();

  // 异步统计 Token（不阻塞响应）
  collectTokenUsage(streamForMetrics, session.user.id, model).catch(console.error);

  return new Response(streamForClient, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive',
      'X-RateLimit-Remaining': String(remaining),
    },
  });
}

// Provider 配置
function getProviderConfig(provider: string) {
  const configs: Record<string, { baseUrl: string; authHeaders: Record<string, string> }> = {
    openai: {
      baseUrl: 'https://api.openai.com/v1/chat/completions',
      authHeaders: { Authorization: `Bearer ${process.env.OPENAI_API_KEY}` },
    },
    anthropic: {
      baseUrl: 'https://api.anthropic.com/v1/messages',
      authHeaders: {
        'x-api-key': process.env.ANTHROPIC_API_KEY!,
        'anthropic-version': '2024-10-22',
      },
    },
    deepseek: {
      baseUrl: 'https://api.deepseek.com/chat/completions',
      authHeaders: { Authorization: `Bearer ${process.env.DEEPSEEK_API_KEY}` },
    },
  };
  return configs[provider] ?? configs.openai;
}

// 构建不同 Provider 的请求体
function buildLLMRequest(
  provider: string,
  params: { messages: Array<{ role: string; content: string }>; model: string }
) {
  const { messages, model } = params;
  const systemMsg = messages.find(m => m.role === 'system');
  const otherMsgs = messages.filter(m => m.role !== 'system');

  if (provider === 'anthropic') {
    return {
      model,
      system: systemMsg?.content,
      messages: otherMsgs,
      max_tokens: 4096,
      stream: true,
    };
  }

  // OpenAI / DeepSeek（兼容 OpenAI 格式）
  return {
    model,
    messages,
    max_tokens: 4096,
    stream: true,
  };
}
ReadableStream.tee() 技巧

tee() 方法将一个可读流分叉为两个独立的流。一个给客户端响应，一个用于后端异步处理（如 Token 统计、日志记录）。这样不会阻塞流式响应的转发速度。

三、流式响应解析​

流式响应是 AI 应用最重要的体验优化——用户无需等待完整回复，首个 Token 到达即可开始阅读。

OpenAI SSE 格式​

lib/openai-stream.ts
// OpenAI 流式响应格式：
// data: {"id":"chatcmpl-xxx","choices":[{"delta":{"role":"assistant"}}]}
// data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"Hello"}}]}
// data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":" world"}}]}
// data: {"id":"chatcmpl-xxx","choices":[{"delta":{}}],"finish_reason":"stop"}
// data: [DONE]
//
// 特点：
// 1. 每行以 "data: " 前缀开始
// 2. 每个事件后跟两个换行 \n\n
// 3. 最后一条是 "data: [DONE]"
// 4. 增量内容在 delta.content 中

interface OpenAIStreamChunk {
  id: string;
  object: 'chat.completion.chunk';
  choices: Array<{
    index: number;
    delta: {
      role?: 'assistant';
      content?: string;
      tool_calls?: Array<{
        index: number;
        id?: string;                   // 只在第一个 chunk 出现
        function?: {
          name?: string;               // 只在第一个 chunk 出现
          arguments?: string;          // 逐步拼接的 JSON 字符串
        };
      }>;
    };
    finish_reason: 'stop' | 'tool_calls' | 'length' | null;
  }>;
  usage?: {                            // 只在最后一个 chunk 出现（需开启 stream_options）
    prompt_tokens: number;
    completion_tokens: number;
    total_tokens: number;
  };
}

// 通用 SSE 行解析器：处理粘包和分包
async function* readSSELines(response: Response): AsyncGenerator<string> {
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    // stream: true 确保多字节字符（如中文）不会被截断
    buffer += decoder.decode(value, { stream: true });

    // 按换行分割，最后一个可能不完整，留到下次
    const lines = buffer.split('\n');
    buffer = lines.pop() || '';

    for (const line of lines) {
      const trimmed = line.trim();
      if (trimmed) yield trimmed;
    }
  }

  // 处理最后残留的数据
  if (buffer.trim()) yield buffer.trim();
}

// OpenAI 流解析器
async function* parseOpenAIStream(
  response: Response
): AsyncGenerator<{
  type: 'text' | 'tool_call_start' | 'tool_call_args' | 'usage' | 'done';
  content: string;
  meta?: Record<string, unknown>;
}> {
  for await (const line of readSSELines(response)) {
    if (!line.startsWith('data: ')) continue;

    const data = line.slice(6); // 去掉 "data: " 前缀
    if (data === '[DONE]') {
      yield { type: 'done', content: '' };
      return;
    }

    try {
      const chunk: OpenAIStreamChunk = JSON.parse(data);
      const choice = chunk.choices[0];

      // 文本内容
      if (choice?.delta?.content) {
        yield { type: 'text', content: choice.delta.content };
      }

      // 工具调用开始（包含 id 和 function.name）
      if (choice?.delta?.tool_calls) {
        for (const tc of choice.delta.tool_calls) {
          if (tc.id && tc.function?.name) {
            yield {
              type: 'tool_call_start',
              content: '',
              meta: { id: tc.id, name: tc.function.name, index: tc.index },
            };
          }
          if (tc.function?.arguments) {
            yield {
              type: 'tool_call_args',
              content: tc.function.arguments,
              meta: { index: tc.index },
            };
          }
        }
      }

      // 使用量统计（需在请求中设置 stream_options: { include_usage: true }）
      if (chunk.usage) {
        yield {
          type: 'usage',
          content: '',
          meta: {
            promptTokens: chunk.usage.prompt_tokens,
            completionTokens: chunk.usage.completion_tokens,
          },
        };
      }
    } catch {
      // 跳过解析失败的行（如注释行或空数据）
    }
  }
}

Anthropic SSE 格式​

lib/anthropic-stream.ts
// Anthropic 流式响应格式（事件驱动，比 OpenAI 更结构化）：
//
// event: message_start
// data: {"type":"message_start","message":{"id":"msg_xxx","model":"claude-sonnet-4-20250514","usage":{"input_tokens":25}}}
//
// event: content_block_start
// data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
//
// event: content_block_delta
// data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
//
// event: content_block_stop
// data: {"type":"content_block_stop","index":0}
//
// event: message_delta
// data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":42}}
//
// event: message_stop
// data: {"type":"message_stop"}
//
// 特点：
// 1. 有 event: 前缀标识事件类型
// 2. 内容按 content_block 组织（可以有多个 block）
// 3. 工具调用的参数通过 input_json_delta 逐步传输
// 4. thinking 内容也是独立的 content_block

type AnthropicStreamEvent =
  | { type: 'message_start'; message: { id: string; model: string; usage: { input_tokens: number } } }
  | { type: 'content_block_start'; index: number; content_block: ContentBlockStart }
  | { type: 'content_block_delta'; index: number; delta: ContentDelta }
  | { type: 'content_block_stop'; index: number }
  | { type: 'message_delta'; delta: { stop_reason: string }; usage: { output_tokens: number } }
  | { type: 'message_stop' };

type ContentBlockStart =
  | { type: 'text'; text: string }
  | { type: 'thinking'; thinking: string }
  | { type: 'tool_use'; id: string; name: string; input: Record<string, unknown> };

type ContentDelta =
  | { type: 'text_delta'; text: string }
  | { type: 'thinking_delta'; thinking: string }
  | { type: 'input_json_delta'; partial_json: string };

// Anthropic 流解析器
async function* parseAnthropicStream(
  response: Response
): AsyncGenerator<{
  type: 'text' | 'thinking' | 'tool_use_start' | 'tool_use_args' | 'usage' | 'done';
  content: string;
  meta?: Record<string, unknown>;
}> {
  let inputTokens = 0;
  let outputTokens = 0;

  for await (const line of readSSELines(response)) {
    // Anthropic 用 event: 行标识事件类型，但我们直接从 data 中解析 type
    if (!line.startsWith('data: ')) continue;

    try {
      const event: AnthropicStreamEvent = JSON.parse(line.slice(6));

      switch (event.type) {
        case 'message_start':
          inputTokens = event.message.usage.input_tokens;
          break;

        case 'content_block_start':
          // 工具调用开始
          if (event.content_block.type === 'tool_use') {
            yield {
              type: 'tool_use_start',
              content: '',
              meta: {
                id: event.content_block.id,
                name: event.content_block.name,
                index: event.index,
              },
            };
          }
          break;

        case 'content_block_delta':
          if (event.delta.type === 'text_delta') {
            yield { type: 'text', content: event.delta.text };
          } else if (event.delta.type === 'thinking_delta') {
            yield { type: 'thinking', content: event.delta.thinking };
          } else if (event.delta.type === 'input_json_delta') {
            yield {
              type: 'tool_use_args',
              content: event.delta.partial_json,
              meta: { index: event.index },
            };
          }
          break;

        case 'message_delta':
          outputTokens = event.usage.output_tokens;
          yield {
            type: 'usage',
            content: '',
            meta: { promptTokens: inputTokens, completionTokens: outputTokens },
          };
          break;

        case 'message_stop':
          yield { type: 'done', content: '' };
          return;
      }
    } catch {
      // 跳过解析失败的行
    }
  }
}

统一流式处理 Hook​

hooks/useChat.ts
import { useState, useCallback, useRef } from 'react';

interface Message {
  id: string;
  role: 'user' | 'assistant';
  content: string;
  thinking?: string;          // 推理过程（Claude extended thinking / DeepSeek R1）
  toolCalls?: ToolCallInfo[]; // 工具调用信息
  status: 'pending' | 'streaming' | 'done' | 'error';
  usage?: { promptTokens: number; completionTokens: number };
}

interface ToolCallInfo {
  id: string;
  name: string;
  args: string;
  result?: string;
  status: 'calling' | 'done' | 'error';
}

interface UseChatOptions {
  provider: 'openai' | 'anthropic';
  model: string;
  systemPrompt?: string;
  onError?: (error: Error) => void;
  onFinish?: (message: Message) => void;
}

export function useChat(options: UseChatOptions) {
  const { provider, model, systemPrompt, onError, onFinish } = options;
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const abortRef = useRef<AbortController | null>(null);

  // 更新特定消息的 helper
  const updateMessage = useCallback((id: string, update: Partial<Message>) => {
    setMessages(prev =>
      prev.map(m => (m.id === id ? { ...m, ...update } : m))
    );
  }, []);

  const sendMessage = useCallback(async (content: string) => {
    const userMsg: Message = {
      id: crypto.randomUUID(),
      role: 'user',
      content,
      status: 'done',
    };
    const assistantMsg: Message = {
      id: crypto.randomUUID(),
      role: 'assistant',
      content: '',
      status: 'pending',
    };

    setMessages(prev => [...prev, userMsg, assistantMsg]);
    setIsStreaming(true);
    abortRef.current = new AbortController();

    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          provider,
          model,
          messages: [
            ...(systemPrompt ? [{ role: 'system', content: systemPrompt }] : []),
            ...messages.map(m => ({ role: m.role, content: m.content })),
            { role: 'user', content },
          ],
        }),
        signal: abortRef.current.signal,
      });

      if (!response.ok) {
        const errorText = await response.text();
        throw new Error(`HTTP ${response.status}: ${errorText}`);
      }

      // 根据 provider 选择解析器
      const parser = provider === 'anthropic'
        ? parseAnthropicStream(response)
        : parseOpenAIStream(response);

      let fullContent = '';
      let fullThinking = '';

      for await (const chunk of parser) {
        switch (chunk.type) {
          case 'text':
            fullContent += chunk.content;
            updateMessage(assistantMsg.id, {
              content: fullContent,
              status: 'streaming',
            });
            break;

          case 'thinking':
            fullThinking += chunk.content;
            updateMessage(assistantMsg.id, { thinking: fullThinking });
            break;

          case 'usage':
            updateMessage(assistantMsg.id, {
              usage: chunk.meta as Message['usage'],
            });
            break;
        }
      }

      const finalMsg = { ...assistantMsg, content: fullContent, thinking: fullThinking, status: 'done' as const };
      updateMessage(assistantMsg.id, { status: 'done' });
      onFinish?.(finalMsg);
    } catch (error) {
      if ((error as Error).name === 'AbortError') {
        updateMessage(assistantMsg.id, { status: 'done' }); // 用户手动取消
        return;
      }
      updateMessage(assistantMsg.id, {
        content: '生成失败，请重试',
        status: 'error',
      });
      onError?.(error as Error);
    } finally {
      setIsStreaming(false);
    }
  }, [messages, provider, model, systemPrompt, onError, onFinish, updateMessage]);

  const stop = useCallback(() => {
    abortRef.current?.abort();
    setIsStreaming(false);
  }, []);

  // 重试最后一条消息
  const retry = useCallback(async () => {
    const lastUserMsg = [...messages].reverse().find(m => m.role === 'user');
    if (!lastUserMsg) return;
    setMessages(prev => prev.slice(0, -2)); // 移除最后的 user + assistant
    await sendMessage(lastUserMsg.content);
  }, [messages, sendMessage]);

  // 清空对话
  const clear = useCallback(() => {
    setMessages([]);
  }, []);

  return { messages, isStreaming, sendMessage, stop, retry, clear };
}

四、Function Calling（函数调用）​

Function Calling 让 LLM 决定「何时调用什么工具、传什么参数」，但实际执行在你的代码中。LLM 不直接调用任何外部服务。

完整流程图​

OpenAI Function Calling​

lib/function-calling.ts
// 1. 定义工具（JSON Schema 描述参数）
const tools: OpenAITool[] = [
  {
    type: 'function',
    function: {
      name: 'get_weather',
      description: '获取指定城市的当前天气信息。当用户询问天气相关问题时调用。',
      // strict: true 开启 Structured Outputs 模式，保证参数格式正确
      strict: true,
      parameters: {
        type: 'object',
        properties: {
          city: { type: 'string', description: '城市名称，如"北京"' },
          unit: {
            type: 'string',
            enum: ['celsius', 'fahrenheit'],
            description: '温度单位',
          },
        },
        required: ['city', 'unit'],
        additionalProperties: false, // strict 模式必须
      },
    },
  },
  {
    type: 'function',
    function: {
      name: 'search_docs',
      description: '搜索技术文档库。当用户询问技术问题、需要查找文档时调用。',
      strict: true,
      parameters: {
        type: 'object',
        properties: {
          query: { type: 'string', description: '搜索关键词' },
          category: {
            type: 'string',
            enum: ['react', 'vue', 'node', 'css', 'js'],
            description: '限定搜索分类',
          },
          limit: { type: 'number', description: '返回结果数量，默认 5' },
        },
        required: ['query', 'category', 'limit'],
        additionalProperties: false,
      },
    },
  },
];

// 2. 工具执行器映射
const toolExecutors: Record<string, (args: Record<string, unknown>) => Promise<unknown>> = {
  get_weather: async ({ city, unit }) => {
    const res = await fetch(`/api/weather?city=${city}&unit=${unit}`);
    return res.json();
  },
  search_docs: async ({ query, category, limit }) => {
    const res = await fetch(`/api/search?q=${query}&cat=${category}&limit=${limit}`);
    return res.json();
  },
};

// 3. 完整的 Agent 循环（处理多轮工具调用）
async function agentLoop(
  messages: OpenAIMessage[],
  maxIterations: number = 5
): Promise<string> {
  let currentMessages = [...messages];

  for (let i = 0; i < maxIterations; i++) {
    const response = await fetch('/api/llm/openai', {
      method: 'POST',
      body: JSON.stringify({ model: 'gpt-4o', messages: currentMessages, tools }),
    });

    const data = await response.json();
    const assistantMessage = data.choices[0].message;
    currentMessages.push(assistantMessage);

    // 如果没有工具调用，返回最终回答
    if (!assistantMessage.tool_calls) {
      return assistantMessage.content;
    }

    // 并行执行所有工具调用（OpenAI 支持一次返回多个 tool_calls）
    const toolResults = await Promise.all(
      assistantMessage.tool_calls.map(async (toolCall: ToolCall) => {
        const executor = toolExecutors[toolCall.function.name];
        if (!executor) {
          return {
            role: 'tool' as const,
            tool_call_id: toolCall.id,
            content: JSON.stringify({ error: `Unknown tool: ${toolCall.function.name}` }),
          };
        }

        try {
          const args = JSON.parse(toolCall.function.arguments);
          const result = await executor(args);
          return {
            role: 'tool' as const,
            tool_call_id: toolCall.id,
            content: JSON.stringify(result),
          };
        } catch (error) {
          return {
            role: 'tool' as const,
            tool_call_id: toolCall.id,
            content: JSON.stringify({ error: (error as Error).message }),
          };
        }
      })
    );

    currentMessages.push(...toolResults);
    // 继续循环，让模型基于工具结果生成回答或继续调用工具
  }

  return '达到最大迭代次数，请精简问题后重试。';
}

Anthropic Tool Use​

lib/anthropic-tools.ts
// Anthropic 的工具定义格式
const anthropicTools: AnthropicTool[] = [
  {
    name: 'get_weather',
    description: '获取指定城市的当前天气信息',
    input_schema: {
      type: 'object',
      properties: {
        city: { type: 'string', description: '城市名称' },
      },
      required: ['city'],
    },
  },
];

// Anthropic 和 OpenAI 在工具调用结果的传递方式上有关键差异：
// OpenAI：用独立的 `tool` 角色消息
// Anthropic：将 tool_result 放在 `user` 消息的 content block 中

// Anthropic 工具调用的消息链示例：
const anthropicToolMessages: AnthropicMessage[] = [
  // 1. 用户提问
  { role: 'user', content: '北京今天天气怎么样？' },

  // 2. 模型返回工具调用（stop_reason: "tool_use"）
  {
    role: 'assistant',
    content: [
      { type: 'text', text: '让我查一下北京的天气信息。' },
      { type: 'tool_use', id: 'toolu_abc123', name: 'get_weather', input: { city: '北京' } },
    ],
  },

  // 3. 工具结果放在 user 消息中（注意：role 是 user！）
  {
    role: 'user',
    content: [
      {
        type: 'tool_result',
        tool_use_id: 'toolu_abc123',
        content: JSON.stringify({ temp: 25, weather: '晴', humidity: 45 }),
      },
    ],
  },

  // 4. 模型基于结果生成最终回答
  // { role: 'assistant', content: '北京今天天气晴朗，气温 25°C，湿度 45%...' }
];

五、结构化输出（Structured Output）​

让 LLM 返回严格符合 JSON Schema 的数据，而非自由文本。这是 AI 生成 UI 配置、表单数据提取等场景的关键能力。

lib/structured-output.ts
// === OpenAI Structured Outputs（推荐）===

// 方式1: JSON Mode（简单，但不保证 schema 合规）
const jsonModeResponse = await fetch('/api/llm/openai', {
  method: 'POST',
  body: JSON.stringify({
    model: 'gpt-4o',
    messages: [
      {
        role: 'user',
        content: '分析文本情感，返回 JSON 包含 sentiment/confidence/keywords：今天天气真好！',
      },
    ],
    response_format: { type: 'json_object' }, // 只保证返回合法 JSON
  }),
});

// 方式2: Structured Outputs（严格 schema 约束，100% 保证格式）
const structuredResponse = await fetch('/api/llm/openai', {
  method: 'POST',
  body: JSON.stringify({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: '分析"今天天气真好！"的情感' }],
    response_format: {
      type: 'json_schema',
      json_schema: {
        name: 'sentiment_analysis',
        strict: true,               // 严格模式
        schema: {
          type: 'object',
          properties: {
            sentiment: { type: 'string', enum: ['positive', 'negative', 'neutral'] },
            confidence: { type: 'number', description: '0-1 之间的置信度' },
            keywords: {
              type: 'array',
              items: { type: 'string' },
              description: '关键情感词',
            },
            explanation: { type: 'string', description: '分析理由' },
          },
          required: ['sentiment', 'confidence', 'keywords', 'explanation'],
          additionalProperties: false, // strict 模式必须设为 false
        },
      },
    },
  }),
});

const result = JSON.parse((await structuredResponse.json()).choices[0].message.content);
// 类型安全：{ sentiment: "positive", confidence: 0.95, keywords: ["好"], explanation: "..." }

前端实际应用场景​

lib/structured-use-cases.ts
import { generateObject } from 'ai'; // Vercel AI SDK
import { openai } from '@ai-sdk/openai';
import { z } from 'zod';

// 场景1：AI 表单自动填充（从自然语言中提取结构化信息）
const contactSchema = z.object({
  name: z.string().describe('联系人姓名'),
  email: z.string().email().describe('邮箱地址'),
  phone: z.string().describe('手机号码'),
  company: z.string().optional().describe('公司名称'),
});

async function extractContact(text: string) {
  const { object } = await generateObject({
    model: openai('gpt-4o'),
    schema: contactSchema,
    prompt: `从以下文本中提取联系人信息：\n${text}`,
  });
  return object; // 类型安全的 { name, email, phone, company? }
}

// 场景2：AI 内容分类和标签生成
const classificationSchema = z.object({
  category: z.enum(['bug', 'feature', 'question', 'docs']).describe('Issue 分类'),
  priority: z.enum(['critical', 'high', 'medium', 'low']).describe('优先级'),
  tags: z.array(z.string()).describe('相关技术标签'),
  assignee: z.string().optional().describe('建议指派人'),
  summary: z.string().max(100).describe('一句话摘要'),
});

// 场景3：AI 生成图表配置
const chartConfigSchema = z.object({
  chartType: z.enum(['line', 'bar', 'pie', 'scatter']).describe('图表类型'),
  title: z.string(),
  xAxis: z.object({ field: z.string(), label: z.string() }),
  yAxis: z.object({ field: z.string(), label: z.string() }),
  series: z.array(z.object({
    name: z.string(),
    field: z.string(),
    color: z.string().optional(),
  })),
});

// 场景4：AI 生成 UI 布局配置
const layoutSchema = z.object({
  layout: z.enum(['grid', 'list', 'kanban', 'table']),
  columns: z.array(z.object({
    key: z.string(),
    label: z.string(),
    width: z.number(),
    sortable: z.boolean(),
  })),
  defaultSort: z.object({
    field: z.string(),
    direction: z.enum(['asc', 'desc']),
  }).optional(),
});

六、Token 计费与成本控制​

AI 应用的 API 调用按 Token 计费，成本控制是工程化的重要环节。

lib/token-cost.ts
// 2025 年主流模型价格（每百万 Token）
const MODEL_PRICING: Record<string, { input: number; output: number }> = {
  // OpenAI
  'gpt-4o':       { input: 2.50,  output: 10.00 },
  'gpt-4o-mini':  { input: 0.15,  output: 0.60  },
  'o3':           { input: 10.00, output: 40.00 },
  'o4-mini':      { input: 1.10,  output: 4.40  },

  // Anthropic
  'claude-opus-4':   { input: 15.00, output: 75.00 },
  'claude-sonnet-4': { input: 3.00,  output: 15.00 },
  'claude-haiku':    { input: 0.25,  output: 1.25  },

  // DeepSeek
  'deepseek-v3':     { input: 0.27,  output: 1.10  },
  'deepseek-r1':     { input: 0.55,  output: 2.19  },
};

// 计算单次请求的费用
function calculateCost(
  model: string,
  promptTokens: number,
  completionTokens: number
): { cost: number; breakdown: string } {
  const pricing = MODEL_PRICING[model];
  if (!pricing) return { cost: 0, breakdown: 'Unknown model' };

  const inputCost = (promptTokens / 1_000_000) * pricing.input;
  const outputCost = (completionTokens / 1_000_000) * pricing.output;
  const total = inputCost + outputCost;

  return {
    cost: total,
    breakdown: `输入 ${promptTokens} tokens ($${inputCost.toFixed(4)}) + 输出 ${completionTokens} tokens ($${outputCost.toFixed(4)}) = $${total.toFixed(4)}`,
  };
}

// Token 用量统计服务
class TokenUsageTracker {
  async recordUsage(params: {
    userId: string;
    model: string;
    promptTokens: number;
    completionTokens: number;
    requestId: string;
  }): Promise<void> {
    const { cost } = calculateCost(params.model, params.promptTokens, params.completionTokens);

    await db.tokenUsage.create({
      data: {
        userId: params.userId,
        model: params.model,
        promptTokens: params.promptTokens,
        completionTokens: params.completionTokens,
        cost,
        requestId: params.requestId,
        createdAt: new Date(),
      },
    });
  }

  // 获取用户月度使用情况
  async getMonthlyUsage(userId: string): Promise<{
    totalTokens: number;
    totalCost: number;
    byModel: Record<string, { tokens: number; cost: number }>;
  }> {
    const startOfMonth = new Date();
    startOfMonth.setDate(1);
    startOfMonth.setHours(0, 0, 0, 0);

    const records = await db.tokenUsage.findMany({
      where: { userId, createdAt: { gte: startOfMonth } },
    });

    const byModel: Record<string, { tokens: number; cost: number }> = {};
    let totalTokens = 0;
    let totalCost = 0;

    for (const record of records) {
      const tokens = record.promptTokens + record.completionTokens;
      totalTokens += tokens;
      totalCost += record.cost;

      if (!byModel[record.model]) {
        byModel[record.model] = { tokens: 0, cost: 0 };
      }
      byModel[record.model].tokens += tokens;
      byModel[record.model].cost += record.cost;
    }

    return { totalTokens, totalCost, byModel };
  }
}

// Token 数估算（不需要精确时）
// 经验法则：英文约 1 token/word，中文约 1.5 token/字
function estimateTokens(text: string): number {
  const chineseChars = (text.match(/[\u4e00-\u9fff]/g) || []).length;
  const otherChars = text.length - chineseChars;
  return Math.ceil(chineseChars * 1.5 + otherChars / 4);
}

七、错误处理与重试策略​

LLM API 的错误处理需要区分可重试和不可重试的错误。

lib/api-error-handling.ts
// LLM API 常见错误码及处理策略
const ERROR_HANDLERS: Record<number, {
  message: string;
  retryable: boolean;
  action: string;
}> = {
  400: { message: '请求格式错误', retryable: false, action: '检查参数格式' },
  401: { message: 'API Key 无效或过期', retryable: false, action: '更新 API Key' },
  403: { message: '权限不足', retryable: false, action: '检查 API 权限和账户额度' },
  404: { message: '模型不存在', retryable: false, action: '检查模型名称' },
  429: { message: '请求频率超限', retryable: true, action: '使用 Retry-After 等待后重试' },
  500: { message: '服务端内部错误', retryable: true, action: '指数退避重试' },
  502: { message: '网关错误', retryable: true, action: '稍后重试' },
  503: { message: '服务暂不可用', retryable: true, action: '等待恢复' },
  529: { message: '服务过载（Anthropic 特有）', retryable: true, action: '等待后重试' },
};

// 指数退避重试（带抖动）
async function callWithRetry<T>(
  fn: () => Promise<T>,
  options: {
    maxRetries?: number;
    baseDelay?: number;
    maxDelay?: number;
    retryableStatuses?: number[];
  } = {}
): Promise<T> {
  const {
    maxRetries = 3,
    baseDelay = 1000,
    maxDelay = 30000,
    retryableStatuses = [429, 500, 502, 503, 529],
  } = options;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      const status = (error as { status?: number }).status;
      const isRetryable = status ? retryableStatuses.includes(status) : true;

      // 不可重试的错误直接抛出
      if (!isRetryable) throw error;

      // 最后一次也失败了
      if (attempt === maxRetries) throw error;

      // 计算等待时间
      let delay: number;

      // 429 错误优先使用服务端返回的 Retry-After
      const retryAfter = (error as { headers?: Headers }).headers?.get('retry-after');
      if (retryAfter) {
        delay = parseInt(retryAfter) * 1000;
      } else {
        // 指数退避 + 随机抖动（避免惊群效应）
        delay = Math.min(
          baseDelay * Math.pow(2, attempt) + Math.random() * 1000,
          maxDelay
        );
      }

      console.warn(`LLM API 请求失败 (${status})，${delay}ms 后重试 (${attempt + 1}/${maxRetries})`);
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }

  throw new Error('Unreachable');
}

// 超时 + 心跳检测（防止流式连接挂死）
async function robustStreamFetch(
  url: string,
  body: object,
  options: {
    totalTimeout?: number;     // 总超时（ms）
    heartbeatTimeout?: number; // 心跳超时（两个 chunk 之间的最大间隔）
  } = {}
): Promise<Response> {
  const { totalTimeout = 120_000, heartbeatTimeout = 30_000 } = options;
  const controller = new AbortController();

  // 总超时
  const totalTimer = setTimeout(() => controller.abort('Total timeout'), totalTimeout);

  const response = await fetch(url, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(body),
    signal: controller.signal,
  });

  clearTimeout(totalTimer);

  if (!response.ok) {
    const error = new Error(`HTTP ${response.status}`);
    (error as any).status = response.status;
    throw error;
  }

  // 包装 ReadableStream 加入心跳检测
  const originalStream = response.body!;
  let lastChunkTime = Date.now();

  const heartbeatStream = new ReadableStream({
    async start(streamController) {
      const reader = originalStream.getReader();

      const heartbeatCheck = setInterval(() => {
        if (Date.now() - lastChunkTime > heartbeatTimeout) {
          clearInterval(heartbeatCheck);
          streamController.error(new Error('Heartbeat timeout'));
          reader.cancel();
        }
      }, 5000);

      try {
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          lastChunkTime = Date.now();
          streamController.enqueue(value);
        }
        streamController.close();
      } catch (error) {
        streamController.error(error);
      } finally {
        clearInterval(heartbeatCheck);
      }
    },
  });

  return new Response(heartbeatStream, { headers: response.headers });
}

八、多 Provider 统一抽象​

lib/llm-client.ts
// 统一接口定义，屏蔽不同 Provider 的 API 差异
interface LLMClient {
  chat(params: ChatParams): Promise<ChatResponse>;
  stream(params: ChatParams): AsyncGenerator<StreamChunk>;
}

interface ChatParams {
  messages: Array<{ role: string; content: string }>;
  model: string;
  temperature?: number;
  maxTokens?: number;
  tools?: ToolDefinition[];
  structuredOutput?: {
    name: string;
    schema: Record<string, unknown>;
  };
}

interface StreamChunk {
  type: 'text' | 'thinking' | 'tool_call' | 'usage' | 'done';
  content: string;
  meta?: Record<string, unknown>;
}

interface ChatResponse {
  content: string;
  usage: { promptTokens: number; completionTokens: number };
  toolCalls?: Array<{ id: string; name: string; args: Record<string, unknown> }>;
}

// OpenAI 客户端
class OpenAIClient implements LLMClient {
  constructor(private baseUrl: string, private apiKey: string) {}

  async chat(params: ChatParams): Promise<ChatResponse> {
    const response = await fetch(`${this.baseUrl}/v1/chat/completions`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${this.apiKey}`,
      },
      body: JSON.stringify({
        model: params.model,
        messages: params.messages,
        temperature: params.temperature,
        max_tokens: params.maxTokens,
        ...(params.tools && {
          tools: params.tools.map(t => ({
            type: 'function',
            function: { name: t.name, description: t.description, parameters: t.parameters },
          })),
        }),
        ...(params.structuredOutput && {
          response_format: {
            type: 'json_schema',
            json_schema: {
              name: params.structuredOutput.name,
              strict: true,
              schema: params.structuredOutput.schema,
            },
          },
        }),
      }),
    });

    const data = await response.json();
    const message = data.choices[0].message;

    return {
      content: message.content || '',
      usage: {
        promptTokens: data.usage.prompt_tokens,
        completionTokens: data.usage.completion_tokens,
      },
      toolCalls: message.tool_calls?.map((tc: ToolCall) => ({
        id: tc.id,
        name: tc.function.name,
        args: JSON.parse(tc.function.arguments),
      })),
    };
  }

  async *stream(params: ChatParams): AsyncGenerator<StreamChunk> {
    const response = await fetch(`${this.baseUrl}/v1/chat/completions`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${this.apiKey}`,
      },
      body: JSON.stringify({
        ...this.buildRequestBody(params),
        stream: true,
        stream_options: { include_usage: true },
      }),
    });

    yield* parseOpenAIStream(response);
  }

  private buildRequestBody(params: ChatParams) {
    return {
      model: params.model,
      messages: params.messages,
      temperature: params.temperature,
      max_tokens: params.maxTokens,
    };
  }
}

// Anthropic 客户端
class AnthropicClient implements LLMClient {
  constructor(private baseUrl: string, private apiKey: string) {}

  async chat(params: ChatParams): Promise<ChatResponse> {
    const systemMsg = params.messages.find(m => m.role === 'system');
    const otherMsgs = params.messages.filter(m => m.role !== 'system');

    const response = await fetch(`${this.baseUrl}/v1/messages`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': this.apiKey,
        'anthropic-version': '2024-10-22',
      },
      body: JSON.stringify({
        model: params.model,
        system: systemMsg?.content,
        messages: otherMsgs,
        max_tokens: params.maxTokens || 4096,
        temperature: params.temperature,
        ...(params.tools && {
          tools: params.tools.map(t => ({
            name: t.name,
            description: t.description,
            input_schema: t.parameters,
          })),
        }),
      }),
    });

    const data = await response.json();
    const textBlock = data.content.find((b: ContentBlock) => b.type === 'text');
    const toolBlocks = data.content.filter((b: ContentBlock) => b.type === 'tool_use');

    return {
      content: textBlock?.text || '',
      usage: {
        promptTokens: data.usage.input_tokens,
        completionTokens: data.usage.output_tokens,
      },
      toolCalls: toolBlocks.map((tb: any) => ({
        id: tb.id,
        name: tb.name,
        args: tb.input,
      })),
    };
  }

  async *stream(params: ChatParams): AsyncGenerator<StreamChunk> {
    const systemMsg = params.messages.find(m => m.role === 'system');
    const otherMsgs = params.messages.filter(m => m.role !== 'system');

    const response = await fetch(`${this.baseUrl}/v1/messages`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'x-api-key': this.apiKey,
        'anthropic-version': '2024-10-22',
      },
      body: JSON.stringify({
        model: params.model,
        system: systemMsg?.content,
        messages: otherMsgs,
        max_tokens: params.maxTokens || 4096,
        stream: true,
      }),
    });

    yield* parseAnthropicStream(response);
  }
}

// 工厂函数：根据 Provider 创建对应的客户端
function createLLMClient(provider: string): LLMClient {
  switch (provider) {
    case 'openai':
      return new OpenAIClient(
        process.env.OPENAI_BASE_URL || 'https://api.openai.com',
        process.env.OPENAI_API_KEY!
      );
    case 'anthropic':
      return new AnthropicClient(
        process.env.ANTHROPIC_BASE_URL || 'https://api.anthropic.com',
        process.env.ANTHROPIC_API_KEY!
      );
    default:
      throw new Error(`Unknown provider: ${provider}`);
  }
}

常见面试问题​

Q1: OpenAI 和 Anthropic API 的核心差异是什么？​

答案：

核心差异有 6 点：

  1. System Prompt 位置：OpenAI 放在 messages 数组中（role: 'system'），Anthropic 是独立的 system 字段。Anthropic 的设计语义更清晰——system 是元指令，不属于对话历史
  2. Max Tokens：Anthropic 必填，OpenAI 可选有默认值
  3. 流式格式：OpenAI 用 data: {...}\n\n 纯 SSE 格式，最后 data: [DONE]；Anthropic 用 event: xxx\ndata: {...} 事件驱动格式，按 content_block 组织
  4. 工具调用：OpenAI 在 assistant 消息的 tool_calls 字段中，结果用独立的 tool 角色消息；Anthropic 用 tool_use content block，结果放在 user 消息的 tool_result block 中
  5. 推理能力：OpenAI 需要切换独立模型（o3、o4-mini），Anthropic 在同一模型上通过 thinking 参数开启扩展思考
  6. 认证：OpenAI 用 Authorization: Bearer header，Anthropic 用 x-api-key header + anthropic-version

Q2: 为什么不能在前端直接调用 LLM API？​

答案：

有 5 个核心原因：

  1. API Key 泄露：前端打包产物完全透明（JS 可被反编译），Key 被盗用会产生巨额费用。真实案例中有公司因此损失数万美元
  2. CORS 限制：OpenAI、Anthropic 的 API 均不允许浏览器直接跨域请求（没有配置 Access-Control-Allow-Origin）
  3. 无法限流：没有后端层就无法做用户级别的频率限制和 Token 预算控制
  4. 缺少审计：无法记录请求日志、统计用量、监控异常调用
  5. 灵活性差：无法在请求前注入 system prompt、做 Prompt 注入检测、做内容过滤

后端代理层（BFF）是 AI 应用的标配架构，承担鉴权、限流、计费、日志、安全过滤等职责。

Q3: 如何解析 SSE 流式响应？​

答案：

SSE（Server-Sent Events）是 LLM 流式输出的标准协议。前端用 fetch + ReadableStream 解析：

const response = await fetch('/api/chat', { method: 'POST', body: JSON.stringify(params) });
const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = '';

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split('\n');
  buffer = lines.pop() || ''; // 最后一行可能不完整

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = line.slice(6);
      if (data === '[DONE]') return;
      const chunk = JSON.parse(data);
      // 处理 chunk...
    }
  }
}

关键要点：

  • decoder.decode(value, { stream: true }) 确保中文等多字节字符不被截断
  • 需要维护 buffer 处理粘包/分包
  • 用 AsyncGenerator 封装可以让调用方用 for await...of 优雅消费

Q4: Function Calling 的完整流程是什么？​

答案：

Function Calling 分为 5 步：

  1. 定义工具：用 JSON Schema 描述每个工具的名称、描述和参数格式
  2. 发送请求：将 messages 和 tools 定义一起发给 LLM
  3. 模型决策：LLM 分析用户意图，决定是直接回复还是调用工具（返回 tool_calls）
  4. 执行工具：在你的代码中执行对应的函数（调 API、查数据库等），获取结果
  5. 二次请求：将工具结果追加到 messages 中再次发给 LLM，模型基于结果生成自然语言回答

关键理解：LLM 不直接调用任何外部服务。它只「决定」调用什么工具、传什么参数。实际的 HTTP 请求/数据库查询/代码执行都在你的后端完成。这是安全设计——你完全控制哪些工具可用、执行结果如何返回。

Q5: 什么是结构化输出？和 JSON Mode 有什么区别？​

答案：

  • JSON Mode (response_format: { type: 'json_object' })：保证返回合法 JSON，但不保证字段结构。如果你的 Prompt 写了"返回 name 和 age"，模型可能返回 { "name": "张三" } 缺少 age
  • Structured Outputs (response_format: { type: 'json_schema', json_schema: {...} })：提供严格的 JSON Schema 约束，模型 100% 保证 返回符合 schema 的数据。通过 strict: true + additionalProperties: false 实现

实际应用建议：需要可靠解析的场景（表单填充、数据提取、UI 配置生成）必须用 Structured Outputs。简单的对话场景可以用 JSON Mode。

Q6: 如何处理 LLM API 的 429 错误（频率限制）？​

答案：

429 错误表示请求频率超过了 API 的速率限制。处理策略：

  1. 读取 Retry-After header：服务端通常会返回建议的等待时间
  2. 指数退避重试：每次重试等待时间翻倍（1s → 2s → 4s）
  3. 随机抖动：在退避时间上加随机偏移，避免多个客户端同时重试（惊群效应）
  4. 最大重试次数：通常设 3 次，超过后返回友好错误提示
  5. 前端限流：在 BFF 层对每个用户做频率限制，避免触发上游 API 的 429
  6. 请求队列：使用队列串行化请求，控制并发数

Q7: 如何实现多模型/多 Provider 切换？​

答案：

使用策略模式 + 工厂模式，通过统一接口抽象不同 Provider 的差异：

  1. 定义统一接口 LLMClient：包含 chat() 和 stream() 方法
  2. 每个 Provider 实现一个 Client：OpenAIClient、AnthropicClient、DeepSeekClient
  3. 工厂函数 createLLMClient(provider) 返回对应实例
  4. 前端 UI 提供模型选择器，用户切换时只改变 provider + model 参数

需要统一的层面：

  • Stream chunk 格式（text / thinking / tool_call / done）
  • 错误码映射
  • 使用量统计（不同 Provider 的 usage 字段位置不同）
  • Tool calling 协议（OpenAI 的 tool_calls vs Anthropic 的 tool_use）

Q8: Token 是什么？如何估算和控制 Token 用量？​

答案：

Token 是 LLM 处理文本的基本单位，不等于字符也不等于单词。经验法则：

  • 英文：约 1 token ≈ 0.75 个单词（或 4 个字符）
  • 中文：约 1 个汉字 ≈ 1.5 个 token（因为 BPE 编码对中文不友好）

控制 Token 用量的策略：

  1. 设置 max_tokens：限制输出长度
  2. Prompt 压缩：滑动窗口保留最近 N 轮对话，或用摘要压缩旧对话
  3. 模型路由：简单问题用小模型（GPT-4o-mini、Haiku），复杂问题用大模型
  4. 缓存：语义缓存相似问题的回答
  5. Token 预算：后端设置每用户每日/每月的 Token 限额

Q9: 如何让不同 Provider 的流式响应在前端有统一的处理逻辑？​

答案：

核心是将不同 Provider 的流式格式统一为相同的 AsyncGenerator<StreamChunk> 输出：

// 统一的 chunk 类型
type StreamChunk = { type: 'text' | 'thinking' | 'tool_call' | 'usage' | 'done'; content: string };

// 每个 Provider 有独立的 parser
const parser = provider === 'anthropic'
  ? parseAnthropicStream(response)  // 处理 event: + data: 格式
  : parseOpenAIStream(response);    // 处理 data: 格式

// 消费方代码完全一样
for await (const chunk of parser) {
  if (chunk.type === 'text') updateContent(chunk.content);
  if (chunk.type === 'thinking') updateThinking(chunk.content);
}

这样前端组件完全不需要知道当前用的是哪个 Provider。新增 Provider 只需实现一个新的 parser。

Q10: ReadableStream.tee() 在 AI 应用中有什么用？​

答案：

tee() 将一个可读流分叉为两个独立的流，在 AI 应用后端代理层非常实用：

const [streamForClient, streamForMetrics] = llmResponse.body.tee();
// streamForClient → 直接转发给前端（不增加延迟）
// streamForMetrics → 后端异步收集 Token 用量、记录日志

这样可以不阻塞流式响应的转发，同时完成 Token 统计、日志记录、内容审计等后端操作。注意：两个流共享底层 buffer，内存开销比两次请求小得多。

Q11: 如何处理流式连接断开的情况？​

答案：

流式连接可能因网络波动、服务端超时等原因断开。应对策略：

  1. 总超时：设置 AbortController 的总超时（如 120 秒）
  2. 心跳检测：监控两个 chunk 之间的间隔，超过 30 秒视为断连
  3. 断点续传：保存已收到的内容，断连后带上已有内容重新请求
  4. 用户端体验：流中断时显示"连接断开"状态，提供"重试"按钮
  5. AbortController：提供"停止生成"功能，让用户可以手动中断

Q12: OpenAI 的 stream_options include_usage 有什么用？​

答案：

默认情况下，OpenAI 的流式响应不包含 Token 使用量信息（usage 字段只在非流式响应中返回）。设置 stream_options: { include_usage: true } 后，最后一个 chunk 会包含完整的 usage 信息：

{ "usage": { "prompt_tokens": 25, "completion_tokens": 42, "total_tokens": 67 } }

这对于后端实时统计 Token 消耗和成本计算非常重要。Anthropic 的流式响应天然在 message_start 和 message_delta 事件中包含 usage。

相关链接​