第 53 章：一个 HTTP 请求之外

// → src/my-agent/api-client.ts
const response = await fetch("https://api.anthropic.com/v1/messages", {
  method: "POST",
  headers: {
    "x-api-key": process.env.ANTHROPIC_API_KEY!,
    "anthropic-version": "2023-06-01",
    "content-type": "application/json",
  },
  body: JSON.stringify({
    model: "claude-sonnet-4-6",
    max_tokens: 1024,
    messages: [{ role: "user", content: "你好，Claude。" }],
  }),
});

const data = await response.json();
console.log(data.content[0].text);

// → src/my-agent/types.ts
interface MessageResponse {
  id: string;           // "msg_01XFDUDYJgAACzvnptvVo4EL"
  type: "message";      // 固定值
  role: "assistant";    // 固定值
  model: string;        // 实际使用的模型
  content: ContentBlock[];  // 回复内容（核心）
  stop_reason: "end_turn" | "tool_use" | "max_tokens" | "stop_sequence";
  stop_sequence: string | null;
  usage: {
    input_tokens: number;
    output_tokens: number;
    cache_creation_input_tokens: number;
    cache_read_input_tokens: number;
  };
}

// → src/my-agent/api-client.ts
export interface ApiClientConfig {
  apiKey: string;
  baseUrl?: string;  // 默认 https://api.anthropic.com/v1
}

export class ApiClient {
  private baseUrl: string;
  private headers: Record<string, string>;

  constructor(config: ApiClientConfig) {
    this.baseUrl = config.baseUrl ?? "https://api.anthropic.com/v1";
    this.headers = {
      "x-api-key": config.apiKey,
      "anthropic-version": "2023-06-01",
      "content-type": "application/json",
    };
  }

  async createMessage(params: MessageCreateParams): Promise<MessageResponse> {
    const response = await fetch(`${this.baseUrl}/messages`, {
      method: "POST",
      headers: this.headers,
      body: JSON.stringify(params),
    });

    if (!response.ok) {
      const error = await response.json();
      throw new ApiError(response.status, error);
    }

    return response.json();
  }
}

// → src/my-agent/api-client.ts
export class ApiError extends Error {
  constructor(
    public status: number,
    public body: { error: { type: string; message: string } }
  ) {
    super(`Anthropic API error ${status}: ${body.error.message}`);
  }

  get isRetryable(): boolean {
    return this.status === 429 || this.status === 529 || this.status >= 500;
  }
}

// → ApiClient 配置中支持自定义 CA
export interface ApiClientConfig {
  apiKey: string
  baseUrl?: string
  ca?: string  // 自定义 CA 证书路径
}

// 在 fetch 时通过 undici agent 传入
import { Agent } from "undici"
import { readFileSync } from "fs"

function createDispatcher(config: ApiClientConfig) {
  if (config.ca) {
    return new Agent({
      connect: { ca: readFileSync(config.ca) }
    })
  }
  return undefined  // 使用默认行为
}

// → 最简代理支持（使用 undici ProxyAgent）
import { ProxyAgent } from "undici"

function getProxyDispatcher(): ProxyAgent | undefined {
  const proxyUrl = process.env.HTTPS_PROXY || process.env.HTTP_PROXY
  if (!proxyUrl) return undefined
  return new ProxyAgent({ uri: proxyUrl })
}

const response = await fetch(url, {
  method: "POST",
  headers,
  body,
  dispatcher: getProxyDispatcher(),  // 经过代理发送
})

function shouldBypassProxy(host: string): boolean {
  const noProxy = process.env.NO_PROXY?.split(",").map(s => s.trim()) ?? []
  return noProxy.some(pattern =>
    host === pattern || host.endsWith("." + pattern)
  )
}

// → 带指数退避的 fetch
async function fetchWithRetry(
  url: string,
  options: RequestInit,
  retries = 3,
): Promise<Response> {
  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      const response = await fetch(url, options)

      if ((response.status === 429 || response.status >= 500) && attempt < retries) {
        const delay = Math.min(1000 * Math.pow(2, attempt), 30000)
        await sleep(delay + Math.random() * 1000) // + jitter 防惊群
        continue
      }

      return response
    } catch (err) {
      if (attempt === retries) throw err
      const delay = Math.min(1000 * Math.pow(2, attempt), 30000)
      await sleep(delay + Math.random() * 1000)
    }
  }

  throw new Error("unreachable")
}

function sleep(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms))
}

// → 令牌桶速率限制器
class TokenBucket {
  private tokens: number
  private lastRefill: number

  constructor(
    private maxTokens: number,    // 桶容量
    private refillRate: number,   // 每秒补充的 token 数
  ) {
    this.tokens = maxTokens
    this.lastRefill = Date.now()
  }

