AI APIリクエスト・レスポンスアーカイブシステムの構築方法

AI APIを本番環境に組み込む際、最も重要なのがリクエストとレスポンスの記録（アーカイブ）です。料金の確認、障害時の調査、コンプライアンス対応、いずれにも不可或缺的の存在となります。

なぜアーカイブシステムが必要か

• API利用料の正確な算出とコスト最適化
• 障害発生時の迅速な原因特定
• LLM出力の品質検証と改善
• コンプライアンス・監査対応

私は以前、レート制限超えによる突然のシステム停止に苦しめられた経験があります。あの時、リクエストログがあれば、事前に対策を打てたはずです。

基本的な実装

SDKを使ったシンプルなアプローチ

import openai
import json
from datetime import datetime
import sqlite3

HolySheep AI のエンドポイントを設定
client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

class RequestArchiver:
    def __init__(self, db_path="archive.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_table()
    
    def _init_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS api_requests (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                model TEXT,
                prompt_tokens INTEGER,
                completion_tokens INTEGER,
                total_cost_usd REAL,
                latency_ms INTEGER,
                request_text TEXT,
                response_text TEXT,
                status TEXT,
                error_message TEXT
            )
        """)
        self.conn.commit()
    
    def archive(self, model, request_text, response, latency_ms, error=None):
        prompt_tokens = response.usage.prompt_tokens if hasattr(response, 'usage') else 0
        completion_tokens = response.usage.completion_tokens if hasattr(response, 'usage') else 0
        # HolySheep AI の料金計算（GPT-4.1の場合）
        cost_per_mtok = 8.00  # USD/MTok
        total_cost = (prompt_tokens + completion_tokens) * cost_per_mtok / 1_000_000
        
        self.conn.execute("""
            INSERT INTO api_requests 
            (timestamp, model, prompt_tokens, completion_tokens, 
             total_cost_usd, latency_ms, request_text, response_text, 
             status, error_message)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            datetime.utcnow().isoformat(),
            model,
            prompt_tokens,
            completion_tokens,
            total_cost,
            latency_ms,
            request_text,
            response.choices[0].message.content if not error else None,
            "success" if not error else "error",
            str(error) if error else None
        ))
        self.conn.commit()

archiver = RequestArchiver()

実際のリクエスト例
start = datetime.now()
try:
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": "Hello, world!"}]
    )
    latency = (datetime.now() - start).total_seconds() * 1000
    archiver.archive("gpt-4.1", "Hello, world!", response, latency)
    print(f"成功: レイテンシ {latency:.2f}ms")
except Exception as e:
    print(f"エラー: {e}")

Proxyパターンを使用した中間層アプローチ

# archive_proxy.py
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
import httpx
import asyncio
import json
from datetime import datetime
from typing import Generator
import hashlib

app = FastAPI()

