=2.12.0"
- ComponentName: "aws.greengrass.LocalHttpServer"
VersionRequirement: ">=2.1.0"
# greengrass_ai_component/artifacts/com.holysheep.ai-inference/1.2.0/inference_component.py
import asyncio
import hashlib
import json
import logging
import sqlite3
import time
from dataclasses import dataclass
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
import threading
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class InferenceConfig:
"""推論設定クラス - HolySheep API 統合"""
api_base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY" # 環境変数から注入推奨
model: str = "deepseek-chat"
max_retries: int = 3
timeout: int = 10
concurrent_limit: int = 5
cache_ttl: int = 3600
class TokenBucket:
"""トークンバケツ方式のレート制限"""
def __init__(self, rate: int, capacity: int):
self.rate = rate # tokens/second
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self._lock = threading.Lock()
def acquire(self, tokens: int = 1) -> bool:
with self._lock:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_time(self, tokens: int = 1) -> float:
with self._lock:
if self.tokens >= tokens:
return 0.0
return (tokens - self.tokens) / self.rate
class SemanticCache:
"""セマンティックキャッシュ - LLM 出力の重複排除"""
def __init__(self, db_path: str = "/greengrass/state/semantic_cache.db"):
self.db_path = db_path
self._init_db()
self._lock = threading.Lock()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn