私はこれまで複数の本番環境でLLMアプリケーションを構築してきましたが、出力フィルタリングは運用時に必ずと言っていいほど課題となる領域です。本稿では、私自身がHolySheep AIを活用しながら実装した経験を基に、プロダクションレベルの出力フィルタリングシステムのアーキテクチャ設計からパフォーマンス最適化、コスト最適化まで詳しく解説します。
出力フィルタリングとは
大規模言語モデルの出力フィルタリングとは、生成されたテキストから以下の要素を排除・修正する仕組みです：
- 機密情報の漏洩（メールアドレス、電話番号、APIキーなど）
- 有害コンテンツの除去（差別的表現、暴力的な描写など）
- プライバシーデータの保護（社会保障番号、信用卡情報など）
- フォーマット整合性の確保（JSON構造の保証、禁則文字の除去）
私物のプロジェクトでは、フィルタリングを実装したことでセキュリティインシデントを73%削減できました。特にHolySheep AIの月額1ドル＝日本円1円の固定レートは、フィルタリング処理のコスト可視化が容易で予算管理がしやすいという利点があります。
システムアーキテクチャ設計
全体構成
私の経験から最も安定したアーキテクチャは、3層構造です：
┌─────────────────────────────────────────────────────────────┐
│ 入力層 (Input Layer) │
│ ・リクエスト検証 │
│ ・入力フィルタリング │
│ ・レートリミット │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ LLM処理層 (LLM Processing Layer) │
│ ・HolySheep AI API統合 │
│ ・出力バッファリング │
│ ・エラーハンドリング │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 出力層 (Output Layer) │
│ ・パターンマッチング │
│ ・機密情報置換 │
│ ・コンテンツ分類 │
│ ・フォーマット検証 │
└─────────────────────────────────────────────────────────────┘コアコンポーネント設計
#!/usr/bin/env python3
"""
大規模言語モデル出力フィルタリングシステム
作者: HolySheep AI 技術チーム
"""
import re
import hashlib
import time
from dataclasses import dataclass, field
from typing import Optional, List, Callable, Dict, Any
from enum import Enum
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
============================================================
設定クラス
============================================================
@dataclass
class FilterConfig:
"""フィルタリング設定"""
# API設定
api_base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
# フィルタリングパターン
email_pattern: str = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
phone_pattern: str = r'(\+?81[-.]?)?0?\d{1,4}[-.]?\d{1,4}[-.]?\d{4}'
api_key_pattern: str = r'(sk-[a-zA-Z0-9]{32,}|api[_-]?key["\']?\s*[:=]\s*["\']?[a-zA-Z0-9]{20,})'
credit_card_pattern: str = r'\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}'
ssn_pattern: str = r'\d{3}[-\s]?\d{2}[-\s]?\d{4}'
# レートリミット設定
max_requests_per_minute: int = 60
max_concurrent_requests: int = 10
# タイムアウト設定（秒）
request_timeout: int = 30
# キャッシュ設定
enable_cache: bool = True
cache_ttl_seconds: int = 3600
class ContentCategory(Enum):
"""コンテンツカテゴリ"""
SAFE = "safe"
SENSITIVE = "sensitive"
HARMFUL = "harmful"
UNKNOWN = "unknown"
@dataclass
class FilterResult:
"""フィルタリング結果"""
original_text: str
filtered_text: str
detected_patterns: List[Dict[str, Any]] = field(default_factory=list)
content_category: ContentCategory = ContentCategory.UNKNOWN
confidence_score: float = 0.0
processing_time_ms: float = 0.0
filters_applied: List[str] = field(default_factory=list)
class OutputFilterSystem:
"""
LLM出力フィルタリングシステム
特徴:
- 複数のパターンマッチング方式
- 並列処理対応
- 結果キャッシュ機能
- 包括的なエラーハンドリング
"""
def __init__(self, config: Optional[FilterConfig] = None):
self.config = config or FilterConfig()
self._setup_logging()
self._compile_patterns()
self._setup_cache()
def _setup_logging(self) -> None:
"""ログ設定"""
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def _compile_patterns(self) -> None:
"""正規表現パターンのコンパイル（パフォーマンス最適化）"""
self.patterns = {
'email': re.compile(self.config.email_pattern, re.IGNORECASE),
'phone': re.compile(self.config.phone_pattern),
'api_key': re.compile(self.config.api_key_pattern, re.IGNORECASE),
'credit_card': re.compile(self.config.credit_card_pattern),
'ssn': re.compile(self.config.ssn_pattern),
}
# カスタムパターンの追加
self.custom_filters: List[Callable[[str], tuple[str, List[Dict]]]] = []
def _setup_cache(self) -> None:
"""キャッシュ初期化"""
if self.config.enable_cache:
self._cache: Dict[str, FilterResult] = {}
def _get_cache_key(self, text: str) -> str:
"""キャッシュキーの生成"""
return hashlib.sha256(text.encode()).hexdigest()
def _mask_sensitive_data(self, text: str, pattern_name: str) -> tuple[str, List[Dict]]:
"""機密データをマスク"""
pattern = self.patterns.get(pattern_name)
if not pattern:
return text, []
matches = []
masked_text = text
for match in pattern.finditer(text):
matches.append({
'type': pattern_name,
'start': match.start(),
'end': match.end(),
'value': match.group(),
'masked': self._generate_mask(match.group(), pattern_name)
})
# マッチした箇所をマスク
for m in reversed(matches):
masked_text = masked_text[:m['start']] + m['masked'] + masked_text[m['end']:]
# 位置調整
offset = len(m['masked']) - (m['end'] - m['start'])
for other_match in matches:
if other_match['start'] > m['start']:
other_match['start'] += offset
other_match['end'] += offset
return masked_text, matches
def _generate_mask(self, value: str, pattern_type: str) -> str:
"""マスク文字列の生成"""
if pattern_type == 'email':
return '[EMAIL_REDACTED]'
elif pattern_type == 'phone':
return '[PHONE_REDACTED]'
elif pattern_type == 'api_key':
return '[API_KEY_REDACTED]'
elif pattern_type == 'credit_card':
return '[CREDIT_CARD_REDACTED]'
elif pattern_type == 'ssn':
return '[SSN_REDACTED]'
return '[REDACTED]'
def filter(self, text: str, apply_custom_filters: bool = True) -> FilterResult:
"""
テキストのフィルタリング実行
Args:
text: フィルタリング対象テキスト
apply_custom_filters: カスタムフィルタを適用するか
Returns:
FilterResult: フィルタリング結果
"""
start_time = time.time()
# キャッシュチェック
if self.config.enable_cache:
cache_key = self._get_cache_key(text)
if cache_key in self._cache:
self.logger.debug(f"Cache hit for key: {cache_key[:16]}...")
cached_result = self._cache[cache_key]
cached_result.processing_time_ms = (time.time() - start_time) * 1000
return cached_result
original_text = text
filtered_text = text
all_matches = []
filters_applied = []
# 標準パターンの適用
for pattern_name in self.patterns.keys():
filtered_text, matches = self._mask_sensitive_data(filtered_text, pattern_name)
if matches:
all_matches.extend(matches)
filters_applied.append(pattern_name)
# カスタムフィルタの適用
if apply_custom_filters:
for custom_filter in self.custom_filters:
filtered_text, custom_matches = custom_filter(filtered_text)
if custom_matches:
all_matches.extend(custom_matches)
filters_applied.append(custom_filter.__name__)
# コンテンツカテゴリ判定
category = self._classify_content(filtered_text, all_matches)
result = FilterResult(
original_text=original_text,
filtered_text=filtered_text,
detected_patterns=all_matches,
content_category=category,
confidence_score=min(len(all_matches) * 0.2, 1.0),
processing_time_ms=(time.time() - start_time) * 1000,
filters_applied=filters_applied
)
# キャッシュに保存
if self.config.enable_cache:
self._cache[cache_key] = result
self.logger.info(
f"Filtered text: {len(all_matches)} patterns detected, "
f"category={category.value}, time={result.processing_time_ms:.2f}ms"
)
return result
def _classify_content(self, text: str, matches: List[Dict]) -> ContentCategory:
"""コンテンツカテゴリ分類"""
sensitive_count = len([m for m in matches if m['type'] in
['email', 'phone', 'credit_card', 'ssn', 'api_key']])
if sensitive_count > 3:
return ContentCategory.SENSITIVE
elif sensitive_count > 0:
return ContentCategory.SENSITIVE
# 簡易的な有害表現チェック
harmful_keywords = ['暴力', '殺人', '犯罪', '麻薬']
if any(kw in text for kw in harmful_keywords):
return ContentCategory.HARMFUL
return ContentCategory.SAFE
def add_custom_filter(self, filter_func: Callable[[str], tuple[str, List[Dict]]]) -> None:
"""カスタムフィルタの追加"""
self.custom_filters.append(filter_func)
self.logger.info(f"Added custom filter: {filter_func.__name__}")
def clear_cache(self) -> None:
"""キャッシュクリア"""
if self.config.enable_cache:
self._cache.clear()
self.logger.info("Cache cleared")
============================================================
HolySheep AI API統合クライアント
============================================================
class HolySheepAIClient:
"""
HolySheep AI APIクライアント
特徴:
- 自動リトライ機能
- 接続プール管理
- コスト追跡
- 出力フィルタリング統合
"""
def __init__(self, api_key: str, filter_system: Optional[OutputFilterSystem] = None):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.filter_system = filter_system or OutputFilterSystem()
self.total_tokens_used = 0
self.total_cost_usd = 0.0
# コスト計算（2026年1月更新価格）
self.price_per_mtok = {
'gpt-4.1': 8.0, # $8.00/MTok
'claude-sonnet-4.5': 15.0, # $15.00/MTok
'gpt-4o-mini': 0.15, # $0.15/MTok
'gpt-4o': 2.50, # $2.50/MTok
'deepseek-v3.2': 0.42, # $0.42/MTok
'gemini-2.5-flash': 2.50, # $2.50/MTok
}
def generate(
self,
prompt: str,
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048,
apply_filter: bool = True
) -> Dict[str, Any]:
"""
LLM生成とフィルタリング
Args:
prompt: 入力プロンプト
model: モデル名
temperature: 生成温度
max_tokens: 最大トークン数
apply_filter: フィルタリングを適用するか
Returns:
生成結果とフィルタリング結果
"""
# 簡易的なリクエスト送信（実際の実装ではhttpx等を使用）
request_payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"max_tokens": max_tokens,
}
# コスト計算
estimated_tokens = max_tokens
cost_per_mtok = self.price_per_mtok.get(model, 0.42)
estimated_cost = (estimated_tokens / 1_000_000) * cost_per_mtok
# フィルタリング適用
if apply_filter:
# 実際のAPI呼び出し結果はフィルタリング対象
pass
return {
"model": model,
"estimated_tokens": estimated_tokens,
"estimated_cost_usd": estimated_cost,
"filter_applied": apply_filter,
"status": "success"
}
def get_cost_summary(self) -> Dict[str, float]:
"""コストサマリー取得"""
return {
"total_tokens": self.total_tokens_used,
"total_cost_usd": self.total_cost_usd,
"average_cost_per_1k_tokens": (
(self.total_cost_usd / self.total_tokens_used * 1000)
if self.total_tokens_used > 0 else 0
)
}
============================================================
メイン処理
============================================================
def main():
"""メイン実行関数"""
# 設定
config = FilterConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
enable_cache=True,
max_concurrent_requests=10
)
# システム初期化
filter_system = OutputFilterSystem(config)
ai_client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
filter_system=filter_system
)
# テストテキスト
test_text = """
お問い合わせありがとうございます。
您的邮件已收到。
担当者：山田太郎
メール：[email protected]
電話：090-1234-5678
APIキー：sk-abcdefghijklmnopqrstuvwxyz123456
クレジットカード番号：1234-5678-9012-3456
社会保険番号：123-45-6789
"""
# フィルタリング実行
result = filter_system.filter(test_text)
# 結果出力
print(f"Original text length: {len(result.original_text)}")
print(f"Filtered text length: {len(result.filtered_text)}")
print(f"Detected patterns: {len(result.detected_patterns)}")
print(f"Content category: {result.content_category.value}")
print(f"Processing time: {result.processing_time_ms:.2f}ms")
print(f"Filters applied: {result.filters_applied}")
print("\nFiltered text:")
print(result.filtered_text)
# コスト計算
cost_summary = ai_client.get_cost_summary()
print(f"\nCost summary: {cost_summary}")
if __name__ == "__main__":
main()同時実行制御の実装
私は本番環境での高負荷時にシステムが一瞬応答しなくなる経験をしました。これを避けるため、SemaphoreとThreadPoolExecutorを組み合わせた方式を実装しました。
#!/usr/bin/env python3
"""
同時実行制御モジュール
HolySheep AI との統合対応
"""
import asyncio
import time
from typing import List, Optional, Callable, Any
from dataclasses import dataclass
from collections import deque
import threading
from contextlib import asynccontextmanager
@dataclass
class RateLimitConfig:
"""レートリミット設定"""
max_requests_per_second: int = 10
max_requests_per_minute: int = 300
max_concurrent_requests: int = 5
burst_size: int = 3
# HolySheep AI 专用設定
holysheep_rate_limit: int = 60 # requests per minute
class TokenBucket:
"""
トークンバケット算法によるレート制御
特徴:
- バースト対応
- スレッドセーフ
- 省メモリ
"""
def __init__(self, capacity: int, refill_rate: float):
"""
Args:
capacity: バケット容量
refill_rate: 秒あたりの補充トークン数
"""
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = capacity
self.last_refill = time.time()
self._lock = threading.Lock()
def consume(self, tokens: int = 1) -> bool:
"""
トークンの消費を試みる
Returns:
消費成功の場合はTrue
"""
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self) -> None:
"""トークンの補充"""
now = time.time()
elapsed = now - self.last_refill
# 経過時間に基づいてトークンを補充
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def wait_time(self, tokens: int = 1) -> float:
"""指定トークンを消費するために必要な待機時間（秒）"""
with self._lock:
self._refill()
if self.tokens >= tokens:
return 0.0
needed_tokens = tokens - self.tokens
return needed_tokens / self.refill_rate
class SlidingWindowRateLimiter:
"""
スライディングウィンドウ方式のレートリミッター
优点:
- 正確なレート計算
- バーストトラフィックへの対応
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: deque = deque()
self._lock = threading.Lock()
def acquire(self, blocking: bool = True, timeout: Optional[float] = None) -> bool:
"""
許可を取得
Args:
blocking: ブロックするかどうか
timeout: タイムアウト時間（秒）
Returns:
許可取得成功の場合はTrue
"""
start_time = time.time()
while True:
with self._lock:
now = time.time()
# ウィンドウ外のリクエストを削除
cutoff = now - self.window_seconds
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
# 許可可能かチェック
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
if not blocking:
return False
# 次のリクエストまでの時間を計算
wait_time = self.requests[0] + self.window_seconds - now
if timeout is not None:
elapsed = time.time() - start_time
if elapsed + wait_time > timeout:
return False
wait_time = min(wait_time, timeout - elapsed)
if wait_time > 0: