PHASE11_QUICK_REFERENCE.md

Phase 11 クイックリファレンス

Dynamic Model & Database Switching - API ガイド

---

🚀 クイックスタート

インポート

# モデルマネージャー
from phase1_skeleton.model_manager import (
    DynamicModelManager, ModelConfig, ModelSource, ModelStatus,
    create_model_manager
)

# データベースマネージャー
from phase1_skeleton.db_manager import (
    DynamicDatabaseManager, DatabaseConfig, DatabaseBackend,
    create_database_manager
)

# ランタイム調整層
from phase1_skeleton.runtime_coordinator import (
    RuntimeCoordinator, SwitchingStrategy,
    create_runtime_coordinator
)

初期化

# マネージャー作成
model_mgr = create_model_manager(cache_size=3)
db_mgr = create_database_manager()
coordinator = create_runtime_coordinator(model_mgr, db_mgr)

---

📋 DynamicModelManager API

モデル登録

config = ModelConfig(
    name="deepseek_main",
    model_id="mlx-community/DeepSeek-R1-Distill-Qwen-7B-4bit",
    source=ModelSource.HUGGINGFACE,  # or LOCAL_PATH, OLLAMA, ANTHROPIC
    device="auto",  # or "cpu", "gpu"
    quantization="4-bit",  # optional
    temperature=0.7,
    max_tokens=512,
    constraints_enabled=True,
    metadata={"version": "1.0"}
)

manager.register_model(config)

モデルロード

# ロード
success = await manager.load_model("deepseek_main")

# ロード状態確認
instance = manager.get_model("deepseek_main")
print(instance.status)  # ModelStatus.READY
print(instance.metrics.load_time_ms)  # 1234.5

モデル切り替え

# 切り替え
success = await manager.switch_model("deepseek_shadow")

# 現在のモデル確認
current = manager.get_current_model()
print(current.config.name)  # "deepseek_shadow"

メトリクス取得

# 全モデル一覧
models = manager.list_available_models()  # ["deepseek_main", "deepseek_shadow"]
loaded = manager.list_loaded_models()     # ["deepseek_shadow"]

# メトリクス
metrics = manager.get_metrics("deepseek_shadow")
print(metrics["avg_latency_ms"])  # 123.45
print(metrics["throughput_tps"])  # 456.78
print(metrics["success_rate"])    # 99.5

# 全体ステータス
status = manager.get_status()
print(status["current_model"])
print(status["cache_stats"])

アンロード

success = await manager.unload_model("deepseek_main")

MCP イベント処理

result = await manager.handle_mcp_model_switch_event(
    model_name="deepseek_shadow",
    metadata={
        "source": "user_request",
        "priority": "high",
        "reason": "performance_optimization"
    }
)

print(result["success"])  # True/False
print(result["model"])    # "deepseek_shadow"
print(result["status"])   # Full status dict

---

📦 DynamicDatabaseManager API

バックエンド登録

# SQLite
sqlite_config = DatabaseConfig(
    backend=DatabaseBackend.SQLITE,
    path="./memory.db",
    connection_pool_size=10,
    timeout_seconds=30
)
db_mgr.register_backend(sqlite_config, name="sqlite_main")

# ChromaDB
chroma_config = DatabaseConfig(
    backend=DatabaseBackend.CHROMA,
    path="./chroma_db",
    metadata={"embedding_model": "all-MiniLM-L6-v2"}
)
db_mgr.register_backend(chroma_config, name="chroma_main")

接続/切断

# 接続
success = await db_mgr.connect("sqlite_main")

# 切断
success = await db_mgr.disconnect("sqlite_main")

# 再接続
success = await db_mgr.connect("sqlite_main")

バックエンド切り替え

# 切り替え (自動データ移行)
success = await db_mgr.switch_backend("chroma_main")

# 現在のバックエンド確認
current = db_mgr.get_current_backend()
print(current)  # "chroma_main"

メトリクス取得

# 登録済みバックエンド一覧
backends = db_mgr.list_backends()  # ["sqlite_main", "chroma_main"]

# メトリクス
status = db_mgr.get_status()
print(status["current_backend"])
print(status["backends"])  # Dict with all backend details

# 特定バックエンドの詳細
instance = db_mgr.instances["sqlite_main"]
print(instance.metrics.to_dict())
# {
#     "total_queries": 100,
#     "avg_query_time_ms": 45.23,
#     "success_rate": 99.0,
#     ...
# }

MCP イベント処理

result = await db_mgr.handle_mcp_db_switch_event(
    backend_name="chroma_main",
    metadata={
        "reason": "performance_upgrade",
        "expected_duration_ms": 5000
    }
)

print(result["success"])
print(result["backend"])

---

