nodejs

Node.js LLM 集成实战：OpenAI 与 Anthropic

完整讲解 Node.js 接入大语言模型：OpenAI 与 Anthropic SDK 调用、流式输出 SSE、超时重试与指数退避、提示词工程与结构化输出、令牌与成本控制，以及把 LLM 调用异步任务化的工程实践。

Leeting Yan 6 分钟阅读 2656 字

LLM 集成已经成为后端标配能力：文本总结、结构化抽取、代码生成、客服助手……但 LLM 的调用形态与普通 REST 不同——流式、超时、令牌成本、重试策略都有讲究。本文以 OpenAI 与 Anthropic 为例，从 SDK 调用讲到 SSE 流式、退避重试、结构化输出与异步任务化。

1. LLM 集成全景

1.1 调用形态

同步对话（聊天/问答）       → 用户等待，直连
流式输出（打字机/长文）     → SSE 逐 token 推送
批量任务（摘要/抽取海量文） → 队列异步处理
工具调用（Function/Tool）   → 模型决策 + 代码执行

1.2 SDK 与原生 HTTP 的选型

方案优点缺点
官方 SDK类型齐全、流式封装好、省心多一层依赖
原生 fetch零依赖、可控流式/重试要自己写

推荐 SDK：接口演进快，SDK 能帮你跟上 breaking change。两个 SDK 都基于底层 fetch，与本文主题天然衔接。

1.3 环境变量与密钥管理

export OPENAI_API_KEY=sk-xxx
export ANTHROPIC_API_KEY=sk-ant-xxx

密钥绝不写进代码仓库，进程内用环境变量读取，云端用 Secret Manager 注入。

一句话：LLM 集成 = SDK 直连 + 流式 SSE + 退避重试 + 令牌核算四个模块；密钥走环境变量，绝不入库。

2. OpenAI SDK 调用

2.1 最小调用

import OpenAI from 'openai';

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const res = await client.chat.completions.create({
  model: 'gpt-4o-mini',
  messages: [
    { role: 'system', content: '你是一个简洁的中文助手。' },
    { role: 'user', content: '用一句话解释什么是索引。' },
  ],
  max_tokens: 200,
});

console.log(res.choices[0].message.content);

2.2 参数详解

参数作用建议
temperature随机性，越高越发散抽取用 0，创意用 0.7
max_tokens输出上限按任务估算，防跑飞
top_p核采样，与 temperature 二选一一般不用
presence/frequency_penalty重复惩罚摘要场景可调

2.3 多轮对话

const messages = [{ role: 'system', content: '你是技术支持。' },
  { role: 'user', content: '我的服务 500 了。' },
  { role: 'assistant', content: '先看错误日志和最近变更。' },
  { role: 'user', content: '日志里全是连接超时。' }];

一句话：OpenAI 调用 = chat.completions.create + messages 数组；抽取任务 temperature=0 保稳定，max_tokens 设上限防开销失控。

3. Anthropic SDK 调用

3.1 最小调用

import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });

const res = await client.messages.create({
  model: 'claude-sonnet-4-5',
  max_tokens: 500,
  system: '你是严谨的中文写作助手。',
  messages: [{ role: 'user', content: '帮我总结这段代码的职责。' }],
});

console.log(res.content[0].text);

3.2 与 OpenAI 的差异

维度OpenAIAnthropic
顶层 system放在 messages 里独立 system 字段
max_tokens可选默认大必填
返回结构choices[0].messagecontent[0].text
流式stream: truestream: true
工具toolstools（新版）

3.3 统一封装适配

多模型供应商并存时，封装一层适配器，业务侧只面对统一接口：

async function chat(provider, { system, user, maxTokens = 500 }) {
  if (provider === 'openai') {
    const res = await openai.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [{ role: 'system', content: system }, { role: 'user', content: user }],
      max_tokens: maxTokens,
    });
    return res.choices[0].message.content;
  }
  const res = await anthropic.messages.create({
    model: 'claude-sonnet-4-5', max_tokens: maxTokens,
    system, messages: [{ role: 'user', content: user }],
  });
  return res.content[0].text;
}

一句话：Anthropic 的关键差异 = system 独立 + max_tokens 必填；多供应商并存务必加一层适配器，业务代码不感知具体 SDK。

4. 流式输出 SSE

4.1 SDK 流式调用

const stream = await client.chat.completions.create({
  model: 'gpt-4o-mini',
  messages: [{ role: 'user', content: '写一首四行诗' }],
  stream: true,          // 开启流式
});
for await (const chunk of stream) {
  const delta = chunk.choices[0]?.delta?.content;
  if (delta) process.stdout.write(delta);
}

4.2 原生 fetch 消费 SSE

不想用 SDK 时，SSE 就是一段 HTTP 流，按行解析 data: 事件：

import { createInterface } from 'node:readline';

