Claude API のストリーミングで詰まる3つの落とし穴——実装してわかった本当の難しさ

// ❌ Edge Runtime で詰まりやすいパターン
import Anthropic from '@anthropic-ai/sdk';
 
export const runtime = 'edge'; // ← これが曲者
 
const client = new Anthropic();
 
export async function POST(req: Request) {
  const stream = await client.messages.stream({
    model: 'claude-opus-4-6',
    max_tokens: 1024,
    messages: [{ role: 'user', content: 'Hello' }],
  });
 
  // Edge Runtime では for await が期待通りに動かないことがある
  for await (const chunk of stream) {
    // ...
  }
}

// ✅ Node.js Runtime で動く安定パターン
export const runtime = 'nodejs'; // Edge→Node.js に変更
 
export async function POST(req: Request) {
  const client = new Anthropic();
 
  const stream = client.messages.stream({
    model: 'claude-opus-4-6',
    max_tokens: 2048,
    messages: [{ role: 'user', content: await req.text() }],
  });
 
  // SDK の toReadableStream() で Web 標準に変換
  return new Response(stream.toReadableStream(), {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

// ❌ エラーが漏れがちなパターン
async function streamMessage(content: string) {
  try {
    const stream = client.messages.stream({ ... });
 
    for await (const event of stream) {
      // ここの処理中にエラーが起きた場合...
      process.stdout.write(event.delta?.text ?? '');
    }
  } catch (e) {
    // 接続前のエラーは捕捉できるが、
    // ストリーム途中のネットワーク断は別の扱いが必要
    console.error(e);
  }
}

// ✅ 2種類のエラーを分けて処理するパターン
async function streamMessage(content: string): Promise<string> {
  const stream = client.messages.stream({
    model: 'claude-opus-4-6',
    max_tokens: 2048,
    messages: [{ role: 'user', content }],
  });
 
  // ストリーム中のエラーを個別にハンドリング
  stream.on('error', (error) => {
    console.error('Stream error:', error);
    // 必要に応じてリトライロジックを追加
  });
 
  // テキストだけ逐次処理
  stream.on('text', (text) => {
    process.stdout.write(text);
  });
 
  // 完了を待って最終メッセージを取得
  // ここで接続前エラーも含めて throw される
  try {
    const finalMessage = await stream.finalMessage();
    return finalMessage.content[0].type === 'text'
      ? finalMessage.content[0].text
      : '';
  } catch (e) {
    // API エラー・タイムアウト等
    throw new Error(`Stream failed: ${e instanceof Error ? e.message : String(e)}`);
  }
}

const finalMessage = await stream.finalMessage();
 
console.log({
  inputTokens: finalMessage.usage.input_tokens,
  outputTokens: finalMessage.usage.output_tokens,
  totalTokens: finalMessage.usage.input_tokens + finalMessage.usage.output_tokens,
  model: finalMessage.model,
});

// ストリーミングでコスト上限を設ける実践例
const MAX_INPUT_TOKENS = 8000;
 
function trimMessages(
  messages: Anthropic.MessageParam[],
  maxTokens: number
): Anthropic.MessageParam[] {
  // 概算：1メッセージ平均500トークンと仮定して直近N件を保持
  const estimatedPerMessage = 500;
  const maxMessages = Math.floor(maxTokens / estimatedPerMessage);
 
  if (messages.length <= maxMessages) return messages;
 
  // 最初のシステムコンテキスト（user メッセージ）は保持
  const systemMessage = messages[0];
  const recentMessages = messages.slice(-(maxMessages - 1));
  return [systemMessage, ...recentMessages];
}