ARCHIVE_ENDPOINT = "https://api.holysheep.ai/v1"
HEADERS = {"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}

async def archive_log(
    request_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    latency_ms: float,
    status: str
):
    # アーカイブログ保存処理（Redis, S3, BigQuery等）
    log_entry = {
        "request_id": request_id,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_cost": (input_tokens + output_tokens) / 1_000_000 * 8.00,  # GPT-4.1
        "latency_ms": latency_ms,
        "status": status,
        "timestamp": datetime.utcnow().isoformat()
    }
    print(f"[ARCHIVE] {json.dumps(log_entry)}")

@app.api_route("/v1/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(path: str, request: Request):
    request_id = hashlib.md5(f"{datetime.now()}".encode()).hexdigest()[:16]
    start_time = datetime.now()
    
    body = await request.body()
    headers = dict(request.headers)
    headers["Authorization"] = HEADERS["Authorization"]
    
    async with httpx.AsyncClient(timeout=120.0) as client:
        try:
            upstream = client.build_request(
                method=request.method,
                url=f"{ARCHIVE_ENDPOINT}/{path}",
                headers=headers,
                content=body
            )
            
            response = await client.send(upstream)
            latency_ms = (datetime.now() - start_time).total_seconds() * 1000
            
            # レスポンスボディを解析してトークン数等を記録
            try:
                resp_data = response.json()
                usage = resp_data.get("usage", {})
                await archive_log(
                    request_id=request_id,
                    model=resp_data.get("model", "unknown"),
                    input_tokens=usage.get("prompt_tokens", 0),
                    output_tokens=usage.get("completion_tokens", 0),
                    latency_ms=latency_ms,
                    status="success"
                )
            except:
                pass
            
            return Response(
                content=response.content,
                status_code=response.status_code,
                headers=dict(response.headers)
            )
            
        except httpx.TimeoutException as e:
            latency_ms = (datetime.now() - start_time).total_seconds() * 1000
            await archive_log(request_id, "unknown", 0, 0, latency_ms, "timeout")
            return Response(
                content=json.dumps({"error": "timeout"}).encode(),
                status_code=504
            )
        except httpx.HTTPStatusError as e:
            await archive_log(request_id, "unknown", 0, 0, 0, f"http_{e.response.status_code}")
            raise

HolySheep AI の料金優位性

今すぐ登録して、成本効率の高いAI APIを活用しましょう。HolySheep AI は2026年現在の料金を次に示します：

モデル	価格 (USD/MTok)
GPT-4.1	$8.00
Claude Sonnet 4.5	$15.00
Gemini 2.5 Flash	$2.50
DeepSeek V3.2	$0.42

公式為替レート¥7=$1のところ、HolySheep AI は¥1=$1（85%節約）で提供。さらに50ミリ秒未満の低レイテンシを実現しています。

高度なアーカイブ戦略

ベクトルデータベースを使ったセマンティック検索

# レスポンスのベクトル化と保存
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticArchiver:
    def __init__(self, dimension=384):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.index = faiss.IndexFlatL2(dimension)
        self.responses = []
    
    def add_response(self, text: str, metadata: dict):
        embedding = self.model.encode([text])
        self.index.add(np.array(embedding).astype('float32'))
        self.responses.append({
            "text": text,
            "metadata": metadata,
            "embedding_id": len(self.responses)
        })
    
    def search(self, query: str, top_k: int = 5):
        query_embedding = self.model.encode([query])
        distances, indices = self.index.search(
            np.array(query_embedding).astype('float32'), top_k
        )
        return [self.responses[i] for i in indices[0] if i < len(self.responses)]
    
    def find_similar_responses(self, problematic_output: str):
        """類似の問題出力查找 — パターン発見に有効"""
        return self.search(problematic_output, top_k=10)

使用例
archiver = SemanticArchiver()
archiver.add_response(
    "日本の首都は東京です。",
    {"model": "gpt-4.1", "timestamp": "2026-01-15T10:30:00Z"}
)

類似回答を検索
similar = archiver.find_similar_responses("東京都が首都です")
for item in similar:
    print(f"スコア: {item['metadata']}")

よくあるエラーと対処法

エラー1: ConnectionError: timeout

# 症状: requests.exceptions.ConnectionError: HTTPSConnectionPool
原因: ネットワークタイムアウトまたはプロキシ設定の問題

解決策: タイムアウト設定とリトライロジックを追加
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def robust_request(prompt: str) -> str:
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(60.0, connect=10.0)
    ) as client:
        response = await client.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}]
            }
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]

これでConnectionErrorが発生しても自動リトライ

エラー2: 401 Unauthorized

# 症状: {"error": {"message": "Incorrect API key provided", "type": "invalid_request_error"}}
原因: APIキーが無効または期限切れ

解決策: 環境変数から安全にキーを読み込み、有効性チェック
import os
import httpx

