AI Agent の本番運用において、複数の LLM を効率的に调度し、失敗時のリトライ戦略を構築し、長期的な会話の文脈を管理することは、システムを安定稼働させるための三大柱です。本稿では、HolySheep AI の Agent 工程チームが本番環境で実践しているアーキテクチャと実装パターンを、比較表から具体的なコード事例まで詳細に解説します。
HolySheep vs 公式API vs 他のリレーサービスの比較
| 比較項目 | HolySheep AI | OpenAI 公式API | Anthropic 公式API | 一般的なリレーサービス |
|---|---|---|---|---|
| 為替レート | ¥1 = $1（85%節約） | ¥7.3 = $1 | ¥7.3 = $1 | ¥6.5〜7.0 = $1 |
| GPT-4.1 出力価格 | $8/MTok | $15/MTok | — | $10〜14/MTok |
| Claude Sonnet 4.5 出力 | $15/MTok | — | $18/MTok | $15〜17/MTok |
| Gemini 2.5 Flash 出力 | $2.50/MTok | — | — | $2〜3/MTok |
| DeepSeek V3.2 出力 | $0.42/MTok | — | — | $0.50〜0.80/MTok |
| レイテンシ | <50ms | 80〜200ms | 100〜250ms | 60〜150ms |
| 支払い方法 | WeChat Pay / Alipay / USDT | 国際クレジットカード | 国際クレジットカード | 限定的 |
| 無料クレジット | 登録で即時付与 | $5〜18相当 | $5相当 | 无几或无 |
| マルチLLM統一エンドポイント | 対応 | OpenAI専用品 | Anthropic専用品 | 対応している場合も |
向いている人・向いていない人
✅ HolySheep AI が向いている人
- コスト最適化を重視する開発チーム — 公式API比85%のコスト削減は、本番運用の月額コストに大きな影響を与えます
- 中国本土開発者・中国企业 — WeChat Pay / Alipay による地元決済手段で、素早く導入を開始できます
- マルチLLMアーキテクチャを採用しているプロジェクト — 単一エンドポイントで複数のモデルを試行錯誤でき、プロンプトエンジニアリング効率が向上します
- 低レイテンシが求められるリアルタイム Agent — <50msのオーバーヘッドは、ユーザーの体感品質に直結します
- DeepSeek V3.2 を低コストで活用したい人 — $0.42/MTok という破格の価格は、大量処理ワークロードに最適です
❌ HolySheep AI が向いていない人
- 公式APIの保証されたSLA・法的対応が必要不可欠なEnterprise向け — ただしHolySheepも十分な可用性を提供しており、多くのユースケースで対応可能です
- 日本円の請求書を法人経費として処理する必要があるケース — 現状はUSD建てメインのため、経費精算フローに組み込む際に追加検討が必要です
- 特定のモデル（例：o1, o3）のみを仕様書で義務付けられている場合 — 対応モデルの最新版確認を事前に行ってください
価格とROI
HolySheep AI の料金体系は、開発者にとって非常に透明で計算しやすい設計になっています。以下に代表的なモデルのコスト比較を示します。
| モデル | 公式価格 ($/MTok出力) | HolySheep ($/MTok出力) | 1MTokあたりの節約額 | 100MTok/月運用時の月間節約 |
|---|---|---|---|---|
| GPT-4.1 | $15.00 | $8.00 | $7.00（47%OFF） | 約$700 |
| Claude Sonnet 4.5 | $18.00 | $15.00 | $3.00（17%OFF） | 約$300 |
| Gemini 2.5 Flash | $2.50 | $2.50 | 同等 | — |
| DeepSeek V3.2 | $1.10（推定） | $0.42 | $0.68（62%OFF） | 約$68 |
ROI計算の例：
私自身のプロジェクトでは、GPT-4.1 を月間500MTok程度使用するAgentアプリケーションを運用しています。公式APIでは月間で$7,500のところ、HolySheepでは$4,000に抑えられます。年間では約$42,000の節約になり、これは開発リソースへの追加投資に十分充てられる金額です。
多LLM并发调度の实战アーキテクチャ
HolySheep Agent 工程チームが実践している并发调度のパターンについて、Pythonでの実装例を見ていきましょう。
1. 基本の并发リクエスト実装
import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
import time
HolySheep API設定
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class MultiLLMDispatcher:
"""複数のLLMへの并发リクエストを効率的に處理するクラス"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def _make_request(
self,
session: aiohttp.ClientSession,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 1024
) -> Dict[str, Any]:
"""单个LLMリクエストを送信"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.time()
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
result = await response.json()
latency = (time.time() - start_time) * 1000 # ミリ秒に変換
return {
"model": model,
"status_code": response.status,
"latency_ms": round(latency, 2),
"response": result,
"error": None if response.status == 200 else result.get("error", {})
}
async def concurrent_chat(
self,
messages: List[Dict],
models: List[str],
temperature: float = 0.7,
max_tokens: int = 1024
) -> List[Dict[str, Any]]:
"""複数のLLMに并发でリクエストを送信し、最速の回答を返す"""
async with aiohttp.ClientSession() as session:
tasks = [
self._make_request(session, model, messages, temperature, max_tokens)
for model in models
]
# 全リクエストを并发実行
results = await asyncio.gather(*tasks, return_exceptions=True)
# 結果を整理
valid_results = [r for r in results if isinstance(r, dict) and r.get("status_code") == 200]
return valid_results
async def fastest_response(
self,
messages: List[Dict],
models: List[str],
timeout: float = 10.0
) -> Optional[Dict[str, Any]]:
"""最快レスポンスを返す（ Fail-Fast 戦略）"""
async with aiohttp.ClientSession() as session:
tasks = [
self._make_request(session, model, messages)
for model in models
]
done, pending = await asyncio.wait(
tasks,
timeout=timeout,
return_when=asyncio.FIRST_COMPLETED
)
# 完了したタスクの結果を返す
for task in done:
try:
result = task.result()
if result.get("status_code") == 200:
# 残りのリクエストをキャンセル
for p in pending:
p.cancel()
return result
except Exception as e:
continue
# 全て失敗した場合
return None
使用例
async def main():
dispatcher = MultiLLMDispatcher(api_key=HOLYSHEEP_API_KEY)
messages = [
{"role": "system", "content": "你是日本市場の専門家です。简潔に回答してください。"},
{"role": "user", "content": "2026年のAI市場動向について3点で教えて"}
]
# 例1: 全モデル并发呼び出し
results = await dispatcher.concurrent_chat(
messages=messages,
models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
)
print("=== 并发调用结果 ===")
for r in results:
print(f"モデル: {r['model']}, レイテンシ: {r['latency_ms']}ms")
if "error" not in r or not r["error"]:
content = r["response"].get("choices", [{}])[0].get("message", {}).get("content", "")
print(f"回答: {content[:100]}...")
# 例2: 最速レスポンスを返す
fastest = await dispatcher.fastest_response(
messages=messages,
models=["gemini-2.5-flash", "deepseek-v3.2"] # 高速モデル优先
)
if fastest:
print(f"\n最速回答: {fastest['model']} ({fastest['latency_ms']}ms)")
if __name__ == "__main__":
asyncio.run(main())
2. インテリジェントなモデル选择とフォールバック
import asyncio
import aiohttp
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TaskPriority(Enum):
"""タスクの優先度级别"""
HIGH = "high" # 正確に执行力が必要（例：医疗、金融判断）
MEDIUM = "medium" # バランス型（一般的なAgentタスク）
LOW = "low" # コスト最优先（大量処理・试行錯誤）
@dataclass
class ModelConfig:
"""モデル設定"""
name: str
cost_per_mtok: float # $ per million tokens output
avg_latency_ms: float
capabilities: list[str]
priority: TaskPriority
モデルコスト設定（2026年5月時点）
MODEL_CONFIGS = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
cost_per_mtok=8.0,
avg_latency_ms=120,
capabilities=["reasoning", "coding", "analysis"],
priority=TaskPriority.HIGH
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
cost_per_mtok=15.0,
avg_latency_ms=150,
capabilities=["reasoning", "writing", "analysis"],
priority=TaskPriority.HIGH
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
cost_per_mtok=2.50,
avg_latency_ms=45,
capabilities=["fast", "multimodal", "coding"],
priority=TaskPriority.MEDIUM
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
cost_per_mtok=0.42,
avg_latency_ms=50,
capabilities=["reasoning", "coding", "cost-efficient"],
priority=TaskPriority.LOW
)
}
class IntelligentModelSelector:
"""タスク性质に応じて最適なモデルを選択する"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.retry_history: dict[str, list] = {}
def select_model(
self,
task_type: str,
priority: TaskPriority = TaskPriority.MEDIUM,
budget_constraint: Optional[float] = None
) -> str:
"""タスク类型に基づいてモデルを選択"""
# 能力マッピング
capability_map = {
"coding": ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"],
"analysis": ["gpt-4.1", "claude-sonnet-4.5"],
"reasoning": ["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"],
"fast_response": ["gemini-2.5-flash", "deepseek-v3.2"],
"creative": ["gpt-4.1", "claude-sonnet-4.5"],
"cost_efficient": ["deepseek-v3.2", "gemini-2.5-flash"]
}
candidates = capability_map.get(task_type, list(MODEL_CONFIGS.keys()))
# 優先度によるフィルタリング
if priority == TaskPriority.HIGH:
candidates = [m for m in candidates
if MODEL_CONFIGS[m].priority == TaskPriority.HIGH]
if not candidates:
candidates = ["gpt-4.1"] # フォールバック
# 予算制約の適用
if budget_constraint:
candidates = [m for m in candidates
if MODEL_CONFIGS[m].cost_per_mtok <= budget_constraint]
if not candidates:
# フォールバック: コスト効率のいいモデル
candidates = ["deepseek-v3.2"]
# 最低コストのモデルを選択
selected = min(candidates, key=lambda m: MODEL_CONFIGS[m].cost_per_mtok)
logger.info(f"Selected model: {selected} for task: {task_type}")
return selected
async def smart_request(
self,
messages: list,
task_type: str,
priority: TaskPriority = TaskPriority.MEDIUM,
max_retries: int = 3
) -> Optional[dict]:
"""スマートなリクエストを実行、自动フォールバック付き"""
model = self.select_model(task_type, priority)
for attempt in range(max_retries):
try:
result = await self._execute_request(model, messages)
if result.get("success"):
return result
# 失敗時: より高性能なモデルにフォールバック
if attempt < max_retries - 1:
fallback_model = self._get_fallback_model(model)
if fallback_model:
logger.warning(f"Attempt {attempt+1} failed, falling back to {fallback_model}")
model = fallback_model
except Exception as e:
logger.error(f"Request error on attempt {attempt+1}: {e}")
if attempt == max_retries - 1:
return {"success": False, "error": str(e)}
return {"success": False, "error": "Max retries exceeded"}
async def _execute_request(
self,
model: str,
messages: list
) -> dict:
"""单个リクエストを実行"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
return {
"success": True,
"model": model,
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {})
}
else:
error_data = await response.json()
return {
"success": False,
"model": model,
"error": error_data.get("error", {}).get("message", "Unknown error")
}
def _get_fallback_model(self, current_model: str) -> Optional[str]:
"""フォールバック先のモデルを取得"""
fallback_chain = {
"deepseek-v3.2": "gemini-2.5-flash",
"gemini-2.5-flash": "gpt-4.1",
"gpt-4.1": "claude-sonnet-4.5",
"claude-sonnet-4.5": None # 最終フォールバックなし
}
return fallback_chain.get(current_model)
使用例
async def main():
selector = IntelligentModelSelector(api_key="YOUR_HOLYSHEEP_API_KEY")
# タスク别にモデルを選択
tasks = [
("coding", TaskPriority.HIGH),
("fast_response", TaskPriority.LOW),
("analysis", TaskPriority.MEDIUM),
]
messages = [
{"role": "user", "content": "Pythonでクイックソートを実装してください"}
]
for task_type, priority in tasks:
model = selector.select_model(task_type, priority)
config = MODEL_CONFIGS[model]
print(f"タスク: {task_type} / 優先度: {priority.value}")
print(f" 選択モデル: {model}")
print(f" コスト: ${config.cost_per_mtok}/MTok")
print(f" 예상レイテンシ: {config.avg_latency_ms}ms")
print()
if __name__ == "__main__":
asyncio.run(main())
リトライ戦略の最佳プラクティス
Agent工程チームのリトライ戦略は、単純な指数バックオフだけでなく、エラー種别に応じた適応的なアプローチを採用しています。
import asyncio
import aiohttp
from typing import Callable, Any, Optional
from dataclasses import dataclass
from enum import Enum
import random
import time
class ErrorType(Enum):
"""エラー类型の分類"""
RATE_LIMIT = "rate_limit" # 429 Rate Limit
SERVER_ERROR = "server_error" # 5xx エラー
TIMEOUT = "timeout" # タイムアウト
VALIDATION_ERROR = "validation" # 400 Bad Request
AUTH_ERROR = "auth" # 401/403
UNKNOWN = "unknown"
@dataclass
class RetryConfig:
"""リトライ設定"""
max_retries: int = 3
base_delay: float = 1.0 # 基础延迟（秒）
max_delay: float = 60.0 # 最大延迟
exponential_base: float = 2.0 # 指数バックオフの底
jitter: bool = True # ランダムジャダー
# エラー类型別の特殊設定
rate_limit_max_retries: int = 5
rate_limit_base_delay: float = 2.0
class AdaptiveRetryHandler:
"""適応的なリトライハンドラ — エラー类型に応じて戦略を変更"""
def __init__(self, config: Optional[RetryConfig] = None):
self.config = config or RetryConfig()
self.error_counts: dict[str, int] = {}
def classify_error(
self,
status_code: int,
response_data: Optional[dict] = None
) -> ErrorType:
"""HTTPステータスとレスポンスからエラー类型を分類"""
if status_code == 429:
return ErrorType.RATE_LIMIT
elif 500 <= status_code < 600:
return ErrorType.SERVER_ERROR
elif status_code == 400:
return ErrorType.VALIDATION_ERROR
elif status_code in (401, 403):
return ErrorType.AUTH_ERROR
elif response_data and "error" in response_data:
error_msg = str(response_data["error"]).lower()
if "timeout" in error_msg or "timed out" in error_msg:
return ErrorType.TIMEOUT
return ErrorType.UNKNOWN
def calculate_delay(
self,
error_type: ErrorType,
attempt: int
) -> float:
"""エラー类型と試行回数に基づいて延迟時間を計算"""
if error_type == ErrorType.RATE_LIMIT:
base_delay = self.config.rate_limit_base_delay
# Retry-After ヘッダーがあればそちらを優先
max_retries = self.config.rate_limit_max_retries
else:
base_delay = self.config.base_delay
max_retries = self.config.max_retries
# 指数バックオフの計算
delay = base_delay * (self.config.exponential_base ** attempt)
delay = min(delay, self.config.max_delay)
# ジャダーの追加（ブレ感を出す）
if self.config.jitter:
jitter_range = delay * 0.1
delay += random.uniform(-jitter_range, jitter_range)
return max(0.1, delay)
async def execute_with_retry(
self,
request_func: Callable,
*args,
**kwargs
) -> dict:
"""リトライ逻辑を組み込んだリクエスト実行"""
last_error = None
last_error_type = ErrorType.UNKNOWN
# エラー类型に応じた最大リトライ数を決定
for attempt in range(self.config.max_retries + 1):
try:
result = await request_func(*args, **kwargs)
# 成功チェック
if isinstance(result, dict):
status_code = result.get("status_code", 200)
if status_code == 200:
return {"success": True, "data": result}
error_type = self.classify_error(status_code, result)
# リトライ不要のエラー
if error_type in (ErrorType.VALIDATION_ERROR, ErrorType.AUTH_ERROR):
return {"success": False, "error": result, "error_type": error_type}
last_error_type = error_type
last_error = result
except asyncio.TimeoutError:
last_error_type = ErrorType.TIMEOUT
last_error = {"message": "Request timeout"}
except Exception as e:
last_error_type = ErrorType.UNKNOWN
last_error = {"message": str(e)}
# リトライ判定
if attempt < self.config.max_retries:
delay = self.calculate_delay(last_error_type, attempt)
print(f"[Retry] Attempt {attempt+1} failed ({last_error_type.value}). "
f"Waiting {delay:.2f}s before retry...")
await asyncio.sleep(delay)
return {
"success": False,
"error": last_error,
"error_type": last_error_type.value,
"attempts": attempt + 1
}
def get_error_summary(self) -> dict:
"""エラー統計のサマリーを返す"""
return dict(self.error_counts)
async function wrapper for retry
async def wrapped_api_call(
api_key: str,
model: str,
messages: list,
handler: AdaptiveRetryHandler
) -> dict:
"""HolySheep APIを呼び出すラッパー関数"""
async def _call():
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages
}
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
data = await response.json()
return {
"status_code": response.status,
"data": data
}
result = await handler.execute_with_retry(_call)
return result
使用例
async def main():
config = RetryConfig(
max_retries=3,
base_delay=1.0,
max_delay=30.0,
rate_limit_max_retries=5,
rate_limit_base_delay=2.0
)
handler = AdaptiveRetryHandler(config)
messages = [
{"role": "user", "content": "Hello, world!"}
]
result = await wrapped_api_call(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2",
messages=messages,
handler=handler
)
if result["success"]:
print("Request successful!")
print(f"Response: {result['data']}")
else:
print(f"Request failed after retries:")
print(f" Error Type: {result.get('error_type')}")
print(f" Attempts: {result.get('attempts')}")
if __name__ == "__main__":
asyncio.run(main())
文脈管理と長期メモリ戦略
Agentが複数ターンにわたる会話で文脈を維持するための戦略を実装します。
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum
import json
import tiktoken # トークン数を正確にカウント
from collections import deque
class SummaryStrategy(Enum):
"""要約戦略の类型"""
NONE = "none" # 切捨てのみ
TRUNCATE = "truncate" # 古いメッセージから切捨て
SUMMARIZE = "summarize" # LLMで要約
HYBRID = "hybrid" # 重要度スコア付きハイブリッド
@dataclass
class Message:
"""メッセージオブジェクト"""
role: str
content: str
metadata: dict = field(default_factory=dict)
importance: float = 1.0 # 0.0〜1.0 の重要度スコア
@dataclass
class ContextWindow:
"""コンテキストウィンドウ管理"""
max_tokens: int = 128000 # コンテキスト上限
reserved_tokens: int = 2000 # システムプロンプト用に予約
min_messages: int = 2 # 最小保持メッセージ数
def effective_limit(self) -> int:
"""実際に使用可能なトークン数"""
return self.max_tokens - self.reserved_tokens
class ConversationContext:
"""会話文脈を管理するクラス"""
def __init__(
self,
api_key: str,
max_tokens: int = 128000,
summary_strategy: SummaryStrategy = SummaryStrategy.HYBRID
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.context_window = ContextWindow(max_tokens=max_tokens)
self.summary_strategy = summary_strategy
# メッセージ履歴
self.messages: deque = deque()
# システムプロンプト
self.system_prompt: Optional[str] = None
# エンコーダー（トークンカウント用）
try:
self.encoding = tiktoken.get_encoding("cl100k_base")
except:
self.encoding = None # フォールバック
def set_system_prompt(self, prompt: str):
"""システムプロンプトを設定"""
self.system_prompt = prompt
# システムプロンプト変更時にコンテキストを調整
self._ensure_within_limit()
def add_message(
self,
role: str,
content: str,
importance: float = 1.0,
metadata: Optional[dict] = None
):
"""新しいメッセージを追加"""
message = Message(
role=role,
content=content,
importance=importance,
metadata=metadata or {}
)
self.messages.append(message)
self._ensure_within_limit()
def count_tokens(self, text: str) -> int:
"""トークン数をカウント"""
if self.encoding:
return len(self.encoding.encode(text))
# フォールバック: приблизительная 計算
return len(text) // 4
def get_context_size(self) -> int:
"""現在のコンテキストサイズ（トークン数）を取得"""
total = 0
if self.system_prompt:
total += self.count_tokens(self.system_prompt)
for msg in self.messages:
# role + content + overhead
total += self.count_tokens(msg.content) + 10
return total
def _ensure_within_limit(self):
"""コンテキストウィンドウ内に収めるよう調整"""
limit = self.context_window.effective_limit()
while self.get_context_size() > limit and len(self.messages) > self.context_window.min_messages:
if self.summary_strategy == SummaryStrategy.TRUNCATE:
self._truncate_oldest()
elif self.summary_strategy == SummaryStrategy.HYBRID:
self._hybrid_cleanup()
else:
self._truncate_oldest()
def _truncate_oldest(self):
"""最も古いメッセージを削除"""
if self.messages:
self.messages.popleft()
def _hybrid_cleanup(self):
"""重要度ベースのハイブリッドクリーンアップ"""
if not self.messages:
return
# 重要度の低いメッセージを優先的に削除
messages_list = list(self.messages)
# システムメッセージは保持
system_messages = [m for m in messages_list if m.role == "system"]
other_messages = [m for m in messages_list if m.role != "system"]
# 重要度でソートして削除
other_messages.sort(key=lambda m: m.importance)
# 半分を削除（重要度の低い方から）
remove_count = len(other_messages) // 2
other_messages = other_messages[remove_count:]
self.messages = deque(system_messages + other_messages)
def _summarize_old_messages(self, model: str = "deepseek-v3.2") -> Optional[str]:
"""古いメッセージをLLMで要約"""
if len(self.messages) < 4:
return None
# 半分より古いメッセージを抽出
messages_list = list(self.messages)
midpoint = len(messages_list) // 2
old_messages = messages_list[:midpoint]
new_messages = messages_list[midpoint:]
# 要約プロンプトを構築
summary_prompt = f"""以下の会話の歴史を简潔に要約してください。
重要な情報・决定・文脈を保持し、冗長な对话は省略します。
要約対象:
{chr(10).join([f'{m.role}: {m.content}' for m in old_messages])}
简潔な要約:"""
# LLMで要約（同期的に実行）
import aiohttp
import asyncio
async def _summarize():
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "user", "content": summary_prompt}
],
"max_tokens": 500,
"temperature": 0.3
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
data = await response.json()
return data["choices"][0]["message"]["content"]
return None
summary = asyncio.run(_summarize())
if summary:
# 要約を先頭に追加
self.messages = deque(new_messages)
self.messages.appendleft(Message