第 55 章：文字如溪流

// → src/my-agent/sse-parser.ts
export interface SSEEvent {
  event: string;
  data: string;
}

export function parseSSE(chunk: string): SSEEvent[] {
  const events: SSEEvent[] = [];
  const lines = chunk.split("\\n");

  let currentEvent = "";
  let currentData = "";

  for (const line of lines) {
    if (line.startsWith("event: ")) {
      currentEvent = line.slice(7);
    } else if (line.startsWith("data: ")) {
      currentData = line.slice(6);
    } else if (line === "" && currentData) {
      events.push({ event: currentEvent || "message", data: currentData });
      currentEvent = "";
      currentData = "";
    }
  }

  return events;
}

// → src/my-agent/stream-parser.ts
export type StreamEvent =
  | { type: "message_start"; message: MessageStartData }
  | { type: "content_block_start"; index: number; block: ContentBlockStart }
  | { type: "content_block_delta"; index: number; delta: Delta }
  | { type: "content_block_stop"; index: number }
  | { type: "message_delta"; delta: MessageDeltaData }
  | { type: "message_stop" }
  | { type: "error"; error: Error };

export async function* parseStream(
  response: Response
): AsyncGenerator<StreamEvent> {
  if (!response.ok) {
    throw new ApiError(response.status, await response.json());
  }

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const events = parseSSEBuffer(buffer);

      for (const sse of events) {
        const json = JSON.parse(sse.data);
        switch (sse.event) {
          case "message_start":
            yield { type: "message_start", message: json.message };
            break;
          case "content_block_start":
            yield {
              type: "content_block_start",
              index: json.index,
              block: json.content_block,
            };
            break;
          case "content_block_delta":
            yield {
              type: "content_block_delta",
              index: json.index,
              delta: json.delta,
            };
            break;
          case "content_block_stop":
            yield { type: "content_block_stop", index: json.index };
            break;
          case "message_delta":
            yield { type: "message_delta", delta: json.delta };
            break;
          case "message_stop":
            yield { type: "message_stop" };
            break;
        }
      }

      // 更新 buffer（去掉已消费的部分）
      buffer = getRemainingBuffer(buffer);
    }
  } finally {
    reader.releaseLock();
  }
}

// → src/my-agent/stream-example.ts
async function streamText(client: ApiClient, prompt: string) {
  const response = await client.createMessageStream({
    model: "claude-sonnet-4-6",
    max_tokens: 1024,
    messages: [{ role: "user", content: prompt }],
  });

  let textBuffer = "";
  for await (const event of parseStream(response)) {
    if (
      event.type === "content_block_delta" &&
      event.delta.type === "text_delta"
    ) {
      process.stdout.write(event.delta.text);  // 即时输出
      textBuffer += event.delta.text;
    }
  }
  return textBuffer;
}

// → src/my-agent/tool-call-parser.ts
export function accumulateToolCall(
  deltas: { type: "input_json_delta"; partial_json: string }[]
): object {
  const json = deltas.map(d => d.partial_json).join("");
  return JSON.parse(json);
}

// → AsyncGenerator 的自动 Backpressure
async function* fastProducer(): AsyncGenerator<string> {
  for (let i = 0; i < 100; i++) {
    yield `event-${i}`  // 每次 yield 后暂停，直到消费者 ready
  }
}

// 慢消费者
for await (const event of fastProducer()) {
  await sleep(100)  // 模拟慢处理
  console.log(event)
}
// 不会丢失任何事件。生产者自动等待。

// → 带手动暂停的流消费者
class PausableStream<T> {
  private paused = false
  private pending: T[] = []

  pause(): void {
    this.paused = true
  }

  resume(): T[] {
    this.paused = false
    const drained = [...this.pending]
    this.pending = []
    return drained
  }

  async *consume(source: AsyncGenerator<T>): AsyncGenerator<T> {
    for await (const item of source) {
      if (this.paused) {
        this.pending.push(item)  // 缓存，不传递给下游
        continue
      }

      // 先排空缓存
      while (this.pending.length > 0) {
        yield this.pending.shift()!
      }

      yield item
    }
  }
}

// → SSE 断线处理策略
async function* resilientStream(
  params: MessageCreateParams,
  client: ApiClient,
  retries = 2,
): AsyncGenerator<SSEEvent> {
  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      const stream = await client.createMessageStream(params)
      for await (const event of stream) {
        yield event
      }
      return  // 成功完成
    } catch (err) {
      if (attempt === retries) throw err

      if (err instanceof NetworkError) {
        const delay = Math.min(1000 * Math.pow(2, attempt), 10000)
        console.warn(`SSE 断线，${delay}ms 后重试 (${attempt + 1}/${retries})`)
        await sleep(delay)
        // 重新发送请求——流会从头开始
        continue
      }

      throw err  // 非网络错误，不重试
    }
  }
}

// → 流内错误容忍策略
function safeAccumulateAndParse(
  deltas: { partial_json: string }[],
): object | null {
  const json = deltas.map(d => d.partial_json).join("")

  try {
    return JSON.parse(json)
  } catch {
    // 策略 1：部分修复（简单场景）
    // 尝试加闭合括号
    const fixed = autoCloseBrackets(json)
    if (fixed) {
      try { return JSON.parse(fixed) } catch {}
    }

    // 策略 2：放弃本工具调用，返回错误
    console.error(`无法解析累积的 tool_use input: ${json.substring(0, 200)}`)
    return null
  }
}

function autoCloseBrackets(json: string): string | null {
  const opens = (json.match(/[{[]/g) || []).length
  const closes = (json.match(/[}\]]/g) || []).length
  if (opens === closes) return null  // 不是括号不匹配

  // 补齐缺失的闭合符号
  let fixed = json
  let depth = 0
  for (const ch of json) {
    if (ch === '{' || ch === '[') depth++
    if (ch === '}' || ch === ']') depth--
  }
  while (depth > 0) {
    fixed += (json.includes('{') && depth === 1) ? '}' : ']'
    depth--
  }
  return fixed
}