  async acquire(tokens = 1): Promise<void> {
    this.refill()

    if (this.tokens >= tokens) {
      this.tokens -= tokens
      return
    }

    // 计算需要等待多少秒
    const deficit = tokens - this.tokens
    const waitMs = (deficit / this.refillRate) * 1000 + 50
    await sleep(waitMs)

    this.refill()
    this.tokens -= tokens
  }

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

// 使用：每分钟最多 50 次请求
const limiter = new TokenBucket(50, 50 / 60)
await limiter.acquire(1)

import { Agent, setGlobalDispatcher } from "undici"

setGlobalDispatcher(new Agent({
  connections: 16,               // 每个 origin 最多 16 个连接
  keepAliveTimeout: 30_000,      // 空闲连接保持 30 秒
}))

// → src/my-agent/api-client.ts (升级版)
export class ApiClient {
  private rateLimiter: TokenBucket
  private dispatcher: ProxyAgent | undefined

  constructor(config: ApiClientConfig) {
    this.baseUrl = config.baseUrl ?? "https://api.anthropic.com/v1"
    this.headers = { /* 同上 */ }
    this.rateLimiter = new TokenBucket(
      config.rpm ?? 50,
      (config.rpm ?? 50) / 60,
    )
    this.dispatcher = getProxyDispatcher()  // 仅在代理环境创建
  }

  async createMessage(params: MessageCreateParams): Promise<MessageResponse> {
    await this.rateLimiter.acquire(1)  // 控速

    const response = await fetchWithRetry(
      `${this.baseUrl}/messages`,
      {
        method: "POST",
        headers: this.headers,
        body: JSON.stringify(params),
        dispatcher: this.dispatcher,
      },
      3  // 最多重试 3 次
    )

    if (!response.ok) {
      const error = await response.json()
      throw new ApiError(response.status, error)
    }

    return response.json()
  }
}

// → src/my-agent/providers/ollama.ts
interface OllamaConfig {
  baseUrl: string  // "http://localhost:11434"
  model: string    // "qwen2.5-coder:latest"
}

class OllamaClient {
  constructor(private config: OllamaConfig) {}

  async createMessage(params: MessageCreateParams): Promise<MessageResponse> {
    // Ollama 的 API 格式与 Anthropic 不同，需要转换
    const ollamaBody = {
      model: this.config.model,
      messages: this.convertMessages(params.messages),
      stream: false,
      options: {
        temperature: params.temperature ?? 0.3,
        num_predict: params.max_tokens,
      },
    }

    const response = await fetch(`${this.config.baseUrl}/api/chat`, {
      method: "POST",
      body: JSON.stringify(ollamaBody),
    })

    const data = await response.json()
    return this.convertResponse(data)
  }

  private convertMessages(messages: MessageParam[]): Array<{ role: string; content: string }> {
    return messages.map(m => ({
      role: m.role,
      content: typeof m.content === "string" ? m.content : JSON.stringify(m.content),
    }))
  }

  private convertResponse(data: any): MessageResponse {
    return {
      id: data.id ?? `local-${Date.now()}`,
      type: "message",
      role: "assistant",
      model: this.config.model,
      content: [{ type: "text", text: data.message?.content ?? "" }],
      stop_reason: "end_turn",
      stop_sequence: null,
      usage: {
        input_tokens: data.prompt_eval_count ?? 0,
        output_tokens: data.eval_count ?? 0,
        cache_creation_input_tokens: 0,
        cache_read_input_tokens: 0,
      },
    }
  }
}

// → src/my-agent/provider-interface.ts
interface LLMProvider {
  createMessage(params: MessageCreateParams): Promise<MessageResponse>
  createMessageStream(params: MessageCreateParams): AsyncGenerator<SSEEvent>
  supportsFeature(feature: string): boolean  // thinking? caching? tools?
}

class ProviderRouter {
  private providers = new Map<string, LLMProvider>()

  register(name: string, provider: LLMProvider): void {
    this.providers.set(name, provider)
  }

  get(modelId: string): LLMProvider {
    // 按模型名路由
    if (modelId.startsWith("claude-")) return this.providers.get("anthropic")!
    if (modelId.startsWith("gpt-")) return this.providers.get("openai")!
    // 本地模型
    return this.providers.get("local")!
  }

  // 自动降级：API 不可用时切换到本地
  async getWithFallback(preferredModel: string): Promise<LLMProvider> {
    try {
      const provider = this.get(preferredModel)
      await provider.createMessage({
        model: preferredModel,
        max_tokens: 0,
        messages: [{ role: "user", content: "ping" }],
      })
      return provider  // 首选可用
    } catch {
      console.warn(`${preferredModel} 不可用，降级到本地模型`)
      return this.providers.get("local")!
    }
  }
}

await client.createMessage({
  model: "claude-sonnet-4-6",
  max_tokens: 1024,
  system: "你是一个只用文言文回答的助手。",
  messages: [{ role: "user", content: "今天天气如何？" }],
});