def get_validated_client():
    api_key = os.environ.get("HOLYSHEEP_API_KEY")
    if not api_key or not api_key.startswith("sk-"):
        raise ValueError("無効なAPIキー: sk-から始まるキーを設定してください")
    
    client = httpx.AsyncClient(
        base_url="https://api.holysheep.ai/v1",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30.0
    )
    return client

async def verify_connection():
    async with get_validated_client() as client:
        response = await client.get("/models")
        if response.status_code == 401:
            raise RuntimeError("APIキーが無効です。HolySheepで新しいキーを発行してください")
        return response.json()

エラー3: RateLimitError: レート制限超過

# 症状: {"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
原因: 短時間过多リクエスト

解決策: 指数関数的バックオフで自動リトライ
import asyncio
import time
from collections import deque

class RateLimitHandler:
    def __init__(self, max_requests_per_minute=60):
        self.max_requests = max_requests_per_minute
        self.request_times = deque()
    
    async def wait_if_needed(self):
        now = time.time()
        # 1分以内に许可されたリクエストのみ残す
        while self.request_times and self.request_times[0] < now - 60:
            self.request_times.popleft()
        
        if len(self.request_times) >= self.max_requests:
            wait_time = 60 - (now - self.request_times[0])
            await asyncio.sleep(wait_time)
            return self.wait_if_needed()
        
        self.request_times.append(time.time())
        return True

async def rate_limited_request(prompt: str, handler: RateLimitHandler):
    await handler.wait_if_needed()
    
    async with httpx.AsyncClient() as client:
        for attempt in range(3):
            try:
                response = await client.post(
                    "https://api.holysheep.ai/v1/chat/completions",
                    headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
                    json={"model": "gpt-4.1", "messages": [{"role": "user", "content": prompt}]}
                )
                if response.status_code == 429:
                    await asyncio.sleep(2 ** attempt)
                    continue
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                if e.response.status_code != 429:
                    raise
        raise RuntimeError("レート制限超過: 多次リトライ後も失敗")

エラー4: InvalidRequestError: モデル指定エラー

# 症状: {"error": {"message": "Model not found", "type": "invalid_request_error"}}
原因: 存在しないモデル名を指定

解決策: 利用可能なモデルを動的に取得してバリデーション
async def get_available_models():
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.holysheep.ai/v1/models",
            headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
        )
        response.raise_for_status()
        data = response.json()
        return {m["id"] for m in data["data"]}

利用可能なモデルリストでバリデーション
AVAILABLE_MODELS = None

async def validate_and_send(model: str, prompt: str):
    global AVAILABLE_MODELS
    if AVAILABLE_MODELS is None:
        AVAILABLE_MODELS = await get_available_models()
    
    if model not in AVAILABLE_MODELS:
        supported = ", ".join(sorted(AVAILABLE_MODELS))
        raise ValueError(f"不明なモデル: {model}. 利用可能: {supported}")
    
    # リクエスト送信
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}]
            }
        )
        return response.json()

安全な呼び出し
try:
    result = await validate_and_send("gpt-4.1", "Hello")
except ValueError as e:
    print(f"バリデーションエラー: {e}")

ベストプラクティス

• 非同期処理: 高負荷環境ではasync/awaitを活用し、パフォーマンスを最大化
• 構造化ログ: JSON形式的死前でログを記録し、後続の分析を容易に
• 暗号化: 機密情報を含むリクエストボディはAES-256で暗号化保存
• パーティション分割: 大量ログは日付/時間でパーティション化し、クエリ性能を向上
• コスト監視: リアルタイムでAPI利用料を集計し、予算超過をアラート

まとめ

AI APIのリクエスト・レスポンスアーカイブは、運用において不可或缺的の存在です。適切なログ設計とエラーーハンドリングを組み合わせることで、安定したAI活用が可能になります。

HolySheep AI は¥1=$1の為替レートと50ミリ秒未満のレイテンシで、コストパフォーマンスに优异的です。今すぐ登録して、新規ユーザー向けの無料クレジット用于获取始めましょう。

👉 HolySheep AI に登録して無料クレジットを獲得