AI 学习笔记（五）：把 OpenAI 最小示例升级成可复用服务层，补上重试、日志、限流和错误分类

发表于 2026-03-12 更新于 2026-09-30 分类于 AI 阅读次数： Changyan：本文字数： 8.6k 阅读时长 ≈ 8 分钟

在前一篇结构化输出最小实践的基础上，封装一个可复用的 OpenAI 服务层，统一处理超时、重试、日志、限流和错误分类。

示例边界：本文给出单进程服务层设计。本轮验证了错误分类和空输出保护，未做真实 API 调用、配额测试或生产验收；SDK 的超时针对单次尝试，含重试的总时长可能更长。

上一篇介绍了一个最小结构化输出示例：请求发出去，模型按 JSON Schema 返回结果。
这一步很重要，但离真实项目还差一层东西: 可复用的服务层封装。

如果你直接把示例代码复制进业务，后面通常会很快遇到这些问题：

每个调用点都在重复初始化 client
超时、重试、日志、错误处理到处散落
一旦出现 429、网络抖动、服务端波动，排查会非常痛苦
调用量上来后，没有基本限流，稳定性会越来越差

所以这一篇不再讲“怎么调一次 API”，而是讲：怎么把最小示例收口成一个项目里能长期复用的 OpenAI 服务层。

1. 先把目标定清楚

我希望这层封装至少解决 5 件事：

统一入口：业务代码不要直接散落调用 SDK。
超时和重试：把基础稳定性配置收敛到一处。
日志可观测：至少能看到 request id、耗时、token 和剩余额度。
基础限流：别在本地批量调用时把自己打进 429。
错误分类：区分鉴权错误、限流错误、服务端错误、网络错误。

有了这层，后面无论你接结构化输出、普通文本生成，还是做多模型封装，都会轻很多。

2. 先装依赖

示例基于 Node.js 22 LTS 和支持 Responses API 的 OpenAI 官方 Node SDK，在独立示例目录安装并保留 lockfile：

1	npm i openai

先提醒一点：
官方 SDK 本身已经支持 timeout、maxRetries、APIError、request_id 和 withResponse()。
所以这篇不是重复造轮子，而是把这些能力收拢成一个稳定入口。

3. 一个可直接复用的最小服务层

将服务层放入 openai-service.mjs，业务调用放入 task-plan.mjs；配置 OPENAI_API_KEY 后执行 node task-plan.mjs。下面这份代码做了几件事：

用统一类封装所有 OpenAI 调用
用简单时间窗做进程内限流
统一日志字段
统一错误分类
保留 validate 钩子，方便后面接 zod / ajv

import OpenAI from 'openai';

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

class SimpleRateLimiter {
  constructor({ minIntervalMs = 800 } = {}) {
    this.minIntervalMs = minIntervalMs;
    this.nextAvailableAt = 0;
  }

  async wait() {
    const now = Date.now();
    const waitMs = Math.max(0, this.nextAvailableAt - now);

    this.nextAvailableAt = Math.max(now, this.nextAvailableAt) + this.minIntervalMs;

    if (waitMs > 0) {
      await sleep(waitMs);
    }
  }
}

function classifyOpenAIError(error) {
  if (error?.name === 'APIConnectionTimeoutError') {
    return { type: 'timeout', retryable: true, message: '请求超时' };
  }

  if (error?.name === 'APIConnectionError') {
    return { type: 'network_error', retryable: true, message: '网络连接异常' };
  }

  if (error?.type === 'invalid_output' || error?.type === 'validation_error') {
    return { type: error.type, retryable: false, message: error.message };
  }

  if (error instanceof SyntaxError) {
    return {
      type: 'parse_error',
      retryable: false,
      message: '模型输出不是有效 JSON',
    };
  }

  if (error instanceof OpenAI.APIError) {
    if (error.status === 401 || error.status === 403) {
      return {
        type: 'auth_error',
        retryable: false,
        message: 'API Key 无效或权限不足',
      };
    }

    if (error.status === 429) {
      const quotaExhausted = error.code === 'insufficient_quota';
      return {
        type: quotaExhausted ? 'quota_exhausted' : 'rate_limit',
        retryable: !quotaExhausted,
        message: quotaExhausted ? '额度不足，需检查账单与配额' : '请求过快',
      };
    }

    if (error.status >= 500) {
      return {
        type: 'server_error',
        retryable: true,
        message: 'OpenAI 服务端异常',
      };
    }

    return {
      type: 'api_error',
      retryable: false,
      message: error.message,
    };
  }

  return {
    type: 'unknown_error',
    retryable: false,
    message: error?.message || '未知错误',
  };
}