🎛️ RuntimeCoordinator API

リクエスト作成

request = await coordinator.submit_switch_request(
    switch_model="deepseek_shadow",           # optional
    switch_backend="chroma_main",             # optional
    strategy=SwitchingStrategy.GRACEFUL,      # IMMEDIATE, GRACEFUL, SCHEDULED
    timeout_seconds=300,
    auto_rollback=True,
    metadata={"initiator": "system"}
)

print(request.request_id)  # "switch_a1b2c3d4"

切り替え実行

result = await coordinator.execute_switch(request)

# 結果確認
print(result.success)                 # True/False
print(result.switching_state)         # SwitchingState.ACTIVE
print(result.switch_time_ms)          # 1234.5
print(result.model_switched)          # True/False
print(result.backend_switched)        # True/False
print(result.rollback_performed)      # True/False
print(result.error_message)           # Error message if failed

# 結果を辞書で取得
result_dict = result.to_dict()

ステータス監視

# 現在のステータス
status = coordinator.get_status()
print(status["switching_state"])         # "idle" / "preparing" / "switching" / etc
print(status["request_queue"])           # Queue info
print(status["model_status"])            # Model manager status
print(status["database_status"])         # Database manager status
print(status["health"])                  # Health check results
print(status["switch_history_length"])   # Number of past switches

# リクエストキューの詳細
queue_status = status["request_queue"]
print(queue_status["in_flight_count"])
print(queue_status["queued_size"])
print(queue_status["completed_count"])

履歴確認

# 最近の切り替え操作
history = coordinator.get_switch_history(limit=20)
for switch in history:
    print(f"{switch['request_id']}: {switch['success']}")
    print(f"  Model: {switch['previous_model']} → {switch['new_model']}")
    print(f"  DB: {switch['previous_backend']} → {switch['new_backend']}")
    print(f"  Time: {switch['switch_time_ms']}ms")

MCP イベント処理

result = await coordinator.handle_mcp_runtime_switch_event(
    event_type="model_and_db_switch",
    switch_model="deepseek_main",
    switch_backend="sqlite_main",
    metadata={
        "source": "mcp",
        "priority": "high"
    }
)

# resultはSwitchingResult.to_dict()と同じ
print(result["success"])
print(result["switching_state"])

---

🔍 パターン集

パターン1: 安全な切り替え (推奨)

# グレースフルな切り替え + 自動ロールバック
request = await coordinator.submit_switch_request(
    switch_model="new_model",
    strategy=SwitchingStrategy.GRACEFUL,  # リクエスト完了待機
    auto_rollback=True,                   # 自動ロールバック有効
    timeout_seconds=300
)

result = await coordinator.execute_switch(request)

if result.success:
    logger.info(f"✓ Switch successful ({result.switch_time_ms:.0f}ms)")
else:
    logger.error(f"✗ Switch failed: {result.error_message}")
    if result.rollback_performed:
        logger.info("  Auto-rollback completed")

パターン2: 複数コンポーネントの同時切り替え

# モデルとDB両方を切り替え
request = await coordinator.submit_switch_request(
    switch_model="deepseek_shadow",
    switch_backend="chroma_primary",
    strategy=SwitchingStrategy.GRACEFUL
)

result = await coordinator.execute_switch(request)

if result.success:
    print(f"✓ Model: {result.previous_model} → {result.new_model}")
    print(f"✓ DB: {result.previous_backend} → {result.new_backend}")
    print(f"✓ Total time: {result.switch_time_ms:.0f}ms")

パターン3: パフォーマンス改善による自動切り替え

async def auto_optimize_on_low_throughput():
    """スループット低下時に自動最適化"""

    model = manager.get_current_model()
    if model:
        throughput = model.metrics.throughput

        if throughput < 100:  # tokens/sec
            logger.warning(f"Low throughput detected: {throughput}")

            # 別のモデルに切り替え
            result = await coordinator.handle_mcp_runtime_switch_event(
                event_type="performance_optimization",
                switch_model="deepseek_main",
                metadata={"reason": "low_throughput", "threshold": 100}
            )

            if result["success"]:
                logger.info("✓ Switched to better model")

パターン4: スケジュール済み切り替え

import asyncio
from datetime import datetime, timedelta

async def scheduled_maintenance_switch():
    """定期メンテナンス時の切り替え"""

    # 次のメンテナンス時刻まで待機
    maintenance_time = datetime.now() + timedelta(hours=2)
    sleep_duration = (maintenance_time - datetime.now()).total_seconds()

    await asyncio.sleep(sleep_duration)

    # 切り替え実行
    request = await coordinator.submit_switch_request(
        switch_model="deepseek_shadow",
        switch_backend="sqlite_main",
        strategy=SwitchingStrategy.SCHEDULED
    )

    result = await coordinator.execute_switch(request)

