はじめに
ナレッジ基盤やAIエージェントの外部知識獲得（RAG / データ収集）において、公的機関のAPI（e-Gov法令API、国会会議録API、e-Stat統計API）や学術・百科API（Wikidata SPARQL, PhilArchive, OpenAlex, ESV Bible, Wikipedia API）など、多種多様な外部Web APIから大量のデータを取得・同期するニーズが高まっています。
しかし、同期的・逐次的に HTTP リクエストを発行するとネットワーク I/O がボトルネックとなり、また過度な並列リクエストは API 側のレートリミット（Rate Limit）や IP 遮断（429 Too Many Requests / 403 Forbidden）を引き起こします。
本記事では、Python の httpx と asyncio を活用し、セマフォ制御・指数バックオフ再試行・多層ディスクキャッシュ を組み込んだ、高速かつ極めて安定したデータ収集パイプライン（obsidian-api-pipeline）のアーキテクチャを解説します。
1. データパイプラインの構成と課題
収集対象となる外部 API 群とそれぞれの特性は以下の通りです。
[ データ収集パイプライン (obsidian-api-pipeline) ]
│
├─► e-Gov 法令 API / 国会会議録 API (大量テキスト / レート制限厳しめ)
├─► Wikidata SPARQL / OpenAlex API (複雑なグラフクエリ / 構造化データ)
├─► PhilArchive / ESV / Bolls API (学術・原語リソース)
└─► e-Stat 統計 API (JSON / CSV 大容量レスポンス)
設計上クリアすべき条件
- I/O ブロッキングの解消: 非同期 I/O による複数 API への並列アクセス。
-
過負荷・レート制限の回避: 各 API ごとのセマフォ（
asyncio.Semaphore）による並行数制御。 - 一時的通信エラーへの耐性: 指数バックオフ（Exponential Backoff）による自動リトライ。
- 無駄な再リクエストの撲滅: ローカルディスクへの階層的キャッシュ（SQLite / Hash Cache）。
2. 非同期通信クライントの実装 (httpx + asyncio)
HTTP 1.1 / HTTP/2 をサポートし、完全な非同期 I/O を提供する httpx.AsyncClient をコアエンジンとして使用します。
共通 API クライアント基盤の実装例
import asyncio
import httpx
import logging
from typing import Optional, Dict, Any
logger = logging.getLogger(__name__)
class AsyncAPIPipelineClient:
def __init__(self, max_concurrent_requests: int = 5, timeout_seconds: float = 10.0):
# API サーバーへの同時リクエスト数を制御する Semaphore
self.semaphore = asyncio.Semaphore(max_concurrent_requests)
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(timeout_seconds),
follow_redirects=True,
headers={"User-Agent": "Obsidian-API-Pipeline/1.0 (Knowledge Integration)"}
)
async def fetch_json_with_retry(
self,
url: str,
params: Optional[Dict[str, Any]] = None,
max_retries: int = 3
) -> Optional[Dict[str, Any]]:
async with self.semaphore:
for attempt in range(1, max_retries + 1):
try:
response = await self.client.get(url, params=params)
# レート制限 (429) の場合は Retry-After に従って待機
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", attempt * 2))
logger.warning(f"[RATE LIMIT] 429 Detected. Waiting {retry_after}s...")
await asyncio.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except (httpx.HTTPError, httpx.TimeoutException) as exc:
wait_time = (2 ** attempt) # 指数バックオフ (2s, 4s, 8s)
logger.warning(
f"[HTTP ERROR] Attempt {attempt}/{max_retries} failed: {exc}. "
f"Retrying in {wait_time}s..."
)
if attempt == max_retries:
logger.error(f"[FATAL] Max retries reached for {url}")
return None
await asyncio.sleep(wait_time)
async def close(self):
await self.client.aclose()
3. ディスクキャッシュ層による無制限リクエスト防止
同一パラメータでの問い合わせを短時間で繰り返すのを防ぐため、リクエスト URL とパラメータのハッシュ値をキーとしたローカルディスクキャッシュ構造を導入しています。
import hashlib
import json
from pathlib import Path
CACHE_DIR = Path(r"C:\Projects\obsidian-api-pipeline\.cache")
CACHE_DIR.mkdir(exist_ok=True)
def get_cache_key(url: str, params: Optional[Dict[str, Any]]) -> str:
raw_str = f"{url}?{json.dumps(params or {}, sort_keys=True)}"
return hashlib.sha256(raw_str.encode("utf-8")).hexdigest()
async def fetch_with_cache(client: AsyncAPIPipelineClient, url: str, params: Optional[Dict[str, Any]]):
cache_key = get_cache_key(url, params)
cache_file = CACHE_DIR / f"{cache_key}.json"
# 1. キャッシュが存在すれば即座に返却
if cache_file.exists():
logger.info(f"[CACHE HIT] {url}")
return json.loads(cache_file.read_text(encoding="utf-8"))
# 2. キャッシュがなければ非同期取得して保存
data = await client.fetch_json_with_retry(url, params=params)
if data is not None:
cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
return data
4. バッチ並列処理と API 収集の実践例
例えば、複数の国会議録および哲学者データの情報を一括取得する場合、asyncio.gather を用いて並列にタスクを発行します。
async def main():
client = AsyncAPIPipelineClient(max_concurrent_requests=3)
queries = [
("https://kokkai.ndl.go.jp/api/speech", {"any": "憲法第13条", "recordPacking": "json"}),
("https://api.openalex.org/works", {"search": "Categorical Imperative", "per_page": 5}),
("https://e-gov.api.go.jp/v1/laws", {"lawName": "刑法"})
]
tasks = [fetch_with_cache(client, url, params) for url, params in queries]
results = await asyncio.gather(*tasks)
for idx, data in enumerate(results):
print(f"Query {idx+1} result loaded: {data is not None}")
await client.close()
if __name__ == "__main__":
asyncio.run(main())
5. まとめ
-
ネットワーク I/O の超高速化:
httpx+asyncioにより、複数の公的・学術 API からのデータ取得時間を従来の同期処理と比較して数分の一に短縮しました。 -
API サーバーへの負荷軽減と耐性向上:
asyncio.Semaphoreによる並行数制限と指数バックオフ再試行により、429 や 503 エラーによるパイプラインの停止を防ぎます。 - ローカルキャッシュによる決定論的開発: キャッシュ層の導入により、テスト実行や開発中の重複アクセスを完全にシャットアウトし、外部 API の利用枠を大幅に節約できます。
外部 API 連携を伴うデータ収集システムや AI エージェントのパイプライン開発において、本設計を基盤として活用してください。