DeepSeek Harness StreamChunk プロトコルとエラー処理

アダプターの stream() は実際にどのようなデータを出力するのか？

答えは StreamChunk、厳密な順序を持つチャンクプロトコルです。

この記事では、チャンクシーケンス、重要なルール、そして転送エラー時の対処方法について説明します。

StreamChunk プロトコル

StreamChunk は Harness とアダプターの間のストリーミングプロトコルです。

1つのコンテンツブロックは、最初に block-start で開始し、途中で delta で増分転送し、最後に block-end で終了します。

テキストとツール呼び出しは2種類の異なるコンテンツブロックであり、それぞれ start / delta / end を経ます。

すべてのチャンクの最後に、まず usage でトークン使用量を報告し、その後 finish で終了理由を宣言します。

StreamChunk 分片序列图

上の図は上から下へ、1回の完全なチャンクシーケンスを示しています。

最初にテキストブロック：block-start → text-delta × 2 → block-end。

次にツール呼び出しブロック：block-start → tool-call-delta → block-end。

最後に usage と finish。

finish の reason が stop の場合は正常終了を意味し、tool-calls の場合はツール実行の要求を意味します。

完全なチャンクシーケンス

公式ドキュメントには exampleChunks があり、1回の生成で得られるすべての chunk を順番に出力します。

text-delta は複数のチャンクに分割でき、増分を連結して完全なテキストにします。

tool-call-delta の argumentsDelta は、元の JSON テキストの増分です。

例

// ファイルパス：サンプルコード。1回の完全な chunk シーケンスを示す
import { CallId, type StreamChunk } from '@deepseek-ai/dsh-llm'

async function* exampleChunks(): AsyncIterable<StreamChunk> {
  // 1. Start each content block with block-start.
  // テキストブロックを開始、index は 0
  yield { type: 'block-start', index: 0, blockType: 'text' }

  // 2. Stream text through text-delta.
  // テキスト増分。複数のチャンクに分割可能
  yield { type: 'text-delta', index: 0, text: 'サンプル' }
  yield { type: 'text-delta', index: 0, text: ' 教程' }

  // 3. End each content block with block-end and the complete block.
  // 完全なブロックで終了。index は block-start と一致
  yield {
    type: 'block-end',
    index: 0,
    block: { type: 'text', text: 'サンプル 教程' },
  }

  // 4. Tool-call block.
  // ツール呼び出しブロックを開始、index は 1
  yield { type: 'block-start', index: 1, blockType: 'tool-call' }
  // ツール名とパラメータ増分。id は CallId ファクトリで生成
  yield {
    type: 'tool-call-delta',
    index: 1,
    id: CallId('call-123'),
    name: 'bash',
    argumentsDelta: '{"command":"echo サンプル"}',
  }
  // 完全なブロックで終了。arguments は結合済みの JSON テキスト
  yield {
    type: 'block-end',
    index: 1,
    block: {
      type: 'tool-call',
      id: CallId('call-123'),
      name: 'bash',
      arguments: '{"command":"echo サンプル"}',
    },
  }

  // 5. Token usage.
  // トークン使用量を報告。finish の前である必要がある
  yield { type: 'usage', usage: { inputTokens: 100, outputTokens: 50 } }

  // 6. Finish reason.
  // 最後のチャンク。終了理由を宣言
  yield { type: 'finish', reason: { kind: 'stop' } }
  // Alternatively, { kind: 'tool-calls' } requests tool execution.
}

CallId はプロトコルに組み込まれたファクトリ関数で、ツール呼び出し id を生成するために使用します。

argumentsDelta は1つのチャンクで完全に生成することも、複数のチャンクに分けて増分生成することもできます。

重要なルール

プロトコルには5つの厳格なルールがあります。いずれかに違反すると、消費側の解析エラーが発生します。

ルール説明
block-start と block-end はペア各 block-start には対応する block-end が必要
index は 0 から始まり増加するコンテンツブロックの順序を識別するために使用
argumentsDelta は元の JSON 増分1つのチャンクで完全に生成することも、複数のチャンクに分けて生成することもできる
finish は最後のチャンクでなければならないその後はチャンクを一切送信できない
usage は finish の前でなければならない最初にトークン使用量を報告し、その後終了を宣言する