const res = await fetch('https://api.openai.com/v1/chat/completions', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${process.env.OPENAI_API_KEY}` },
  body: JSON.stringify({ model: 'gpt-4o-mini', stream: true, messages: [] }),
});
for await (const line of createInterface({ input: res.body })) {
  if (!line.startsWith('data:')) continue;
  const payload = line.slice(5).trim();
  if (payload === '[DONE]') break;
  const text = JSON.parse(payload).choices?.[0]?.delta?.content;
  if (text) process.stdout.write(text);
}

4.3 流式转 WebSocket 下发

浏览器端打字机效果，通常由服务端把 SSE 转成 WebSocket/SSE 再下发客户端；注意客户端断连要取消上游流，避免 token 浪费。

一句话：流式 = SDK stream:true 逐 chunk 或 fetch + readline 解析 data: 行；客户端断连时立刻 cancel 上游流，防 token 空烧。

5. 超时重试与指数退避

5.1 LLM 特有错误

429  Too Many Requests      限流，要退避
500/502/503  上游抖动       可重试
400/401/403  参数/鉴权错误  不可重试，直接报错

5.2 指数退避封装

async function withRetry(fn, { retries = 3, baseDelay = 1000 } = {}) {
  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      const status = err?.status;
      if ([400, 401, 403, 404].includes(status)) throw err; // 永久错误
      if (attempt === retries) throw err;
      const delay = baseDelay * 2 ** attempt + Math.random() * 500; // 指数+抖动
      await new Promise((r) => setTimeout(r, delay));
    }
  }
}

const text = await withRetry(() => callChat(question));

5.3 请求超时

const controller = new AbortController();   // SDK 支持 signal 透传
const timer = setTimeout(() => controller.abort(), 60 * 1000);
try {
  const res = await client.chat.completions.create({ ..., signal: controller.signal });
} finally {
  clearTimeout(timer);
}

一句话：LLM 重试 = 429/5xx 才重试，4xx 永久错误直接抛；baseDelay * 2 ** attempt 指数退避加抖动，避免集体重启风暴。

6. 提示词工程与结构化输出

6.1 提示词结构

角色（system）   → 你是谁，什么语气
任务（user）     → 具体做什么，输入在哪
约束（格式）     → 输出格式、长度、禁忌
示例（few-shot） → 给一两个期望样例

6.2 JSON 结构化输出

让模型输出可解析的 JSON，并声明约束：

const res = await client.chat.completions.create({
  model: 'gpt-4o-mini',
  response_format: { type: 'json_object' },   // 强制 JSON
  messages: [{ role: 'user', content: '从评论中抽取信息，输出 JSON，键为 sentiment 和 keywords。评论：服务很好但价格偏贵' }],
});
const data = JSON.parse(res.choices[0].message.content);

6.3 解析兜底

模型偶尔会输出带解释文字的 JSON，解析必须容错：

function safeParse(text) {
  try {
    return JSON.parse(text);
  } catch {
    const m = text.match(/\{[\s\S]*\}/);   // 掐头去尾找对象
    if (m) return JSON.parse(m[0]);
    throw new Error('无法解析模型输出');
  }
}

一句话：结构化输出 = 提示词定格式 + response_format: json_object 强制 + safeParse 容错三层；宁可解析失败显式报错，也不静默吞掉脏数据。

7. 令牌成本控制与异步任务化

7.1 令牌与成本

function estimateTokens(text) {          // 1 汉字约 1 到 1.5 token
  return Math.ceil(text.length / 1.3);
}

调用前先估算输入 token，配合模型单价做预算；输出设 max_tokens 上限。真实计数建议用各家的 tokenizer（如 tiktoken）。

7.2 并发限流

LLM 接口有 RPM/TPM 限流，批量任务必须限并发：

async function mapLimit(items, limit, fn) {
  const queue = [...items];
  const workers = Array.from({ length: limit }, async () => {
    while (queue.length) await fn(queue.shift());
  });
  await Promise.all(workers);
}

await mapLimit(posts, 5, async (post) => {
  await summarize(post);   // 最多 5 个并发调用
});

7.3 异步任务化

长文本批量摘要不该让 HTTP 请求等几分钟，而是入队异步处理：

客户端提交 → 写入队列 → Worker 消费调 LLM → 写回存储 → 客户端取结果

// Worker 侧示意：消费一条任务，调 LLM，写回
await queue.add('summarize', { articleId });
// worker 里 await callChat(...) 后写回数据库

一句话：成本控制 = 估算 token + 限并发 + max_tokens 上限；批量任务 = 入队异步处理，绝不阻塞在 HTTP 请求里（队列实操见 nodejs-message-queue）。

8. 踩坑清单

坑现象对策
密钥写进仓库泄露被滥用环境变量 + Secret Manager
4xx 反复重试白白烧钱4xx 永久错误直接抛
无退避重试 429触发更严限流指数退避 + 抖动
流式不取消断连后 token 空烧客户端断开立刻 cancel
解析脏 JSON静默拿到 undefinedsafeParse 容错 + 显式报错
无 max_tokens长文本费用爆炸设输出上限
并发打爆 RPM429 连环限流器 + 队列
忘传 system输出语气不可控角色/任务/约束/示例四要素
同步等长任务HTTP 超时入队异步化

9. 总结

环节要点
选型官方 SDK 为主，底层是 fetch
OpenAIchat.completions + messages，温度 0 保稳定
Anthropicsystem 独立 + max_tokens 必填
流式stream:true 逐 chunk / SSE 按行解析
重试429/5xx 退避，4xx 直接抛
提示词角色 + 任务 + 约束 + 示例
结构化json_object + safeParse 兜底
成本估算 token + 限并发 + 上限
异步批量任务入队，Worker 消费

一句话记住：LLM 集成是「调用 + 流式 + 重试 + 成本」四位一体——SDK 封装调用形态，SSE 承载打字机体验，指数退避扛住限流，令牌核算守住预算；批量任务必须异步化，否则一个长摘要就把整个服务拖垮。

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「nodejs」更多文章

  1. Node.js 优雅停机与健康检查实战
  2. Node.js 内存泄漏诊断实战
  3. BullMQ 后台任务队列实战