近年、IoTデバイス、製造業の組立ライン 자율判断、遠隔医療、Connected Carsなど、低遅延が求められる applications でAI推論の需要が爆発的に増加しています。クラウド централизованный AI API 调用ではネットワーク往復時間を無視できず、50ms以上のレイテンシが発生することがあります。本稿では、HolySheep AIを活用したエッジ computing 環境向けのAI API中継局アーキテクチャ設計から実装、本番適用までを経験に基づく視点で詳しく解説します。
エッジ computing 環境におけるAI API中継局の必要性
エッジ computing とは、データの生成源に近い場所にcomputing ressourcesを配置するarchitectureです。AI推論をエッジで実行する利点は3つあります。
- 低レイテンシ：クラウド往返のRTTを排除し、10ms未満の推論応答を実現
- 帯域幅最適化：生データをクラウドに送信せず、要約・特徴量のみ転送
- データ主権：機密データを自施設内に留め、コンプライアンス要件を満たす
しかし、エッジデバイスだけでは大規模なLLM推論を処理できません。ここにAI API中継局的价值が現れます。中継局は複数のエッジノードからのリクエストを集約・負荷分散し、最適なタイミングでクラウドAPIを呼び出すことで、コストとパフォーマンスのバランスを取ります。
システムアーキテクチャ設計
全体構成
┌─────────────────────────────────────────────────────────────────────────────┐
│ Edge AI API Relay Architecture │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Edge │ │ Edge │ │ Edge │ ← 工場・倉庫・店舗 │
│ │ Node #1 │ │ Node #2 │ │ Node #N │ │
│ │ Raspberry│ │ Industrial│ │ Jetson │ │
│ │ Pi 5 │ │ Gateway │ │ Nano │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └───────────────┼───────────────┘ │
│ │ mTLS / TLS 1.3 │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Relay Station │ ← 中継局（Kubernetes / Docker Swarm） │
│ │ ┌────────────┐ │ │
│ │ │ API Gateway │ │ 認証・レートリミット・ログ │
│ │ ├────────────┤ │ │
│ │ │ Request │ │ キュー・優先度制御・重了剔除 │
│ │ │ Queue │ │ │
│ │ ├────────────┤ │ │
│ │ │ Load │ │ ヘルスチェック・自動復旧 │
│ │ │ Balancer │ │ │
│ │ └────────────┘ │ │
│ └────────┬─────────┘ │
│ │ │
│ ┌───────────┴───────────┐ │
│ ▼ ▼ │
│ ┌───────────────┐ ┌────────────────┐ │
│ │ HolySheep AI │ │ Fallback │ ← 冗長化・障害対応 │
│ │ API Gateway │ │ Provider │ │
│ │ https://api │ │ │ │
│ │ .holysheep.ai │ │ │ │
│ │ .ai/v1 │ │ │ │
│ └───────────────┘ └────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘コンポーネント詳細設計
中継局は4つの主要コンポーネントで構成されます。各コンポーネントの役割と設計上の考慮点を以下に示します。
| コンポーネント | 技術スタック | 主な役割 | 設計ポイント |
|---|---|---|---|
| API Gateway | nginx + Lua / Envoy | 認証・認可・レート制限 | mTLS終端、IPホワイトリスト |
| Request Queue | Redis Streams / RabbitMQ | リクエストバッファリング | FIFO保証、優先度キュー |
| Load Balancer | Consul + Fabio | バックエンド分散 | サーキットブレーカー |
| Cache Layer | Redis Cluster | 応答キャッシュ・重了排除 | TTL設計、ハッシュキー戦略 |
コア実装コード
Python SDKラッパー：中継局クライアント
エッジノードから中継局への接続を容易にするPythonクライアントSDKを実装します。接続プール、再試行ロジック、タイムアウト処理を組み込んだproduction-readyなコードです。
# edge_relay_client.py
エッジ computing 環境向け AI API 中継局クライアント SDK
HolySheep AI 公式エンドポイント対応
import asyncio
import hashlib
import json
import time
from dataclasses import dataclass, field
from typing import Optional, Any
from urllib.parse import urljoin
import aiohttp
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type
)
@dataclass
class RelayConfig:
"""中継局接続設定"""
relay_url: str = "https://api.holysheep.ai/v1" # HolySheep公式エンドポイント
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
timeout: int = 30
max_retries: int = 3
max_concurrent: int = 10
cache_ttl: int = 3600
enable_fallback: bool = True
@dataclass
class APIResponse:
"""API応答データクラス"""
content: str
model: str
tokens_used: int
latency_ms: float
cached: bool = False
error: Optional[str] = None
class EdgeRelayClient:
"""
エッジ computing 環境向け AI API 中継局クライアント
特徴:
- 非同期通信対応（asyncio）
- 自動リトライ（指数バックオフ）
- 接続プール管理
- 応答キャッシュ
- レート制限対応
"""
def __init__(self, config: Optional[RelayConfig] = None):
self.config = config or RelayConfig()
self._session: Optional[aiohttp.ClientSession] = None
self._semaphore = asyncio.Semaphore(self.config.max_concurrent)
self._cache: dict = {}
async def __aenter__(self):
"""非同期コンテキストマネージャー入口"""
await self._ensure_session()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""非同期コンテキストマネージャー出口"""
await self.close()
async def _ensure_session(self) -> aiohttp.ClientSession:
"""セッションの初期化・再利用"""
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=self.config.timeout)
self._session = aiohttp.ClientSession(timeout=timeout)
return self._session
def _generate_cache_key(self, model: str, messages: list) -> str:
"""キャッシュキーの生成（SHA256ハッシュ）"""
content = f"{model}:{json.dumps(messages, sort_keys=True)}"
return hashlib.sha256(content.encode()).hexdigest()
def _get_cached_response(self, cache_key: str) -> Optional[APIResponse]:
"""キャッシュ応答の取得"""
if cache_key in self._cache:
entry = self._cache[cache_key]
if time.time() - entry['timestamp'] < self.config.cache_ttl:
entry['response'].cached = True
return entry['response']
else:
del self._cache[cache_key]
return None
@retry(
retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def chat_completions(
self,
messages: list,
model: str = "gpt-4o",
temperature: float = 0.7,
max_tokens: int = 1024,
**kwargs
) -> APIResponse:
"""
チャット補完API호출（Edge最適化版）
Args:
messages: 会話メッセージリスト
model: モデル名（gpt-4o, claude-sonnet-4.5, gemini-2.0-flash, deepseek-v3.2）
temperature: 生成多様性パラメータ
max_tokens: 最大出力トークン数
Returns:
APIResponse: 応答データ
"""
cache_key = self._generate_cache_key(model, messages)
cached = self._get_cached_response(cache_key)
if cached:
return cached
async with self._semaphore:
start_time = time.perf_counter()
session = await self._ensure_session()
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json",
"X-Edge-Request": "true",
"X-Client-Version": "2.0.0"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
try:
async with session.post(
urljoin(self.config.relay_url, "/chat/completions"),
json=payload,
headers=headers
) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status == 200:
data = await response.json()
result = APIResponse(
content=data['choices'][0]['message']['content'],
model=data['model'],
tokens_used=data.get('usage', {}).get('total_tokens', 0),
latency_ms=latency_ms,
cached=False
)
self._cache[cache_key] = {
'response': result,
'timestamp': time.time()
}
return result
elif response.status == 429:
raise aiohttp.ClientResponseError(
response.request_info,
response.history,
message="Rate limit exceeded"
)
else:
error_text = await response.text()
return APIResponse(
content="",
model=model,
tokens_used=0,
latency_ms=latency_ms,
error=f"HTTP {response.status}: {error_text}"
)
except Exception as e:
if self.config.enable_fallback:
return await self._fallback_request(model, messages, start_time)
raise
async def _fallback_request(
self,
model: str,
messages: list,
start_time: float
) -> APIResponse:
"""フォールバック処理（代替モデルへの切り替え）"""
fallback_models = {
"gpt-4o": "gemini-2.0-flash",
"claude-sonnet-4.5": "deepseek-v3.2"
}
fallback_model = fallback_models.get(model, "deepseek-v3.2")
return await self.chat_completions(
messages=messages,
model=fallback_model,
_fallback=True
)
async def batch_completions(
self,
requests: list[dict],
concurrency: int = 5
) -> list[APIResponse]:
"""
バッチ処理（複数のリクエストを並列実行）
Args:
requests: リクエストリスト [{"messages": [...], "model": "..."}]
concurrency: 同時実行数
"""
semaphore = asyncio.Semaphore(concurrency)
async def limited_request(req: dict) -> APIResponse:
async with semaphore:
return await self.chat_completions(**req)
tasks = [limited_request(req) for req in requests]
return await asyncio.gather(*tasks, return_exceptions=True)
async def close(self):
"""リソースのクリーンアップ"""
if self._session and not self._session.closed:
await self._session.close()
self._cache.clear()
使用例
async def main():
config = RelayConfig(
relay_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=20,
cache_ttl=7200
)
async with EdgeRelayClient(config) as client:
# 単一リクエスト
response = await client.chat_completions(
messages=[
{"role": "system", "content": "あなたは工場の品質管理AIです。"},
{"role": "user", "content": "画像内の傷検出結果を判定してください。"}
],
model="gpt-4o",
temperature=0.3
)
print(f"応答: {response.content}")
print(f"レイテンシ: {response.latency_ms:.2f}ms")
print(f"トークン使用量: {response.tokens_used}")
print(f"キャッシュ命中: {response.cached}")
# バッチ処理
batch_requests = [
{"messages": [{"role": "user", "content": f"質問{i}"}]}
for i in range(10)
]
results = await client.batch_completions(batch_requests)
print(f"バッチ処理完了: {len(results)}件")
if __name__ == "__main__":
asyncio.run(main())Node.js実装：中継局サーバー
次に、中継局サーバー側の実装を示します。Express.jsを使用し、認証、ログ記録、負荷分散機能を実装します。
// relay-server.js
// Node.js + Express による AI API 中継局サーバー実装
// HolySheep AI へのプロキシ機能付き
const express = require('express');
const axios = require('axios');
const crypto = require('crypto');
const { RateLimiterMemory } = require('rate_limiter_core');
const Redis = require('ioredis');
// 設定
const CONFIG = {
// HolySheep AI 公式エンドポイント
HOLYSHEEP_BASE_URL: 'https://api.holysheep.ai/v1',
API_KEY: process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY',
// サーバー設定
PORT: process.env.PORT || 3000,
HOST: process.env.HOST || '0.0.0.0',
// キャッシュ設定（Redis）
REDIS_URL: process.env.REDIS_URL || 'redis://localhost:6379',
CACHE_TTL: 3600,
// レート制限
RATE_LIMIT_WINDOW: 60 * 1000,
RATE_LIMIT_MAX_REQUESTS: 100,
// タイムアウト
REQUEST_TIMEOUT: 30000,
KEEP_ALIVE_TIMEOUT: 65000
};
class RelayServer {
constructor() {
this.app = express();
this.server = null;
this.redis = null;
this.rateLimiter = new RateLimiterMemory({
points: CONFIG.RATE_LIMIT_MAX_REQUESTS,
duration: CONFIG.RATE_LIMIT_WINDOW / 1000
});
this.setupMiddleware();
this.setupRoutes();
this.setupErrorHandling();
}
setupMiddleware() {
// JSONボディパーサー
this.app.use(express.json({ limit: '10mb' }));
// リクエストログ
this.app.use((req, res, next) => {
req.startTime = Date.now();
res.on('finish', () => {
const duration = Date.now() - req.startTime;
console.log(JSON.stringify({
timestamp: new Date().toISOString(),
method: req.method,
path: req.path,
status: res.statusCode,
duration_ms: duration,
ip: req.ip,
user_agent: req.get('User-Agent')
}));
});
next();
});
// APIキー認証
this.app.use(this.authenticate.bind(this));
// レート制限
this.app.use(this.rateLimit.bind(this));
}
authenticate(req, res, next) {
const apiKey = req.headers['x-api-key'] || req.query.api_key;
if (!apiKey) {
return res.status(401).json({
error: 'Unauthorized',
message: 'API key is required'
});
}
// キーのハッシュ化（ログ出力用）
const keyHash = crypto.createHash('sha256').update(apiKey).digest('hex').substring(0, 8);
req.clientKey = keyHash;
next();
}
async rateLimit(req, res, next) {
const key = req.clientKey || req.ip;
try {
const result = await this.rateLimiter.consume(key);
res.set({
'X-RateLimit-Remaining': result.remainingPoints,
'X-RateLimit-Reset': result.msBeforeNext
});
next();
} catch (error) {
res.status(429).json({
error: 'Too Many Requests',
message: 'Rate limit exceeded. Please wait before retrying.',
retry_after: Math.ceil(error.msBeforeNext / 1000)
});
}
}
setupRoutes() {
// ヘルスチェック
this.app.get('/health', (req, res) => {
res.json({
status: 'healthy',
timestamp: new Date().toISOString(),
version: '2.0.0',
upstream: CONFIG.HOLYSHEEP_BASE_URL
});
});
// 準備状態チェック
this.app.get('/ready', async (req, res) => {
try {
if (this.redis) {
await this.redis.ping();
}
res.json({ status: 'ready', redis: 'connected' });
} catch (error) {
res.status(503).json({ status: 'not ready', error: error.message });
}
});
// チャット補完プロキシ
this.app.post('/chat/completions', this.proxyChatCompletions.bind(this));
// Embeddingプロキシ
this.app.post('/embeddings', this.proxyEmbeddings.bind(this));
// モデルリスト取得
this.app.get('/models', this.proxyModels.bind(this));
// 統計エンドポイント
this.app.get('/stats', this.getStats.bind(this));
}
generateCacheKey(body) {
const content = JSON.stringify(body);
return cache:${crypto.createHash('sha256').update(content).digest('hex')};
}
async proxyChatCompletions(req, res) {
const cacheKey = this.generateCacheKey(req.body);
const startTime = Date.now();
try {
// キャッシュチェック
if (this.redis) {
const cached = await this.redis.get(cacheKey);
if (cached) {
const data = JSON.parse(cached);
return res.json({
...data,
cached: true,
cache_hit: true
});
}
}
// HolySheep APIへの転送
const response = await axios.post(
${CONFIG.HOLYSHEEP_BASE_URL}/chat/completions,
req.body,
{
headers: {
'Authorization': Bearer ${CONFIG.API_KEY},
'Content-Type': 'application/json',
'X-Forwarded-For': req.ip,
'X-Real-Client': req.clientKey
},
timeout: CONFIG.REQUEST_TIMEOUT,
validateStatus: () => true
}
);
const latencyMs = Date.now() - startTime;
// 成功時のみキャッシュ
if (response.status === 200 && this.redis) {
await this.redis.setex(
cacheKey,
CONFIG.CACHE_TTL,
JSON.stringify(response.data)
);
}
// 統計記録
await this.recordMetrics(req.body.model || 'unknown', latencyMs, response.status);
res.status(response.status).json(response.data);
} catch (error) {
console.error('Proxy error:', error.message);
if (error.code === 'ECONNABORTED' || error.code === 'ETIMEDOUT') {
return res.status(504).json({
error: 'Gateway Timeout',
message: 'Upstream API timeout'
});
}
res.status(502).json({
error: 'Bad Gateway',
message: error.message
});
}
}
async proxyEmbeddings(req, res) {
// Embeddings APIのプロキシ実装
try {
const response = await axios.post(
${CONFIG.HOLYSHEEP_BASE_URL}/embeddings,
req.body,
{
headers: {
'Authorization': Bearer ${CONFIG.API_KEY},
'Content-Type': 'application/json'
},
timeout: CONFIG.REQUEST_TIMEOUT
}
);
res.json(response.data);
} catch (error) {
console.error('Embeddings proxy error:', error.message);
res.status(502).json({
error: 'Bad Gateway',
message: error.message
});
}
}
async proxyModels(req, res) {
try {
const response = await axios.get(
${CONFIG.HOLYSHEEP_BASE_URL}/models,
{
headers: {
'Authorization': Bearer ${CONFIG.API_KEY}
},
timeout: 10000
}
);
res.json(response.data);
} catch (error) {
res.status(502).json({
error: 'Failed to fetch models',
message: error.message
});
}
}
async recordMetrics(model, latencyMs, status) {
if (!this.redis) return;
const now = new Date();
const hourKey = stats:${model}:${now.toISOString().substring(0, 13)};
const pipeline = this.redis.pipeline();
pipeline.hincrby(hourKey, 'requests', 1);
pipeline.hincrby(hourKey, status_${status}, 1);
pipeline.hincrbyfloat(hourKey, 'total_latency', latencyMs);
pipeline.expire(hourKey, 86400 * 7); // 7日間保持
await pipeline.exec();
}
async getStats(req, res) {
if (!this.redis) {
return res.json({ message: 'Redis not configured' });
}
const now = new Date();
const hourKey = stats:*:${now.toISOString().substring(0, 13)};
const keys = await this.redis.keys(hourKey);
const stats = {};
for (const key of keys) {
const data = await this.redis.hgetall(key);
if (data) {
const model = key.split(':')[1];
stats[model] = {
requests: parseInt(data.requests) || 0,
success: parseInt(data.status_200) || 0,
errors: Object.entries(data)
.filter(([k]) => k.startsWith('status_') && k !== 'status_200')
.reduce((sum, [, v]) => sum + parseInt(v), 0),
avg_latency_ms: data.requests > 0
? (parseFloat(data.total_latency) / parseInt(data.requests)).toFixed(2)
: 0
};
}
}
res.json({
timestamp: now.toISOString(),
stats
});
}
setupErrorHandling() {
// 404 handler
this.app.use((req, res) => {
res.status(404).json({
error: 'Not Found',
message: Route ${req.method} ${req.path} not found
});
});
// Global error handler
this.app.use((err, req, res, next) => {
console.error('Unhandled error:', err);
res.status(500).json({
error: 'Internal Server Error',
message: process.env.NODE_ENV === 'production'
? 'An unexpected error occurred'
: err.message
});
});
}
async start() {
// Redis接続（オプション）
try {
this.redis = new Redis(CONFIG.REDIS_URL, {
lazyConnect: true,
maxRetriesPerRequest: 3
});
await this.redis.connect();
console.log('Redis connected');
} catch (error) {
console.warn('Redis connection failed, running without cache:', error.message);
this.redis = null;
}
return new Promise((resolve) => {
this.server = this.app.listen(CONFIG.PORT, CONFIG.HOST, () => {
console.log(Relay Server running on http://${CONFIG.HOST}:${CONFIG.PORT});
console.log(Upstream: ${CONFIG.HOLYSHEEP_BASE_URL});
resolve();
});
this.server.keepAliveTimeout = CONFIG.KEEP_ALIVE_TIMEOUT;
});
}
async stop() {
if (this.server) {
await new Promise(resolve => this.server.close(resolve));
}
if (this.redis) {
await this.redis.quit();
}
}
}
// 起動
if (require.main === module) {
const server = new RelayServer();
process.on('SIGTERM', async () => {
console.log('SIGTERM received, shutting down...');
await server.stop();
process.exit(0);
});
process.on('SIGINT', async () => {
console.log('SIGINT received, shutting down...');
await server.stop();
process.exit(0);
});
server.start().catch(console.error);
}
module.exports = RelayServer;同時実行制御の実装
エッジ computing 環境では、リソースが限られるため同時実行制御が重要です。私は以前、工場の品質検査ラインで100台以上のエッジカメラからのリクエストを処理する際に、この制御なしではシステムが不安定になりました。以下に、SemaphoreとWorker Poolパターンを組み合わせた実装を示します。
# concurrent_controller.py
同時実行制御モジュール：Semaphore + Worker Pool
import asyncio
import time
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from enum import Enum
from collections import deque
import threading
class Priority(Enum):
"""リクエスト優先度"""
CRITICAL = 1 # 安全停止命令など
HIGH = 2 # 品質検査など
NORMAL = 3 # 一般的リクエスト
LOW = 4 # ログ・アナリティクス
@dataclass
class QueuedRequest:
"""キュー内リクエスト"""
priority: Priority
timestamp: float
future: asyncio.Future
args: tuple
kwargs: dict
callback: Optional[Callable] = None
def __lt__(self, other):
if self.priority.value != other.priority.value:
return self.priority.value < other.priority.value
return self.timestamp < other.timestamp
class PriorityQueue:
"""優先度キュー実装"""
def __init__(self):
self._queues = {p: deque() for p in Priority}
self._lock = threading.Lock()
self._not_empty = threading.Condition(self._lock)
def push(self, request: QueuedRequest):
with self._not_empty:
self._queues[request.priority].append(request)
self._not_empty.notify()
def pop(self) -> Optional[QueuedRequest]:
with self._not_empty:
while True:
for priority in Priority:
queue = self._queues[priority]
if queue:
return queue.popleft()
self._not_empty.wait(timeout=1.0)
# タイムアウト時はNoneを返す
return None
def size(self) -> int:
return sum(len(q) for q in self._queues.values())
def is_empty(self) -> bool:
return self.size() == 0
class WorkerPool:
"""
ワーカープール管理
指定されたサイズのワーカーで非同期タスクを実行。
優先度キューにより重要なリクエストを先に処理。
"""
def __init__(
self,
max_workers: int = 10,
max_queue_size: int = 1000,
default_timeout: float = 30.0
):
self.max_workers = max_workers
self.max_queue_size = max_queue_size
self.default_timeout = default_timeout
self._queue = PriorityQueue()
self._workers: list[asyncio.Task] = []
self._running = False
self._metrics = {
'processed': 0,
'failed': 0,
'rejected': 0,
'total_latency': 0.0
}
self._metrics_lock = threading.Lock()
async def _worker(self, worker_id: int):
"""Individual worker coroutine"""
print(f"Worker #{worker_id} started")
while self._running:
request = self._queue.pop()
if request is None:
continue
if request.future.cancelled():
continue
start_time = time.perf_counter()
try:
# タイムアウト処理
result = await asyncio.wait_for(
request.future,
timeout=request.kwargs.get('timeout', self.default_timeout)
)
latency = time.perf_counter() - start_time
with self._metrics_lock:
self._metrics['processed'] += 1
self._metrics['total_latency'] += latency
if request.callback:
request.callback(result)
except asyncio.TimeoutError:
with self._metrics_lock:
self._metrics['failed'] += 1
request.future.set_exception(
TimeoutError(f"Request timeout after {self.default_timeout}s")
)
except Exception as e:
with self._metrics_lock:
self._metrics['failed'] += 1
request.future.set_exception(e)
print(f"Worker #{worker_id} stopped")
async def start(self):
"""ワーカープールの起動"""
self._running = True
self._workers = [
asyncio.create_task(self._worker(i))
for i in range(self.max_workers)
]
async def stop(self, wait: bool = True):
"""ワーカープールの停止"""
self._running = False
if wait:
await asyncio.gather(*self._workers, return_exceptions=True)
self._workers.clear()
def submit(
self,
coro: Callable,
priority: Priority = Priority.NORMAL,
timeout: Optional[float] = None,
callback: Optional[Callable] = None,
**kwargs
) -> asyncio.Future:
"""
タスクの提交
Args:
coro: 実行するコルーチン
priority: 優先度
timeout: タイムアウト秒数
callback: 完了時コールバック
Returns:
asyncio.Future: 結果を受け取るFuture
"""
if self._queue.size() >= self.max_queue_size:
raise RuntimeError(f"Queue full (max: {self.max_queue_size})")
future = asyncio.Future()
request = QueuedRequest(
priority=priority,
timestamp=time.time(),
future=future,
args=(coro,),
kwargs={**kwargs, 'timeout': timeout},
callback=callback
)
# コルーチンをFutureに変換
asyncio.create_task(self._execute_request(request))
return future
async def _execute_request(self, request: QueuedRequest):
"""リクエストの実行"""
self._queue.push(request)
try:
# _workerがpopするまでの待機
while request.future.pending():
await asyncio.sleep(0.01)
result = await request.future
return result
except Exception as e:
raise
def get_metrics(self) -> dict:
"""メトリクスの取得"""
with self._metrics_lock:
metrics = self._metrics.copy()
if metrics['processed'] > 0:
metrics['avg_latency_ms'] = (
metrics['total_latency'] / metrics['processed'] * 1000
)
else:
metrics['avg_latency_ms'] = 0
metrics['queue_size'] = self._queue.size()
metrics['active_workers'] = len(self._workers)
return metrics
使用例
async def example_usage():
pool = WorkerPool(max_workers=5, max_queue_size=100)
await pool.start()
async def ai_task(task_id: int, delay: float = 0.1):
await asyncio.sleep(delay)
return f"Task {task_id} completed"
# 優先度の高いリクエスト
critical = pool.submit(
ai_task(1, 0.05),
priority=Priority.CRITICAL,
timeout=5.0
)
# 通常のリクエスト
normal = pool.submit(
ai_task(2, 0.5),
priority=Priority.NORMAL
)
# 結果待機
result = await critical
print(f"Critical result: {result}")
result = await normal
print(f"Normal result: {result}")
# メトリクス確認
print(f"Metrics: {pool.get_metrics()}")
await pool.stop()
if __name__ == "__main__":
asyncio.run(example_usage())パフォーマンスベンチマーク
私の検証環境（Intel Core i7-10700、32GB RAM、Ubuntu 22.04 LTS）で実施したベンチマーク結果を以下に示します。全てのリクエストはHolySheep AIのAPIエンドポイント経由で処理しています。
| シナリオ | 同時接続数 | 平均レイテンシ | P99レイテンシ | エラー率 | スループット
関連リソース関連記事 |
|---|