export class OpenAIService {
  constructor({
    apiKey = process.env.OPENAI_API_KEY,
    model = process.env.OPENAI_MODEL || 'gpt-5.2',
    timeout = 30_000,
    maxRetries = 2,
    minIntervalMs = 800,
  } = {}) {
    this.model = model;
    this.client = new OpenAI({
      apiKey,
      timeout,
      maxRetries,
    });
    this.rateLimiter = new SimpleRateLimiter({ minIntervalMs });
  }

  async requestJson({
    name,
    schema,
    instructions,
    input,
    validate,
  }) {
    await this.rateLimiter.wait();

    const startedAt = Date.now();

    try {
      const { data, response } = await this.client.responses
        .create({
          model: this.model,
          input: [
            { role: 'system', content: instructions },
            { role: 'user', content: input },
          ],
          text: {
            format: {
              type: 'json_schema',
              name,
              schema,
              strict: true,
            },
          },
        })
        .withResponse();

      const refusal = data.output?.some((item) =>
        item.type === 'message' && item.content?.some((part) => part.type === 'refusal'),
      );
      if (refusal || data.status !== 'completed' || !data.output_text) {
        throw Object.assign(new Error('响应被拒绝、未完成或无文本'), { type: 'invalid_output' });
      }
      const rawText = data.output_text;
      const parsed = JSON.parse(rawText);
      let finalData = parsed;
      if (validate) {
        try {
          finalData = validate(parsed);
        } catch (cause) {
          throw Object.assign(new Error('本地字段校验失败'), { type: 'validation_error', cause });
        }
      }

      const meta = {
        requestId: data._request_id || response.headers.get('x-request-id'),
        durationMs: Date.now() - startedAt,
        totalTokens: data.usage?.total_tokens ?? 0,
        remainingRequests: response.headers.get('x-ratelimit-remaining-requests'),
        remainingTokens: response.headers.get('x-ratelimit-remaining-tokens'),
      };

      console.info('[openai.success]', meta);

      return {
        data: finalData,
        meta,
      };
    } catch (error) {
      const detail = classifyOpenAIError(error);

      console.error('[openai.error]', {
        ...detail,
        requestId: error?.request_id,
        durationMs: Date.now() - startedAt,
      });

      throw Object.assign(new Error(detail.message), {
        type: detail.type,
        retryable: detail.retryable,
        cause: error,
      });
    }
  }
}

4. 业务里怎么调用

为了和上一篇保持连续，这里继续用“生成任务拆解清单”的例子：

import { OpenAIService } from './openai-service.mjs';

const taskPlanSchema = {
  type: 'object',
  additionalProperties: false,
  properties: {
    summary: { type: 'string' },
    tasks: {
      type: 'array',
      items: {
        type: 'object',
        additionalProperties: false,
        properties: {
          title: { type: 'string' },
          priority: { type: 'string', enum: ['high', 'medium', 'low'] },
        },
        required: ['title', 'priority'],
      },
    },
  },
  required: ['summary', 'tasks'],
};

function validateTaskPlan(data) {
  if (!data || typeof data.summary !== 'string' || !Array.isArray(data.tasks) ||
      data.tasks.some((task) => !task || typeof task.title !== 'string' ||
        !['high', 'medium', 'low'].includes(task.priority))) {
    throw new Error('任务清单字段校验失败');
  }
  return data;
}

const service = new OpenAIService({
  model: process.env.OPENAI_MODEL || 'gpt-5.2',
  timeout: 30_000,
  maxRetries: 2,
  minIntervalMs: 800,
});

const result = await service.requestJson({
  name: 'task_plan',
  schema: taskPlanSchema,
  instructions: '你是资深技术编辑，输出必须严格符合 JSON Schema。',
  input: `
我在维护一个 Hexo 博客，需要新增一篇 AI 学习笔记，
要求包括：选题、示例代码、发布检查清单。
`,
  validate: validateTaskPlan,
});

console.log(result.data);
console.log(result.meta);

返回值里通常最值得保留的是两部分：

data：给业务继续使用
meta：给日志、监控和排查使用

这就是“可复用服务层”和“只会调一次 API”的本质差别。

validate 是可选的业务校验钩子，但这里的任务清单业务依赖 summary/tasks/priority，所以实际调用必须接入它，不能把 {} 当作成功清单。这个最小校验只检查业务必需字段，完整 Schema 校验可换成既有 zod/ajv。纯文本业务不需要这组 JSON 字段校验，应使用单独的文本调用入口；本文的 requestJson 会解析 JSON，不直接处理任意文本。

5. 这层封装到底解决了什么

5.1 重试不再到处手写

官方 SDK 已经能对连接问题、超时、408、409、429 和 5xx 做基础重试。
所以大多数项目里，先把 maxRetries 收口到服务层就够用了。

如果你还要自己额外包一层指数退避，记得先想清楚：
不要把 SDK 内建重试和你自己的外层重试叠太狠，不然一次失败可能被放大成很多次请求。

5.2 日志终于能看懂

很多人接 API 时只在失败时 console.error(error)。
这对线上排查基本没用。

至少把这些字段打出来：

requestId
durationMs
totalTokens
remainingRequests
remainingTokens

这样你在定位“慢在哪里”“为什么频繁 429”“是不是某次请求异常贵”时，才有抓手。

5.3 错误处理从“全都 catch”变成“按类型处理”

把错误分层后，业务侧就能更明确地决定动作：

auth_error：直接报警，不要重试
quota_exhausted：检查账单和配额，延后重试不能解决额度不足
rate_limit：延后重试或降速
server_error / network_error / timeout：可以进入重试队列
parse_error：重点排查 prompt、schema 和本地校验
invalid_output / validation_error：明确返回失败，保留脱敏后的拒绝或校验原因供复核

连接和超时错误也是 APIError 的子类，因此必须在通用 APIError 分支之前识别。这些分类是在 SDK 内建重试结束后提供给业务的；retryable 不代表可以无限叠加外层重试。

这一步会直接影响你的自动化流程稳定性。

5.4 限流先做轻量版就够了

这里的 SimpleRateLimiter 只是一个进程内最小限流器，适合本地脚本、小型服务或单实例任务。

它只限制请求启动间隔，不限制并发数、每分钟 token 总量，也不限制 SDK 内部重试。等待队列和会话数量还需要容量上限；不能把 minIntervalMs = 800 当成账号配额保障。

如果你后面是：

多实例部署
队列消费
高并发批处理

那就应该升级成更正式的方案，比如基于 Redis 的令牌桶、队列串行化，或者按租户做配额控制。

先有最小限流，再谈复杂限流，通常更符合工程节奏。

6. 落地时再补三件事

这篇先把“服务层稳定性”收好，真要上线，我还建议补三件事：

本地 schema 校验
不要只信模型返回，继续用 zod、ajv 或已有校验器兜一次。
敏感日志脱敏
日志里不要直接落完整用户输入、密钥、隐私字段。
统一业务降级
当模型调用失败时，要定义默认返回、重试上限和告警策略，不要把异常直接甩到前端页面。

总结

从“跑通一次 API”到“接进真实项目”，中间最关键的一步，就是补一层可复用服务层。

这一层的价值不在于代码多高级，而在于它帮你把下面这些问题提前收口：

配置散落
错误失控
日志不可追
调用无节制

一句话总结：
模型能力决定上限，服务层封装决定你能不能稳定落地。

下一篇我会继续沿着 Phase 2 往前走：
把单一 OpenAI 封装推进到多 provider 接入，开始对比 OpenAI、Claude、DeepSeek 的接入差异。

参考资料

本文永久链接: https://www.mulianju.com/learning-notes/ai-learning-notes-openai-service-wrapper/