从零实现ChatGPT API封装：import chatgptapi from chatgpt 的底层原理与最佳实践

2次阅读

共计 3006 个字符，预计需要花费 8 分钟才能阅读完成。

直接调用 ChatGPT 原生 API 时，开发者常遇到以下问题：

认证冗余：每个请求都需要手动添加 Authorization 头，密钥硬编码在业务代码中
错误处理脆弱：网络波动导致请求失败时缺乏自动重试机制，错误类型未分类处理
流式响应复杂 ：处理 Server-Sent Events(SSE) 需要手动拼接数据块，上下文管理困难

使用 axios 创建带连接池的实例，并通过拦截器自动注入 API 密钥：

import axios, {AxiosInstance} from 'axios';

interface HttpClientConfig {
  apiKey: string;
  timeout?: number;
}

class ChatGPTHttpClient {
  private instance: AxiosInstance;

  constructor(config: HttpClientConfig) {
    this.instance = axios.create({
      baseURL: 'https://api.openai.com/v1',
      timeout: config.timeout || 30000,
      maxSockets: 10 // 连接池大小
    });

    // 请求拦截器
    this.instance.interceptors.request.use((cfg) => {cfg.headers.Authorization = `Bearer ${config.apiKey}`;
      return cfg;
    });
  }
}

实现带指数退避的重试策略和 SSE 解析器：

async function withRetry<T>(fn: () => Promise<T>,
  maxRetries = 3,
  baseDelay = 1000
): Promise<T> {
  let attempt = 0;
  while (true) {
    try {return await fn();
    } catch (error) {if (attempt >= maxRetries || !isRetryable(error)) throw error;
      const delay = baseDelay * 2 ** attempt + Math.random() * 500;
      await new Promise((r) => setTimeout(r, delay));
      attempt++;
    }
  }
}

function parseSSE(response: Response): AsyncGenerator<string> {const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  return {async *[Symbol.asyncIterator]() {while (true) {const { done, value} = await reader!.read();
        if (done) return;
        yield decoder.decode(value);
      }
    }
  };
}

使用链表结构维护多轮对话上下文：

interface Message {
  role: 'user' | 'assistant' | 'system';
  content: string;
  next?: Message;
}

class Conversation {
  private head?: Message;
  private tail?: Message;

  addMessage(role: Message['role'], content: string) {const newMsg: Message = { role, content};
    if (!this.head) this.head = newMsg;
    if (this.tail) this.tail.next = newMsg;
    this.tail = newMsg;
  }

  getContext(): Message[] {const ctx: Message[] = [];
    let current = this.head;
    while (current) {ctx.push(current);
      current = current.next;
    }
    return ctx;
  }
}

根据 QPS 调整 maxSockets 参数（建议 10-30）
启用 keep-alive 减少 TCP 握手开销
设置合理的 timeout（30-60 秒）

在响应拦截器中统计 usage 字段：

this.instance.interceptors.response.use((response) => {
  const usage = response.data?.usage;
  if (usage) {
    metrics.track('tokens', {
      prompt: usage.prompt_tokens,
      completion: usage.completion_tokens
    });
  }
  return response;
});

预初始化 HTTP 客户端
实现对话缓存预热
使用 Web Worker 预加载模型

方案	优点	缺点
环境变量	简单易用	需重启服务更新密钥
AWS Secrets Manager	自动轮换密钥	需要云服务依赖
HashiCorp Vault	完善的访问审计	部署复杂度高

使用令牌桶算法控制请求速率：

class RateLimiter {
  private tokens: number;
  private lastRefill: number;

  constructor(private capacity: number, private refillRate: number) {
    this.tokens = capacity;
    this.lastRefill = Date.now();}

  async acquire(): Promise<void> {this.refill();
    while (this.tokens < 1) {await new Promise(r => setTimeout(r, 100));
      this.refill();}
    this.tokens--;
  }

  private refill() {const now = Date.now();
    const elapsed = now - this.lastRefill;
    const newTokens = elapsed * this.refillRate / 1000;
    this.tokens = Math.min(this.capacity, this.tokens + newTokens);
    this.lastRefill = now;
  }
}