LLMからJSONをストリーミング受信すると、正常な応答なのにjson.loads()が失敗することがあります。
原因は、通信で届くチャンクとJSONの構文上の区切りが一致しないためです。
さらに、バイト単位で受信する場合はUTF-8の文字自体が途中で分割される可能性もあります。
受信中はバイト列を復元し、JSONの解析とスキーマ検証は完了後に行うのが基本です。
正常なJSONでもチャンク単位では解析できない
たとえば、LLMに次の構造化データを返してもらう場面を考えます。
{"status":"ok","message":"処理完了"}
完成した文字列は正しいJSONです。
ところが、ストリーミングでは次のように分割される可能性があります。
{"status":
"ok","mess
age":"処理
完了"}
この分割位置にJSONとしての意味はありません。
通信側はJSONのキーや文字列リテラルを理解してチャンクを切っているわけではないためです。
受信するたびにjson.loads()を呼ぶと、最初のチャンクから失敗します。
import json
chunks = [
'{"status":',
'"ok","mess',
'age":"処理',
'完了"}',
]
for chunk in chunks:
try:
print(json.loads(chunk))
except json.JSONDecodeError:
print("parse error")
実行結果は次のとおりです。
parse error
parse error
parse error
parse error
すべてのチャンクが失敗します。
しかし、結合すると問題なく解析できます。
result = json.loads("".join(chunks))
print(result["message"])
処理完了
ここで間違えやすいのは、パースエラーをLLMの出力異常と判断してしまうことです。
json.loads()は、渡された文字列を一つのJSON文書として解析します。後続のチャンクが届くことを前提に、解析状態を保持する仕組みではありません。
したがって、未完成のJSONを渡して失敗するのは正常な動作です。
問題はパーサーではなく、まだ完成していないデータを完成済みとして扱う設計にあります。
UTF-8の境界とJSONの境界は別物
文字列の結合だけで済むのは、受信したチャンクがすでに正しくデコードされたstrである場合です。
HTTPレスポンスなどからbytesを直接扱う場合は、もう一つ問題があります。
UTF-8では、日本語の文字は一般に複数バイトで表現されます。
たとえば「完」はUTF-8で3バイトです。
character = "完"
data = character.encode("utf-8")
print(data.hex())
print(data[:1].decode("utf-8"))
最初のprint()は次の値を表示します。
e5ae8c
続くdecode()はUnicodeDecodeErrorになります。
最初の1バイトだけでは、文字として復元できないためです。
通信のチャンク境界が、この3バイトの途中に入ることもあります。
この場合、チャンクごとにdecode("utf-8")を実行する方法では正しく復元できません。
ここで区別したい境界は二つあります。
- バイト列からUnicode文字列を復元できる境界
- JSON文書として構文解析できる境界
前者は文字コードの問題で、後者はJSONの構文の問題です。
Python標準ライブラリのcodecsには、前者を処理する増分デコーダーがあります。
codecs.getincrementaldecoder("utf-8")で取得したデコーダーは、途中まで受信したバイト列の状態を保持できます。
文字を完成させるだけのバイトが届くと、その文字を返します。
ただし、UTF-8のデコードに成功したからといって、JSON文書が完成したとは限りません。
{"status":"ok"という文字列はUTF-8として正常ですが、JSONとしては未完成です。
文字コードの復元とJSONの確定は、別々の処理として設計する必要があります。
非同期ストリームを完了後に検証する
外部APIを使わずに、この問題を再現します。
JSON全体をUTF-8に変換し、任意のバイト位置で分割する非同期ジェネレーターを用意します。
修正版では、次の順序で処理します。
-
bytesを非同期で受信する - 増分UTF-8デコーダーで文字列へ復元する
- 復元した文字列をバッファに追加する
- ストリーム完了後に
json.loads()を実行する - 必須項目の型を検証して確定する
以下はPython 3.14、標準ライブラリのみで動作する例です。追加の依存パッケージはありません。
import asyncio
import codecs
import json
async def byte_stream(data: bytes, cuts: list[int]):
start = 0
for end in [*cuts, len(data)]:
await asyncio.sleep(0)
yield data[start:end]
start = end
def validate(value: object) -> dict:
if not isinstance(value, dict):
raise ValueError("JSON object required")
if value.get("status") not in ("ok", "error"):
raise ValueError("invalid status")
if not isinstance(value.get("message"), str):
raise ValueError("invalid message")
return value
async def receive(chunks):
decoder = codecs.getincrementaldecoder("utf-8")(
errors="strict"
)
parts = []
async for chunk in chunks:
text = decoder.decode(chunk, final=False)
parts.append(text)
print(f"received: {text!r}")
parts.append(decoder.decode(b"", final=True))
document = "".join(parts)
return validate(json.loads(document))
async def main():
source = {"status": "ok", "message": "処理完了"}
data = json.dumps(
source, ensure_ascii=False
).encode("utf-8")
marker = "完".encode("utf-8")
pos = data.index(marker)
cuts = [5, pos + 1, pos + 2]
result = await receive(byte_stream(data, cuts))
print("confirmed:", result)
asyncio.run(main())
cutsには、JSONの途中と「完」のUTF-8バイト列の途中を指定しています。
pos + 1とpos + 2で、3バイトの文字を意図的に分割しています。
このコードでは、途中のチャンクが文字として未完成でもUnicodeDecodeErrorにはなりません。
増分デコーダーが未完成のバイト列を内部に保持するためです。
最後のfinal=Trueには意味があります。
これは、これ以上バイト列が届かないことをデコーダーへ通知する操作です。
UTF-8の文字が途中で切れたままストリームが終了した場合、errors="strict"なら、この段階でUnicodeDecodeErrorが発生します。
一方、UTF-8として正しく復元できても、JSONの閉じ括弧が欠けていればjson.loads()で失敗します。
さらに、JSONとして正しくてもmessageが数値ならvalidate()で失敗します。
つまり、受信完了後にも三つの異なる失敗があり得ます。
- UTF-8が不正なら
UnicodeDecodeError - JSONの構文が不正なら
JSONDecodeError - アプリケーションが要求する形式と違えば
ValueError
これらを同じ「LLMのJSONエラー」として処理すると、原因の切り分けが難しくなります。
なお、今回のvalidate()は必須項目と型の一部だけを確認する簡易的なものです。未知のキーや文字列長などは制限していません。
実際のアプリケーションでは、利用するデータ契約に応じて検証条件を追加します。
途中表示と確定データを混ぜない
ストリーミングを使う理由の一つは、応答の完了を待たずに画面を更新できることです。
しかし、途中表示と確定データの利用は分けて考えます。
たとえば、LLMが次のJSONを返すとします。
{"status":"ok","message":"承認待ち"}
受信途中で"status":"ok"まで見えたとしても、そこで業務処理を開始するのは危険です。
後続のデータが欠ける可能性がありますし、最後まで受信してもスキーマ検証に失敗するかもしれません。
さらに、statusというキーが見えただけでは、JSON文書全体の整合性は保証されません。
ここでは、二つの状態を明確に分ける設計が扱いやすいと考えています。
receivingは、まだ受信中で確定していない状態です。
confirmedは、ストリームが正常に完了し、デコード、構文解析、スキーマ検証を通過した状態です。
受信中の文字列は、必要ならデバッグ画面や進捗表示に利用できます。
ただし、その文字列を正式なJSONオブジェクトとして扱ってはいけません。
特に、データベースの更新や外部APIの呼び出しに使う値は、confirmedになってから参照する方が安全です。
ここでのポイントは、単にjson.loads()の呼び出し回数を減らすことではありません。
受信中の不確定なデータが、確定済みデータを必要とする処理へ流れ込まないようにすることです。
バッファ方式にも限界がある
今回の実装は、JSON全体をメモリに保持します。
通常の小さな構造化応答なら単純で扱いやすい方法ですが、巨大なJSONを長時間受信する用途には向きません。
また、今回の例では、ストリームが正常終了したことを前提に確定処理を行っています。
実際の通信では、途中切断やタイムアウト、キャンセルも考慮する必要があります。
async forが正常終了した場合と、通信エラーで中断された場合を区別し、中断時には確定処理を実行しない設計にします。
さらに、入力サイズの上限も必要です。
json.loads()を最後に一度だけ呼ぶ設計でも、無制限にデータを蓄積すればメモリ使用量が増え続けます。
本番では受信バイト数を監視し、許容サイズを超えた時点で処理を中断する方が安全です。
もう一つ、今回の実装はストリーム全体が単一のJSON文書であることを前提にしています。
複数のJSON文書が連続するJSON Lines形式や、API独自のイベント形式には、そのまま適用できません。
たとえばServer-Sent Eventsでは、まずイベントのフレーミングを処理し、そのイベントが持つデータを解釈する必要があります。
APIが返すイベント単位と、その中に含まれるJSONの完成単位は、仕様を確認して区別します。
LLMの出力をリアルタイムにJSONとして利用したい場合は、別途、増分JSONパーサーやイベント単位の構造化出力を検討する余地があります。
ただし、途中で取り出せた値が最終的に有効なデータであることまで保証されるわけではありません。
最終的な確定処理は残しておくべきです。
今回の実装は、単一のJSON文書を最後まで受信してから利用する場合に適しています。次に確認したいのは、実際に使うLLM APIが文字列断片、バイト列、イベントのどれを返すかという点です。その境界を確かめたうえで、受信中の表示と確定後の処理を接続する必要があります。