AI APIリクエスト内容フィルタリングと機密情報脱敏の実践的実装

リクエストフィルタリングの基礎設計

AI APIへのリクエストフィルタリングは、以下の3層構造で実装することを推奨します。

1. 入力検証レイヤー

import re
import hashlib
from typing import Optional, Dict, Any, List

class RequestFilter:
    """AI APIリクエストフィルタリングクラス"""
    
    # 機密情報のパターン定義
    SENSITIVE_PATTERNS = {
        'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
        'phone': r'\b0\d{1,4}[-\s]?\d{1,4}[-\s]?\d{4}\b',
        'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        'ssn': r'\b\d{3}[-\s]?\d{2}[-\s]?\d{4}\b',
        'password': r'(password|pwd|pass)[\s:=]+\S+',
        'api_key': r'(api[_-]?key|secret|token)[\s:=]+\S+',
    }
    
    # ブロックリスト（有害コンテンツ）
    BLOCKED_CONTENT = [
        'violence', 'hate', 'illegal', 'fraud', 'malware'
    ]
    
    def __init__(self, api_base_url: str = "https://api.holysheep.ai/v1"):
        self.api_base_url = api_base_url
    
    def detect_sensitive_info(self, text: str) -> Dict[str, List[str]]:
        """テキストから機密情報を検出"""
        detected = {}
        for info_type, pattern in self.SENSITIVE_PATTERNS.items():
            matches = re.findall(pattern, text, re.IGNORECASE)
            if matches:
                detected[info_type] = matches
        return detected
    
    def mask_sensitive_data(self, text: str, mask_char: str = '*') -> str:
        """機密情報をMASK文字に置換"""
        masked_text = text
        
        # クレジットカード
        masked_text = re.sub(
            r'\b(\d{4})[-\s]?(\d{4})[-\s]?(\d{4})[-\s]?(\d{4})\b',
            r'\1****-****-****-\4', masked_text
        )
        
        # メールアドレス
        masked_text = re.sub(
            r'([A-Za-z0-9._%+-]+)@([A-Za-z0-9.-]+\.[A-Z|a-z]{2,})',
            r'****@\2', masked_text
        )
        
        # 電話番号
        masked_text = re.sub(
            r'\b(0\d{1,4})[-\s]?(\d{1,4})[-\s]?(\d{4})\b',
            r'\1***-****', masked_text
        )
        
        # APIキー/パスワード
        masked_text = re.sub(
            r'(api[_-]?key|secret|token|password|pwd)[\s:=]+(\S+)',
            r'\1=MASKED', masked_text, flags=re.IGNORECASE
        )
        
        return masked_text
    
    def validate_request(self, content: str) -> tuple[bool, Optional[str]]:
        """リクエスト内容の検証"""
        # 有害コンテンツチェック
        content_lower = content.lower()
        for blocked in self.BLOCKED_CONTENT:
            if blocked in content_lower:
                return False, f"ブロックされたコンテンツ: {blocked}"
        
        return True, None

