agent-loop/async-agent-lab/agent/llm.ts

// 大模型请求封装：OpenAI Chat Completions 协议，流式读取，把分片拼成一条完整的 assistant 消息。
// 只依赖 Node 自带的 fetch。
import type { AssistantMessage, Message, ToolCall, ToolSpec, Usage } from '../shared/types.ts';
import { setTimeout as delay } from 'node:timers/promises';

export type Provider = {
  displayName: string;
  api: 'openai-completions';
  baseURL: string;
  apiKey: string;
  model: string;
  contextWindow?: number;
  maxOutputTokens?: number;
  inputModalities?: readonly string[];
};

export type TransportEvent =
  | { type: 'http.request'; attempt: number; body: string }
  | { type: 'http.response'; attempt: number; status: number }
  | { type: 'http.error'; attempt: number; error: string };

export type ChatRequest = {
  messages: Message[];
  tools: ToolSpec[];
  signal?: AbortSignal;
  onDelta?: (delta: { content?: string; reasoning?: string }) => void;
  onTransport?: (event: TransportEvent) => void;
  toolChoice?: string;
};

export type ChatResult = {
  message: AssistantMessage;
  reasoning: string;
  usage: Usage | null;
  finishReason: string | null;
  ttftMs: number | null; // 首个分片到达的耗时
  attempt: number;
};

export type LLM = { provider: Provider; chat(request: ChatRequest): Promise<ChatResult> };

// 网关对单个令牌有 RPM 限制（实测 50 次/分钟），429 时退避重试。
const RETRY_DELAYS_MS = [1000, 3000, 8000, 15000, 30000];

export function createLLM(provider: Provider): LLM {
  async function open(request: ChatRequest) {
    for (let attempt = 1; ; attempt += 1) {
      const body = JSON.stringify({
        model: provider.model,
        messages: request.messages,
        tools: request.tools,
        temperature: 0,
        stream: true,
        stream_options: { include_usage: true },
        max_tokens: provider.maxOutputTokens,
        tool_choice: request.toolChoice === undefined || request.toolChoice === 'none'
          ? request.toolChoice
          : { type: 'function', function: { name: request.toolChoice } },
      });
      if (provider.apiKey && body.includes(provider.apiKey)) throw new Error('请求内容含有传输凭据。');
      let failure: string;
      try {
        request.onTransport?.({ type: 'http.request', attempt, body });
        const response = await fetch(`${provider.baseURL}/chat/completions`, {
          method: 'POST',
          redirect: 'error',
          signal: request.signal,
          headers: {
            'Content-Type': 'application/json',
            Authorization: `Bearer ${provider.apiKey}`,
          },
          body,
        });
        request.onTransport?.({ type: 'http.response', attempt, status: response.status });
        if (response.ok) return { response, attempt };
        await response.body?.cancel();
        failure = `模型网关返回 HTTP ${response.status}`;
        const retryable = response.status === 429 || response.status >= 500;
        if (!retryable) throw new Error(failure);
      } catch (error) {
        if (request.signal?.aborted) throw error;
        if (error instanceof Error && error.message.startsWith('模型网关返回')) throw error;
        failure = `连接模型网关失败：${error instanceof Error ? error.message : String(error)}`;
        request.onTransport?.({ type: 'http.error', attempt, error: failure });
      }
      const waitMs = RETRY_DELAYS_MS[attempt - 1];
      if (waitMs === undefined) throw new Error(failure);
      await delay(waitMs, undefined, { signal: request.signal });
    }
  }

  return {
    provider,
    async chat(request) {
      const startedAt = performance.now();
      const { response, attempt } = await open(request);

      let content = '';
      let reasoning = '';
      let usage: Usage | null = null;
      let finishReason: string | null = null;
      let ttftMs: number | null = null;
      const toolCalls: ToolCall[] = [];

      for await (const chunk of readServerSentEvents(response)) {
        if (chunk.error) throw new Error('模型网关在流中返回错误');
        if (ttftMs === null) ttftMs = Math.round(performance.now() - startedAt);
        if (chunk.usage) usage = normalizeUsage(chunk.usage);
        const choice = chunk.choices?.[0];
        if (!choice) continue;
        if (choice.finish_reason) finishReason = choice.finish_reason;
        const delta = choice.delta ?? {};
        if (delta.reasoning_content) {
          reasoning += delta.reasoning_content;
          request.onDelta?.({ reasoning: delta.reasoning_content });
        }
        if (delta.content) {
          content += delta.content;
          request.onDelta?.({ content: delta.content });
        }
        // 工具调用按 index 分片到达：第一片带 id 和函数名，后续分片只追加参数字符串。
        for (const part of delta.tool_calls ?? []) {
          const call = (toolCalls[part.index] ??= {
            id: '',
            type: 'function',
            function: { name: '', arguments: '' },
          });
          if (part.id) call.id = part.id;
          if (part.function?.name) call.function.name += part.function.name;
          if (part.function?.arguments) call.function.arguments += part.function.arguments;
        }
      }

      const message: AssistantMessage = { role: 'assistant', content };
      const calls = toolCalls.filter(Boolean);
      if (calls.length) message.tool_calls = calls;
      if (finishReason === null) throw new Error('模型响应流提前结束');
      return { message, reasoning, usage, finishReason, ttftMs, attempt };
    },
  };
}

async function* readServerSentEvents(response: Response): AsyncGenerator<any> {
  if (!response.body) throw new Error('模型网关没有返回响应体');
  const decoder = new TextDecoder();
  let buffer = '';
  for await (const bytes of response.body as unknown as AsyncIterable<Uint8Array>) {
    buffer += decoder.decode(bytes, { stream: true });
    let newline: number;
    while ((newline = buffer.indexOf('\n')) >= 0) {
      const line = buffer.slice(0, newline).trim();
      buffer = buffer.slice(newline + 1);
      if (!line.startsWith('data:')) continue;
      const data = line.slice(5).trim();
      if (data === '[DONE]') return;
      yield JSON.parse(data);
    }
  }
}

function normalizeUsage(raw: any): Usage {
  return {
    prompt_tokens: raw.prompt_tokens ?? 0,
    completion_tokens: raw.completion_tokens ?? 0,
    total_tokens: raw.total_tokens ?? 0,
    cached_tokens: raw.prompt_tokens_details?.cached_tokens,
    reasoning_tokens: raw.completion_tokens_details?.reasoning_tokens,
  };
}