text-delta と tool-call-delta は両方とも所属ブロックの index を付ける必要があります。コンテンツブロック間で交差してはいけません。

エラー処理：LlmError で失敗を表現する

アダプターは、安定した code を持つ LlmError を使用して、転送およびプロトコル障害をスローする必要があります。

agent-loop はこのエラーとその code を保持し、診断およびポリシー処理に使用します。

通常の Error が自動的に変換されることに依存してはいけません。

安定した code により、上位層はエラー文字列を解析するのではなく、エラータイプを正確にマッチングできます。

各プロバイダーの HTTP リクエストには attributionHeaders() もマージする必要があります。

また、options.signal を fetch に渡し、キャンセルとリソース解放のプロセスを確実に停止させます。

例

// ファイルパス：サンプルコード。エラー処理を備えた HttpAdapter のスケルトン
import {
  attributionHeaders,
  LlmAdapter,
  LlmError,
  type GenerateOptions,
  type StreamChunk,
} from '@deepseek-ai/dsh-llm'

class HttpAdapter extends LlmAdapter {
  // コンストラクタでエンドポイントのアドレスを注入
  constructor(private readonly endpoint: string) {
    super()
  }

  async *stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
    // HTTP リクエストを開始し、帰属ヘッダーをマージし、中止シグナルを渡す
    const response = await fetch(this.endpoint, {
      method: 'POST',
      headers: {
        'content-type': 'application/json',
        ...attributionHeaders(),
      },
      body: JSON.stringify({ model: options.model, messages: options.messages }),
      // 呼び出し側がキャンセルを要求すると、fetch はすぐに中止する
      ...options.signal ? { signal: options.signal } : {},
    })
    if (!response.ok) {
      // 安定した code を持つ LlmError で転送失敗を表現
      throw new LlmError(`Provider API error: ${response.status}`, 'PROVIDER_HTTP_ERROR')
    }
    // A real adapter parses the response and emits the complete chunk sequence.
    // 実際のアダプターはここでレスポンスボディを解析し、完全なチャンクシーケンスを生成する
    yield { type: 'finish', reason: { kind: 'stop' } }
  }
}

attributionHeaders() は帰属情報をリクエストヘッダーにマージします。

options.signal が存在する場合は fetch の signal として渡され、リクエストキャンセル時に即座に停止します。

response.ok が false の場合は LlmError をスローし、code は PROVIDER_HTTP_ERROR です。

スケルトンは finish を1つだけ yield し、実際のアダプターはここでレスポンスボディを解析します。

LlmError の最初のパラメータはメッセージ、2番目のパラメータは安定した code です。

上位層は code に基づいてポリシー判断を行うため、code は一度公開したら変更してはいけません。

黙って破棄できないフィールド

GenerateOptions にアダプターがサポートできないフィールドがある場合も、同様に LlmError をスローし、黙って破棄しないでください。

推論メタデータには、順序付けられた不透明 ID、表示名、およびオプションの構成デフォルト値が含まれます。

アダプターが提供する権威あるオプションリスト（上流のケイパビリティ API が返す off を含む）を保持してください。

オプションの推論強度をコア列挙型に昇格させないでください。そうしないとアダプターは上流の柔軟性を失います。

サービスは集約結果を検証し、stream() を呼び出す前に、明示的に指定されたがサポートされていない推論強度を拒否します。

reasoning を省略すると、そのモデルにオプションの推論強度機能がないことを示します。

まとめと自己テスト

一言でまとめると：StreamChunk プロトコルはペアになったブロックで内容を記述し、usage と finish で締めくくります。失敗時は安定した code を持つ LlmError で表現します。

自己テスト1：テキストブロックには最低限どのチャンクが必要でしょうか？

自己テスト2：usage と finish の順序はどちらが先ですか？

自己テスト3：リクエスト失敗時、なぜ通常の Error ではなく code 付きの LlmError を使用するのですか？

その他の拡張