ストリーミングで文字が流れている途中、ふつりと止まる。例外は飛ばず、ユーザーの画面には書きかけの文章だけが残ります。この静かな失敗が、いちばん厄介です。
原因はネットワークだけではありません。厳しすぎるタイムアウト、チャンク処理の取りこぼし、トークン上限への到達、安全フィルターの発動。切り分け方も直し方もそれぞれ違うので、順に潰していきます。
ストリーミング応答が途切れる5つの原因
原因を特定するチェックリスト
- ネットワークタイムアウト — クライアント側またはサーバー側の接続不安定
- チャンク処理エラー — JSON パースに失敗した場合
- トークン上限超過 — リクエスト＋応答のトークン数が上限を超えた
- 安全フィルター発動 — Gemini が有害なコンテンツを検出して応答中止
- バックプレッシャー — クライアント側の処理が追いつかず、接続がタイムアウト
それぞれの原因と対処法を見ていきましょう。
原因1：ネットワークタイムアウト
タイムアウト発生時のエラーメッセージ
TimeoutError: The operation timed out
socket.timeout: _ssl.c:997: The handshake operation timed out
このエラーは、サーバーからの応答がクライアントに到着する前に、接続がタイムアウトしたことを示します。
タイムアウト時間の設定
Python の requests ライブラリでは、デフォルトのタイムアウトが非常に短く設定されています。ストリーミング応答用に明示的に長めのタイムアウトを設定する必要があります。
import requests
import json
from typing import Generator
def stream_gemini_with_timeout(prompt: str, timeout: int = 60) -> Generator[str, None, None]:
"""
ストリーミングレスポンスを取得し、テキストを逐次利用
Args:
prompt: ユーザープロンプト
timeout: タイムアウト秒数（デフォルト60秒）
"""
url = "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:streamGenerateContent"
headers = {
"x-goog-api-key": "YOUR_GEMINI_API_KEY",
"Content-Type": "application/json"
}
payload = {
"contents": [
{
"role": "user",
"parts": [{"text": prompt}]
}
]
}
try:
response = requests.post(
url,
json=payload,
headers=headers,
stream=True,
timeout=timeout # ストリーミング用に十分な時間を設定
)
response.raise_for_status()
for line in response.iter_lines():
if line:
try:
data = json.loads(line)
# テキストを抽出
if "candidates" in data and data["candidates"]:
candidate = data["candidates"][0]
if "content" in candidate:
for part in candidate["content"]["parts"]:
if "text" in part:
yield part["text"]
except json.JSONDecodeError:
print(f"Failed to parse line: {line}")
except requests.exceptions.Timeout:
print(f"Streaming timed out after {timeout} seconds")
raise
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
raise
# 使用例
for chunk in stream_gemini_with_timeout("Tell me about machine learning"):
print(chunk, end="", flush=True)Server-Sent Events（SSE）での実装
より堅牢なストリーミング実装には、Server-Sent Events がお勧めです。
from aiohttp import ClientSession
import asyncio
async def stream_gemini_sse(prompt: str, timeout: int = 300):
"""
aiohttp を使った非同期ストリーミング
timeout デフォルト = 5分（長い応答対応）
"""
async with ClientSession(timeout=timeout) as session:
url = "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:streamGenerateContent"
headers = {"x-goog-api-key": "YOUR_GEMINI_API_KEY"}
payload = {
"contents": [{"role": "user", "parts": [{"text": prompt}]}]
}
async with session.post(url, json=payload, headers=headers) as resp:
async for line in resp.content:
try:
data = json.loads(line)
if "candidates" in data:
for part in data["candidates"][0]["content"]["parts"]:
if "text" in part:
yield part["text"]
except (json.JSONDecodeError, KeyError):
continue
# 使用例
async def main():
async for chunk in stream_gemini_sse("Explain quantum computing"):
print(chunk, end="", flush=True)
asyncio.run(main())原因2：チャンク処理エラー
JSON パースエラーの検出と対処
ストリーミング応答は JSON 形式で返されますが、ネットワークの問題でデータが破損することがあります。
import json
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
def stream_with_error_handling(prompt: str) -> str:
"""
チャンクエラーを検出して、安全にスキップ
"""
accumulated_text = ""
error_count = 0
max_errors = 5 # 最大エラー許容数
for line in stream_gemini_with_timeout(prompt):
try:
# 各チャンクが有効な JSON か確認
if isinstance(line, str):
accumulated_text += line
else:
raise ValueError(f"Unexpected chunk type: {type(line)}")
except (json.JSONDecodeError, ValueError) as e:
error_count += 1
logger.warning(f"Chunk parse error #{error_count}: {e}")
if error_count > max_errors:
logger.error(f"Too many errors ({error_count}). Stopping stream.")
break
# エラーをスキップして続行
continue
return accumulated_text
# 使用例
result = stream_with_error_handling("What is artificial intelligence?")
print(result)UTF-8 デコードエラーの回避
バイナリチャンクが不完全な状態で到着する場合があります。
def safe_decode_chunk(chunk: bytes) -> str:
"""
不完全な UTF-8 シーケンスをスキップ
"""
try:
return chunk.decode('utf-8')
except UnicodeDecodeError as e:
# 不完全な UTF-8 シーケンスをスキップして再試行
return chunk[:e.start].decode('utf-8', errors='ignore')原因3：トークン上限超過による途中停止
token_count_metadata での確認
Gemini API の応答には、トークン使用量のメタデータが含まれています。これを監視することで、上限到達を事前に検知できます。
def stream_with_token_monitoring(prompt: str, max_tokens: int = 8192):
"""
トークン上限を監視しながらストリーミング
"""
total_tokens = 0
accumulated_text = ""
for chunk_data in stream_gemini_with_timeout(prompt):
# メタデータからトークン数を取得
if "usageMetadata" in chunk_data:
usage = chunk_data["usageMetadata"]
output_tokens = usage.get("outputTokens", 0)
total_tokens += output_tokens
print(f"Token usage: {total_tokens}/{max_tokens}")
if total_tokens >= max_tokens:
print("⚠️ Token limit approaching. Stopping stream.")
break
# テキストを追加
if "candidates" in chunk_data:
for part in chunk_data["candidates"][0]["content"]["parts"]:
if "text" in part:
accumulated_text += part["text"]
return accumulated_text, total_tokens
# 使用例
text, tokens = stream_with_token_monitoring("Write a long essay about AI", max_tokens=5000)
print(f"Generated {tokens} tokens")max_output_tokens の事前指定
応答の途中で打ち切られるのを防ぐには、事前に十分な max_output_tokens を設定します。
def create_stream_with_sufficient_tokens(prompt: str, estimated_output_tokens: int = 2000):
"""
十分なトークン枠を確保してストリーミング開始
"""
api_params = {
"model": "gemini-2.0-flash",
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
"generation_config": {
"max_output_tokens": estimated_output_tokens, # 余裕を持たせる
"temperature": 0.7
}
}
# API 呼び出し処理
# ...ストリーミング実装...
pass原因4：安全フィルター発動による中断
SAFETY_REASON エラーの検出
Gemini は有害なコンテンツを検出すると、応答を中止する場合があります。
def stream_with_safety_detection(prompt: str) -> dict:
"""
安全フィルター状態を監視
"""
result = {
"text": "",
"finish_reason": None,
"safety_ratings": []
}
for chunk in stream_gemini_with_timeout(prompt):
# finish_reason を確認
if "candidates" in chunk:
candidate = chunk["candidates"][0]
# 終了理由を確認
if "finishReason" in candidate:
result["finish_reason"] = candidate["finishReason"]
if candidate["finishReason"] == "SAFETY":
print("⚠️ Response blocked by safety filter")
break
# 安全評価を記録
if "safetyRatings" in candidate:
result["safety_ratings"] = candidate["safetyRatings"]
# テキスト追加
if "content" in chunk["candidates"][0]:
for part in chunk["candidates"][0]["content"]["parts"]:
if "text" in part:
result["text"] += part["text"]
return result
# 使用例
output = stream_with_safety_detection("Some potentially sensitive prompt")
if output["finish_reason"] == "SAFETY":
print("The request triggered the safety filter")
else:
print(output["text"])SafetySettings のカスタマイズ
安全設定を調整することで、フィルター強度を変更できます（※ユースケースに応じて）。
def stream_with_custom_safety_settings(prompt: str):
"""
安全フィルター設定をカスタマイズ
"""
safety_settings = [
{
"category": "HARM_CATEGORY_HARASSMENT",
"threshold": "BLOCK_MEDIUM_AND_ABOVE" # より多くのコンテンツを許可
},
{
"category": "HARM_CATEGORY_HATE_SPEECH",
"threshold": "BLOCK_ONLY_HIGH"
}
]
# API リクエストに safety_settings を含める
# ...実装...
pass原因5：バックプレッシャーと処理速度の不一致
バックプレッシャー制御の実装
サーバーからの応答速度がクライアント処理速度を上回ると、バッファオーバーフロー が発生します。
import asyncio
from asyncio import Queue
class BackpressureControlledStream:
def __init__(self, max_queue_size: int = 10):
self.queue = Queue(maxsize=max_queue_size)
self.max_queue_size = max_queue_size
async def producer(self, prompt: str):
"""ストリーミング応答を キューに入れる"""
async for chunk in stream_gemini_sse(prompt):
await self.queue.put(chunk)
await self.queue.put(None) # 終了シグナル
async def consumer(self):
"""キューから読み出して処理"""
accumulated = ""
while True:
# キューサイズを監視
queue_size = self.queue.qsize()
if queue_size > self.max_queue_size * 0.8:
print(f"⚠️ Backpressure: {queue_size}/{self.max_queue_size}")
chunk = await asyncio.wait_for(self.queue.get(), timeout=30)
if chunk is None:
break
accumulated += chunk
return accumulated
async def run(self, prompt: str) -> str:
"""プロデューサー＆コンシューマーを並行実行"""
producer_task = asyncio.create_task(self.producer(prompt))
consumer_task = asyncio.create_task(self.consumer())
result = await consumer_task
await producer_task
return result
# 使用例
async def main():
stream = BackpressureControlledStream(max_queue_size=20)
result = await stream.run("Explain neural networks")
print(result)
asyncio.run(main())再接続ロジックの実装
自動リトライ付き再接続戦略
ストリーミングが途切れた場合、自動的に再開する仕組みが重要です。
import time
from typing import Optional
class ResilientGeminiStream:
def __init__(self, max_retries: int = 3, backoff_factor: float = 2.0):
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def stream_with_retry(self, prompt: str) -> str:
"""
ストリーミングが途切れた場合、自動的に再接続
"""
accumulated_text = ""
attempt = 0
while attempt < self.max_retries:
try:
for chunk in stream_gemini_with_timeout(prompt):
accumulated_text += chunk
# 成功時
return accumulated_text
except (TimeoutError, ConnectionError) as e:
attempt += 1
if attempt >= self.max_retries:
raise RuntimeError(f"Failed after {self.max_retries} retries: {e}")
wait_time = self.backoff_factor ** attempt
print(f"Reconnecting in {wait_time}s (attempt {attempt}/{self.max_retries})")
time.sleep(wait_time)
return accumulated_text
# 使用例
client = ResilientGeminiStream(max_retries=3)
try:
result = client.stream_with_retry("Tell me about quantum physics")
print(result)
except RuntimeError as e:
print(f"Failed to get streaming response: {e}")チェックポイント ベースの再開
長いストリーミング応答では、定期的にチェックポイントを保存すると、途中から再開できます。
import json
def stream_with_checkpoints(prompt: str, checkpoint_file: str = "stream_checkpoint.json"):
"""
チェックポイントを保存しながらストリーミング
"""
# チェックポイントから再開
checkpoint = {}
try:
with open(checkpoint_file, 'r') as f:
checkpoint = json.load(f)
except FileNotFoundError:
pass
accumulated_text = checkpoint.get("text", "")
chunks_processed = checkpoint.get("chunks", 0)
try:
for i, chunk in enumerate(stream_gemini_with_timeout(prompt)):
if i < chunks_processed:
continue # スキップして再開位置から処理
accumulated_text += chunk
chunks_processed += 1
# 100チャンクごとにチェックポイント保存
if chunks_processed % 100 == 0:
with open(checkpoint_file, 'w') as f:
json.dump({
"text": accumulated_text,
"chunks": chunks_processed,
"timestamp": time.time()
}, f)
finally:
# 完了後、チェックポイントを削除
import os
if os.path.exists(checkpoint_file):
os.remove(checkpoint_file)
return accumulated_textまとめ
Gemini API のストリーミング応答が途切れるのは、単なる一時的な問題ではなく、複数の原因から生じています。
- ネットワークタイムアウト: 十分なタイムアウト時間を設定
- チャンク処理エラー: JSON パースエラーハンドリングを実装
- トークン上限: max_output_tokens を事前指定
- 安全フィルター: finish_reason で SAFETY を検出
- バックプレッシャー: キューイングで処理速度を制御
さらに詳しく学びたい方は、Gemini リアルタイムストリーミング高度パターンで、SSE・WebSocket 連携の実装パターンを解説しています。
また、エラーハンドリング＆リトライ完全ガイドでは、ストリーミング以外のエラーハンドリング全般を学べます。
本番環境でのストリーミング構築について詳しく知りたい方は、以下をお勧めします。