HolySheep AI API呼び出しクラス
class HolySheepAIClient:
    """HolySheep AI APIクライアント（自動脱敏機能付き）"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.filter = RequestFilter()
    
    def send_safe_request(self, user_content: str, model: str = "deepseek-v3.2") -> Dict[str, Any]:
        """安全なリクエストを送信"""
        
        # ステップ1: 機密情報検出
        detected = self.filter.detect_sensitive_info(user_content)
        
        # ステップ2: 自動脱敏
        masked_content = self.filter.mask_sensitive_data(user_content)
        
        # ステップ3: コンテンツ検証
        is_valid, error_msg = self.filter.validate_request(masked_content)
        if not is_valid:
            return {"error": error_msg, "status": "rejected"}
        
        # ログ出力（ MASK処理済み ）
        print(f"検出された機密情報: {detected}")
        print(f"脱敏後テキスト: {masked_content[:100]}...")
        
        # HolySheep API呼び出し
        # （実際のAPI呼び出しコードは省略）
        
        return {
            "status": "success",
            "detected_sensitive": list(detected.keys()),
            "masked_content": masked_content
        }

使用例
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
result = client.send_safe_request(
    " customer's email is [email protected], "
    "credit card: 1234-5678-9012-3456, "
    "API key = abc123xyz"
)
print(result)

2. コンテキストベースのフィルタリング

import json
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

@dataclass
class AuditLog:
    """監査ログ用データクラス"""
    timestamp: str
    request_id: str
    user_id: str
    detected_info: list = field(default_factory=list)
    masked_fields: list = field(default_factory=list)
    action: str
    status: str

class AdvancedFilter:
    """高度なフィルタリングシステム"""
    
    def __init__(self):
        self.audit_logs: list[AuditLog] = []
        self.user_context: dict = {}
    
    def process_with_audit(
        self, 
        content: str, 
        user_id: str,
        session_id: str
    ) -> tuple[str, AuditLog]:
        """
        監査付きのコンテンツ処理
        Returns: (masked_content, audit_log)
        """
        import uuid
        
        request_id = str(uuid.uuid4())
        timestamp = datetime.now().isoformat()
        
        # 機密情報検出
        basic_filter = RequestFilter()
        detected = basic_filter.detect_sensitive_info(content)
        masked = basic_filter.mask_sensitive_data(content)
        
        # コンテキスト評価
        context_risk = self._evaluate_context_risk(user_id, detected)
        
        # 監査ログ作成
        audit_log = AuditLog(
            timestamp=timestamp,
            request_id=request_id,
            user_id=user_id,
            detected_info=list(detected.keys()),
            masked_fields=[k for k in detected.keys()],
            action="masked",
            status="approved" if context_risk < 0.7 else "manual_review"
        )
        
        self.audit_logs.append(audit_log)
        
        return masked, audit_log
    
    def _evaluate_context_risk(
        self, 
        user_id: str, 
        detected: dict
    ) -> float:
        """コンテキストベースのリスク評価"""
        base_risk = 0.3
        
        # 検出された機密情報の種類数でリスク 증가
        risk_increase = len(detected) * 0.15
        
        # ユーザー履歴によるリスク調整
        user_violations = self.user_context.get(user_id, {}).get(
            'violations', 0
        )
        risk_increase += user_violations * 0.1
        
        return min(base_risk + risk_increase, 1.0)
    
    def export_audit_report(self, filepath: str):
        """監査レポートのエクスポート"""
        report = {
            "generated_at": datetime.now().isoformat(),
            "total_requests": len(self.audit_logs),
            "logs": [
                {
                    "timestamp": log.timestamp,
                    "request_id": log.request_id,
                    "user_id": log.user_id,
                    "detected_info": log.detected_info,
                    "status": log.status
                }
                for log in self.audit_logs
            ]
        }
        
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        return filepath

実践的な使用例
filter_system = AdvancedFilter()

複数のリクエストを処理
test_contents = [
    ("user123", "Please summarize my account balance: 1234-5678-9012-3456"),
    ("user456", "Send report to [email protected]"),
    ("user789", "My password is supersecret123"),
]

for user_id, content in test_contents:
    masked, log = filter_system.process_with_audit(content, user_id, "sess_001")
    print(f"[{log.status}] {user_id}: {log.detected_info}")

監査レポート生成
filter_system.export_audit_report("audit_report_2026.json")
print("監査レポートを生成しました")

HolySheep AI API統合のベストプラクティス

HolySheep AIでは、<50msの超低レイテンシとWeChat Pay/Alipay対応で、気軽にAPIを試すことができます。登録時には無料クレジットが付与されるため、実装検証的成本を最小限に抑えられます。

import requests
import time
from typing import Optional

class HolySheepIntegration:
    """
    HolySheep AI API統合クラス
    base_url: https://api.holysheep.ai/v1
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.timeout = 30
        self.max_retries = 3
    
    def chat_completion(
        self, 
        messages: list[dict],
        model: str = "deepseek-v3.2",
        temperature: float = 0.7,
        max_tokens: int = 1000
    ) -> Optional[dict]:
        """
        Chat Completions API呼び出し
        ※ api.openai.com や api.anthropic.com は使用しない
        """
        endpoint = f"{self.base_url}/chat/completions"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        for attempt in range(self.max_retries):
            try:
                response = requests.post(
                    endpoint,
                    headers=headers,
                    json=payload,
                    timeout=self.timeout
                )
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.Timeout:
                print(f"タイムアウト (試行 {attempt + 1}/{self.max_retries})")
                if attempt < self.max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数バックオフ
                    
            except requests.exceptions.RequestException as e:
                print(f"リクエストエラー: {e}")
                break
        
        return None
    
    def streaming_completion(
        self,
        messages: list[dict],
        model: str = "deepseek-v3.2"
    ):
        """ストリーミング応答の取得"""
        endpoint = f"{self.base_url}/chat/completions"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "stream": True
        }
        
        response = requests.post(
            endpoint,
            headers=headers,
            json=payload,
            stream=True,
            timeout=self.timeout
        )
        
        for line in response.iter_lines():
            if line:
                data = line.decode('utf-8')
                if data.startswith('data: '):
                    if data.strip() == 'data: [DONE]':
                        break
                    yield json.loads(data[6:])

実践的な使用例
if __name__ == "__main__":
    client = HolySheepIntegration(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # システムプロンプト（コンテキスト分離）
    system_prompt = """あなたは安全なAIアシスタントです。
    ユーザーの入力から機密情報を検出し、脱敏処理を施してください。"""
    
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": "私のクレジットカード 1234-5678-9012-3456 の利用明細を教えてください"}
    ]
    
    result = client.chat_completion(messages=messages)
    
    if result:
        print(f"モデル: {result.get('model')}")
        print(f"応答: {result['choices'][0]['message']['content']}")

ダッシュボードでの利用状況監視

HolySheep AIのダッシュボードでは、API使用量・コスト・レイテンシをリアルタイムで監視できます。¥1=$1の為替レート 덕분에、コスト計算もシンプルです。

• リアルタイムメトリクス: Token使用量、API呼び出し回数、応答時間
• コスト分析: モデル別・ユーザー別のコスト内訳
• アラート設定: 使用量閾値を超えた場合の通知
• ключ 管理: 複数のAPIキーをプロジェクト別に管理

よくあるエラーと対処法

エラー1: 401 Unauthorized - 認証エラー

原因: APIキーが無効または期限切れ

対処法:

# 正しいkey形式を確認
API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # 先頭に余分なスペースなし

環境変数からの読み込みを推奨
import os
API_KEY = os.environ.get("HOLYSHEEP_API_KEY")

if not API_KEY:
    raise ValueError("APIキーが設定されていません")

エラー2: 429 Rate Limit Exceeded

原因: リクエスト頻度が上限を超過

対処法:

import time
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=60, period=60)  # 1分あたり60リクエスト
def safe_api_call():
    # API呼び出し処理
    pass

または指数バックオフでリトライ
def retry_with_backoff(func, max_retries=5):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            if "429" in str(e):
                wait_time = 2 ** i
                print(f"{wait_time}秒後にリトライ...")
                time.sleep(wait_time)
            else:
                raise

エラー3: 503 Service Unavailable

原因: サーバー過負荷またはメンテナンス

対処法: