はじめに
LangChain と Vertex AI Search を組み合わせて RAG チャットボットを開発している際、サーバー側の一時的な不調（502 Bad Gateway）により、アプリケーションの検索処理が中断してしまう事象に遭遇しました。
「Google Cloud のライブラリなら標準でリトライしてくれるのでは？」という期待に反し、なぜリトライが行われないのか。その理由の調査と、tenacity を使った実装方法、さらにモックを使ったテスト手法までを解説します。
1. 発生した事象
チャットボットの検索処理中に、以下のエラー（Traceback）が発生し処理が中断されました。
google.api_core.exceptions.ServiceUnavailable: 502 Bad Gateway
通常、クラウドサービスにおける 502 や 503 エラーは一時的なものであり、数秒後に再試行すれば成功することがほとんどです。しかし、この SDK（およびそれをラップしている LangChain）は 1回失敗した瞬間に例外を投げ、リトライを一切試みていない ことが判明しました。
2. なぜリトライされないのか？（原因）
調査の結果、Google Cloud Python クライアントライブラリでは Vertex AI Search の検索処理において、デフォルトでリトライを行わない仕様であることがわかりました。
理由：HTTP POST メソッドの特性
Discovery Engine の search メソッドは、内部的に HTTP POST を使用しています。
一般的に、Web API のクライアントライブラリでは、副作用（データの作成や更新）を伴う可能性のある POST メソッドに対して、安全性の観点から自動リトライを慎重に扱う（あるいは行わない）傾向があります。
他言語（JavaやNode.js）の SDK ではリトライが実装されているケースもありますが、Python SDK においてデフォルト無効化されている明確な意図は公式ドキュメント等に記載がなく、原因は不明でした。しかし、POST メソッドの非冪等性（Non-idempotence）を懸念して安全側に倒している可能性が高いと推測されます。
LangChain 利用時の課題
Google Cloud SDK 自体にもリトライ設定を上書きする仕組み（retry 引数）は存在しますが、LangChain のような高レベルなラッパーライブラリを経由している場合、内部のクライアントインスタンスに適切な retry オブジェクトを注入するのが難しい（あるいはコードが複雑になる）ケースがあります。
そのため、SDK 内部の設定に頼るよりも、アプリケーション層で制御する方が確実です。
3. 解決策：tenacity によるアプリケーション側でのリトライ
ライブラリがリトライしない仕様である以上、アプリケーション側で明示的に制御する必要があります。Python では tenacity ライブラリを使用するのが標準的です。今回のように POST メソッドであっても検索処理自体は参照系であり副作用がないため、リトライを行っても安全と判断しました。
実装のポイント
- 指数バックオフ: 待機時間を
2s -> 4s -> 8sと伸ばし、サーバーへの負荷を抑える。 - 対象例外の限定: 502/503 (ServiceUnavailable) やタイムアウト (DeadlineExceeded) に限定する。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from google.api_core.exceptions import ServiceUnavailable, DeadlineExceeded
# リトライ対象のエラーを定義
RETRYABLE_EXCEPTIONS = (ServiceUnavailable, DeadlineExceeded)
@retry(
retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
reraise=True
)
async def execute_search_with_retry(retriever, queries):
return await retriever.abatch(queries)
4. 本当にリトライされるかテストする
本番環境でエラーが起きるのを待つわけにはいかないため、unittest.mock を使って「失敗・失敗・成功」のシナリオをシミュレートします。
import pytest
from unittest.mock import AsyncMock, MagicMock
from google.api_core.exceptions import ServiceUnavailable
@pytest.mark.asyncio
async def test_retry_logic():
mock_retriever = AsyncMock()
# 成功時のダミーレスポンス（ドキュメントのリストを内包したリスト）
# abatchの戻り値構造: [[Doc1, Doc2]]
mock_doc = MagicMock()
mock_doc.page_content = '{"id": "ok"}'
success_response = [[mock_doc]]
# 2回失敗(502)して3回目に成功する side_effect を設定
mock_retriever.abatch.side_effect = [
ServiceUnavailable("502 Bad Gateway"),
ServiceUnavailable("502 Bad Gateway"),
success_response
]
# リトライロジックを含む関数を呼び出す
results = await execute_search_with_retry(mock_retriever, ["test"])
# 検証
# 1. 合計3回呼ばれていること（リトライが機能したこと）
assert mock_retriever.abatch.call_count == 3
# 2. 最終的に成功レスポンスが取得できていること
assert results[0].page_content == '{"id": "ok"}'
まとめ
Google Cloud の Python SDK (GAPIC) において、なぜデフォルトでリトライが行われないのか、その正確な意図は不明ですが、以下の事実は確認できました。
- Discovery Engine (Vertex AI Search) の Python SDK はデフォルトでリトライしない。
- 他言語版でリトライされるからといって、Python 版も同じとは限らない。
- 一時的な 502 エラーでアプリを落とさないためには、自前でのリトライ実装が必須。
tenacity+ 指数バックオフ + 適切な例外キャッチ を組み合わせるのがベストプラクティス。
クラウドネイティブな開発において、「SDK がよしなにやってくれるだろう」という思い込みを捨て、実行時の設定値をコードやテストで検証することの重要性を再認識した事例でした。
簡単30秒