マルチステップストリーミングテキスト処理の設計と実装方法

ストリーミングテキスト処理は、ユーザーに即時性のあるレスポンスを提供する上で重要な技術です。Vercel AI SDK を使用することで、複数のステップに分けて処理を行いながら、リアルタイムにテキストをクライアントへ送信することが可能です。本記事では、createDataStreamResponse を中心に、多段階のストリーミング処理を実装する方法について詳しく解説します。

ストリーミング処理の基本と利点

ストリーミングテキスト処理は、AIによるテキスト生成結果をチャンク単位でクライアントに送信する技術です。通常のレスポンス全体を一度に送る方法に比べて、ユーザーが待つ時間を感じにくくし、インタラクティブな体験を提供できます。

複数ステップでの処理の必要性

実際のアプリケーションでは、次のような処理の分離が必要なケースがあります：

1. ユーザーの意図を解析
2. 解析結果に基づいて処理を分岐
3. 最終的なレスポンスを生成

このような複数ステップに分けて処理を行うことで、各ステップで異なるモデルやパラメータ、ツールの呼び出しを柔軟に制御できます。

サーバーサイドでの実装

1. データストリームの初期化

まず、createDataStreamResponse を使用してストリームのベースを作成します：

return createDataStreamResponse({
  execute: async (dataStream) => {
    // 複数ステップの処理をここに実装
  }
});

2. 初期ステップ：ユーザー意図の抽出

最初のステップでは、ユーザーの意図を抽出するために、ツール呼び出しが必須となるように設定：

const firstStep = streamText({
  model: openai('gpt-4o-mini', { structuredOutputs: true }),
  system: 'ユーザーの目的を抽出してください。',
  messages,
  toolChoice: 'required',
  tools: {
    extractUserGoal: tool({
      parameters: z.object({ goal: z.string() }),
      execute: async ({ goal }) => goal
    })
  }
});

3. ストリームの制御

次のステップに進むために、ストリームの終了イベントを送らないように設定：

firstStep.mergeIntoDataStream(dataStream, {
  experimental_sendFinish: false
});

4. 次のステップ：処理結果を反映した応答生成

前ステップの結果を基に、より詳細な応答を生成：

const secondStep = streamText({
  model: openai('gpt-4o'),
  system: '補足情報を基に、ユーザーに役立つ回答を作成してください。',
  messages: [
    ...convertToCoreMessages(messages),
    ...(await firstStep.response).messages
  ]
});

5. ストリームの継続制御

今度は開始イベントを送らないようにして、ストリームにマージ：

secondStep.mergeIntoDataStream(dataStream, {
  experimental_sendStart: false
});

クライアントサイドでの処理

クライアントでは、標準的なストリーミング処理の方法で対応：

const { messages, input, handleInputChange, handleSubmit } = useChat();

{messages?.map(message => (
  <div key={message.id}>
    {message.parts.map((part, index) => {
      switch (part.type) {
        case 'text':
          return <p>{part.text}</p>;
        case 'tool-invocation':
          return <span>ツール呼び出し中...</span>;
      }
    })}
  </div>
))}

応用ケース

条件分岐処理

if (someCondition) {
  const branch = streamText({...});
  branch.mergeIntoDataStream(dataStream, {...});
} else {
  const alternative = streamText({...});
  alternative.mergeIntoDataStream(dataStream, {...});
}

ループ処理

while (needsProcessing) {
  const loop = streamText({...});
  loop.mergeIntoDataStream(dataStream, {...});
  needsProcessing = checkCondition();
}

複数モデルの協調処理

// 分析用モデルを使用
const analysis = streamText({
  model: openai('analysis-model'),
  // ...
});

// 生成用モデルを使用
const generation = streamText({
  model: openai('generation-model'),
  // ...
});

ベストプラクティス

• 各ステップは単一責任原則に従って設計
• 処理内容に応じて適切なモデルを選択
• エラーハンドリングを各ステップに組み込む
• 不要なステップは省略し、パフォーマンスを最適化
• ステップ間の状態を明確に管理