パターン5: ヘルスモニタリング

async def monitor_system_health():
    """システムヘルスの継続的監視"""

    while True:
        status = coordinator.get_status()
        health = status["health"]

        if health["latest"]["overall_health"]:
            logger.debug("✓ System healthy")
        else:
            logger.warning("✗ System unhealthy")

            # 必要に応じてロールバック
            if health["latest"]["model_health"] == False:
                await coordinator.model_manager.switch_model(
                    coordinator.model_manager.list_available_models()[0]
                )

        await asyncio.sleep(30)  # 30秒ごとにチェック

---

⚙️ 設定リファレンス

ModelConfig パラメータ

パラメータ	型	デフォルト	説明
name	str	-	モデル識別子
model_id	str	-	モデルID (HF/ローカルパス)
source	ModelSource	HUGGINGFACE	ソース指定
device	str	"auto"	デバイス指定
quantization	str	None	量子化フォーマット
temperature	float	0.7	サンプリング温度
max_tokens	int	512	最大トークン数
constraints_enabled	bool	True	制約有効化
metadata	Dict	{}	カスタムメタデータ

DatabaseConfig パラメータ

パラメータ	型	デフォルト	説明
backend	DatabaseBackend	-	DB バックエンド
host	str	"localhost"	ホスト名
port	int	5432	ポート番号
database	str	"refactorium"	DB名
username	str	None	ユーザー名
password	str	None	パスワード
path	str	None	ファイルパス (SQLite用)
connection_pool_size	int	10	コネクション数
timeout_seconds	int	30	タイムアウト
metadata	Dict	{}	カスタムメタデータ

SwitchingRequest パラメータ

パラメータ	型	デフォルト	説明
switch_model	str	None	切り替え先モデル
switch_backend	str	None	切り替え先DB
strategy	SwitchingStrategy	GRACEFUL	切り替え戦略
timeout_seconds	int	300	タイムアウト
verify_health	bool	True	ヘルスチェック実行
auto_rollback_on_error	bool	True	自動ロールバック
metadata	Dict	{}	カスタムメタデータ

---

🐛 トラブルシューティング

モデルロード失敗

# 原因確認
instance = manager.get_model("model_name")
if instance.status == ModelStatus.ERROR:
    print("Model loading failed")

# 解決策
# 1. モデルIDを確認
# 2. ディスク容量を確認
# 3. メモリ不足をチェック
# 4. キャッシュをクリア
manager.cache.remove("model_name")

データベース接続失敗

# ステータス確認
instance = db_mgr.instances["backend_name"]
print(instance.status)  # DISCONNECTED / ERROR

# メトリクス確認
print(instance.metrics.to_dict())

# 再接続試行
await db_mgr.disconnect("backend_name")
await db_mgr.connect("backend_name")

切り替えタイムアウト

# 原因
# - 大量のインフライトリクエスト
# - ネットワーク遅延
# - リソース不足

# 解決策
# 1. タイムアウト値を増やす
request = await coordinator.submit_switch_request(
    ...
    timeout_seconds=600  # 10分に延長
)

# 2. 即座切り替えに変更
request = await coordinator.submit_switch_request(
    ...
    strategy=SwitchingStrategy.IMMEDIATE  # グレースフルを使わない
)

ロールバック失敗

# ログを確認
history = coordinator.get_switch_history(limit=1)
last_switch = history[0]
print(last_switch["error"])

# 手動復旧
if last_switch["previous_model"]:
    await manager.switch_model(last_switch["previous_model"])
if last_switch["previous_backend"]:
    await db_mgr.switch_backend(last_switch["previous_backend"])

---

📊 パフォーマンス最適化

キャッシュサイズ最適化

# メモリ使用量が多い場合
manager = create_model_manager(cache_size=1)  # デフォルト: 3

# メモリ余裕がある場合
manager = create_model_manager(cache_size=5)

リクエストキュー最適化

# キュー最大サイズ変更
coordinator.request_queue = RequestQueue(max_size=500)  # デフォルト: 1000

タイムアウト値の調整

# 遅いネットワーク環境
request = await coordinator.submit_switch_request(
    ...
    timeout_seconds=600  # 10分
)

# 高速環境
request = await coordinator.submit_switch_request(
    ...
    timeout_seconds=60   # 1分
)

---

📚 関連ドキュメント

• PHASE11_DYNAMIC_SWITCHING_COMPLETE.md - 完全な技術仕様
• PHASE10b_MCP_BODY_INTERFACE_COMPLETE.md - MCP統合
• system.yaml - システム設定ファイル

---

最終更新: 2025-12-14 バージョン: 1.0.0