AI業務活用 2026.04.30

AIエージェント暴走対策【料金爆発・誤情報・情報漏洩】本番運用の5つ防止策

タグ:生成AI / エージェント / 本番運用 / セキュリティ / コスト管理

3大リスクと対策マップ

リスク1: コスト暴走
  原因: 無限ループ・max_tokens未設定・バグによる大量リクエスト
  対策: Usage Limits設定 + max_tokens必須 + ループ上限

リスク2: ハルシネーション(誤情報)
  原因: 学習データに情報がない・古い情報で回答
  対策: RAG(検索拡張生成)+ 事実確認ステップ追加

リスク3: 情報漏洩
  原因: 機密情報をAPI経由で送信
  対策: PIIマスキング + 送信前フィルタリング + Enterprise利用

防止策1: レート制限付きAPIクライアント

# rate_limited_client.py
# トークン毎分・リクエスト毎分の制限を持つ安全なクライアント

import anthropic
import time
from collections import deque
from threading import Lock

class RateLimitedClaude:
    """レート制限とコスト監視を内蔵した安全なClaudeクライアント"""
    
    def __init__(self, rpm_limit: int = 50, daily_budget: float = 20.0):
        self.client = anthropic.Anthropic()
        self.rpm_limit = rpm_limit          # 毎分リクエスト上限
        self.daily_budget = daily_budget    # 日額予算
        self.request_times = deque()
        self.daily_cost = 0.0
        self.lock = Lock()
    
    def _check_rate(self):
        """レート制限チェック: 超過時はスリープ"""
        now = time.time()
        # 1分以上前のリクエストを削除
        while self.request_times and now - self.request_times[0] > 60:
            self.request_times.popleft()
        
        if len(self.request_times) >= self.rpm_limit:
            sleep_time = 60 - (now - self.request_times[0])
            if sleep_time > 0:
                print(f"⏳ レート制限: {sleep_time:.1f}秒待機")
                time.sleep(sleep_time)
    
    def ask(self, prompt: str, model: str = "claude-haiku-4-5-20251001",
            max_tokens: int = 512) -> str | None:
        """レート制限とコスト上限をチェックしてAPIコール"""
        with self.lock:
            if self.daily_cost >= self.daily_budget:
                print(f"⛔ 予算上限${self.daily_budget}に達しました")
                return None
            
            self._check_rate()
            self.request_times.append(time.time())
        
        response = self.client.messages.create(
            model=model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        
        inp, out = response.usage.input_tokens, response.usage.output_tokens
        price = {"claude-haiku-4-5-20251001": (0.80, 4.00),
                 "claude-sonnet-4-5": (3.00, 15.00)}.get(model, (3.00, 15.00))
        cost = (inp * price[0] + out * price[1]) / 1_000_000
        self.daily_cost += cost
        
        print(f"コスト: ${cost:.5f} | 今日の累計: ${self.daily_cost:.4f}")
        return response.content[0].text

# 使用例
safe_client = RateLimitedClaude(rpm_limit=30, daily_budget=5.0)
result = safe_client.ask("Pythonの主なデータ構造を3つ教えてください")
print(result)

防止策2: PIIマスキング

# pii_filter.py
# 機密情報をマスクしてからAPIに送信する

import re

PII_PATTERNS = {
    "email": (r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]"),
    "phone": (r"\b\d{2,4}[-\s]?\d{2,4}[-\s]?\d{3,4}\b", "[PHONE]"),
    "credit_card": (r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b", "[CARD]"),
    "api_key": (r"\b(sk-|ghp_|AKIA)[A-Za-z0-9]{10,}\b", "[API_KEY]"),
}

def mask_pii(text: str) -> tuple[str, list[str]]:
    """PIIを検出してマスクした文字列と検出リストを返す"""
    masked = text
    detected = []
    
    for pii_type, (pattern, replacement) in PII_PATTERNS.items():
        matches = re.findall(pattern, masked)
        if matches:
            detected.extend([f"{pii_type}: {m[:6]}..." for m in matches])
            masked = re.sub(pattern, replacement, masked)
    
    return masked, detected

def safe_ask(prompt: str) -> str | None:
    """PIIをマスクしてからAPIに送信する"""
    masked_prompt, detected = mask_pii(prompt)
    
    if detected:
        print(f"⚠️ PIIを検出してマスクしました: {detected}")
    
    import anthropic
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[{"role": "user", "content": masked_prompt}]
    )
    return response.content[0].text

# 使用例
result = safe_ask("田中さん(tanaka@company.co.jp)から API Key: sk-ant-12345abcde の確認がありました")
# ⚠️ PIIを検出: email: tanaka@..., api_key: sk-ant-...

防止策3: フォールバック付きマルチプロバイダー

# llm_with_fallback.py
# プロバイダー障害時に自動切り替えする

import anthropic
import openai
from typing import Optional

claude = anthropic.Anthropic()
oai = openai.OpenAI()

def llm_with_fallback(prompt: str, max_tokens: int = 512) -> tuple[str, str]:
    """Claude → GPT-4o-mini → Gemini の順でフォールバック"""
    
    # Provider 1: Claude Haiku
    try:
        response = claude.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text, "claude-haiku"
    except Exception as e:
        print(f"Claude失敗: {e}")
    
    # Provider 2: GPT-4o-mini
    try:
        response = oai.chat.completions.create(
            model="gpt-4o-mini",
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content, "gpt-4o-mini"
    except Exception as e:
        print(f"GPT失敗: {e}")
    
    return "全プロバイダーが利用不可です", "none"

answer, provider = llm_with_fallback("こんにちは、元気ですか?")
print(f"[{provider}] {answer}")

あわせて読みたい

参考ソース