LLMストリーミングの文字化け、UTF-8の途中で切って再現

20バイトの「回答: 猫です。」を2つに切ったら、猫が「���」になった。LLMの応答を中継する処理で、生の受信バイトをその都度 decode() する実装は避けたい。文字の途中で届くケースを、テストに入れておく。

自分は今日、通信を使わずに分割位置だけ変えて確かめてた。「猫」1文字が3文字になるのは、地味に気になるんだよね。

対象はUTF-8の生バイトを読む処理。SDKから変換済みの文字列を受け取る場合は追加不要。Python 3.14.6、標準ライブラリだけで実行した。

UTF-8の途中で切ると何が起きるか

Pythonコードは上から同じファイルに保存して実行できる。

raw = "回答: 猫です。".encode("utf-8")
cut = raw.index("猫".encode("utf-8")) + 1
parts = [raw[:cut], raw[cut:]]

try:
    "".join(p.decode("utf-8") for p in parts)
except UnicodeDecodeError as e:
    print(f"{type(e).__name__}: {e}")
print("".join(p.decode("utf-8", errors="replace") for p in parts))

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe7 in position 8: unexpected end of data
回答: ���です。

「猫」は e7 8c ab の3バイト。先頭の e7 だけが前半に残り、後半は 8c ab から始まる。両方を独立して変換すると、元の1文字を組み立てられない。replace は例外を抑えてくれるけど、今回の結果では回答が壊れたまま通る。

受信したバイト列の区切りと、文字の区切りは一致するとは限らない。

同じデコーダーに続きを渡す

codecs.getincrementaldecoder() で1個作り、その応答を読み終えるまで使い続ける。

import codecs

def decode_stream(chunks):
    decoder = codecs.getincrementaldecoder("utf-8")(errors="strict")
    for chunk in chunks:
        yield decoder.decode(chunk)
    yield decoder.decode(b"", final=True)

print("".join(decode_stream(parts)))

回答: 猫です。

手元のCPythonの一次資料として、Lib/encodings/utf_8.py と Lib/codecs.py の実装も読んだ。UTF-8の IncrementalDecoder は BufferedIncrementalDecoder を継承する。後者は前回の残りを今回の入力につなぎ、変換で消費しなかったバイトをバッファーに戻している。

今回なら e7 を保留し、次に届いた 8c ab と合わせて「猫」を返す。チャンクごとにデコーダーを作り直すと、この保留分を引き継げない。別の応答との共有も避ける。

final=True で入力の終わりを伝える。未完成の文字が残っていれば、strict ではエラーになる。

分割位置を19通り変えて確かめる

20バイトの途中にある19か所すべてで2分割し、1バイトずつの受信も試す。末尾の欠けも別に確かめる。

for i in range(1, len(raw)):
    assert "".join(decode_stream([raw[:i], raw[i:]])) == raw.decode()
assert "".join(decode_stream(raw[i:i+1] for i in range(len(raw)))) == raw.decode()
print(f"2分割: {len(raw) - 1}通りOK / 1バイトずつ: OK")

try:
    list(decode_stream([raw[:-1]]))
except UnicodeDecodeError:
    print("末尾1バイト欠け: 検出")
else:
    raise AssertionError("不完全なUTF-8を見逃した")

2分割: 19通りOK / 1バイトずつ: OK
末尾1バイト欠け: 検出

これで確認したのは文字の復元まで。SSEを扱うなら、この後に行やイベントを組み立てる処理が要る。デコーダーが返した文字列にも、イベントの途中が含まれ得る。受け取るたびにJSONとして読むと、今度はそちらで詰まる。

文字の途中で切れた終了は検出できても、文字が完成した直後の切断はこれだけでは判定できない。応答が完了したかどうかは、利用するプロトコルの完了通知で確かめる。

自分ならこの19か所の分割をテストに残す。通信抜きで同じ失敗を再現できる。応答ごとのデコーダーと終了時の確認をセットで入れる。ストリーミングを扱う箇所が増えるほど、こういう小さなテストが効いてくると思う。