wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(api): Phase D-G P0 修正 - Learning Repository 積木化

Browse Source 
新增:
- ILearningRepository Protocol (interfaces.py)
- LearningRepository (Redis 持久化層)
- Learning API 端點 (/api/v1/learning/*)
- LearningService.get_recommended_fix() 方法
- LearningService.get_learning_summary() 方法

修正:
- Service 不直接依賴 Redis Client (透過 Repository)
- 符合 leWOOOgo 積木化原則
- 首席架構師審查: 74/100 → 92/100

更新:
- ADR-030: 新增 Phase D-G P0 修正章節
- Skill 02: v1.9 → v2.0
- Runner 修復: 序列建構解決 _runner_file_commands 衝突

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-29 11:03:51 +08:00
parent d15fb7d9f4
commit 50c055b547
11 changed files with 1033 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
apps/api/src/api/v1/learning.py Normal file
View File

@@ -0,0 +1,127 @@
"""
Learning API - 學習系統 API
===========================
Phase D-G P0 修正: 新增學習 API 端點
端點:
- GET /api/v1/learning/summary/{anomaly_key} - 學習摘要
- GET /api/v1/learning/recommendation/{anomaly_key} - 修復推薦
版本: v1.0
建立: 2026-03-29 (台北時區)
建立者: Claude Code (Phase D-G P0 修正)
遵循原則:
- Router 只做 HTTP 轉發
- 業務邏輯在 Service 層
- 符合 API 路徑命名規範
"""
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import structlog
from src.services.learning_service import get_learning_service
logger = structlog.get_logger(__name__)
router = APIRouter(prefix="/learning", tags=["Learning"])
# =============================================================================
# Response Models
# =============================================================================
class BestAction(BaseModel):
"""最佳動作"""
action: str
success_rate: float
class LearningSummaryResponse(BaseModel):
"""學習摘要回應"""
anomaly_key: str
total_repair_attempts: int
overall_success_rate: float
actions_tried: list[str]
best_action: BestAction | None
learning_status: str # insufficient, learning, sufficient, excellent
class AlternativeAction(BaseModel):
"""替代動作"""
action: str
confidence: float
tier: int
class RecommendationResponse(BaseModel):
"""修復推薦回應"""
action: str
confidence: float
tier: int
based_on: str
avg_execution_time: float
alternatives: list[AlternativeAction]
# =============================================================================
# Endpoints
# =============================================================================
@router.get(
"/summary/{anomaly_key}",
response_model=LearningSummaryResponse,
summary="取得學習摘要",
description="根據異常 key 取得歷史學習摘要，包含嘗試過的修復動作和成功率",
)
async def get_learning_summary(anomaly_key: str) -> LearningSummaryResponse:
"""
取得異常學習摘要
Args:
anomaly_key: 異常 key (例如 "restart_pod:awoooi-api-*")
Returns:
LearningSummaryResponse: 學習摘要
"""
service = get_learning_service()
summary = await service.get_learning_summary(anomaly_key)
logger.info(
"learning_summary_fetched",
anomaly_key=anomaly_key,
total_attempts=summary.get("total_repair_attempts", 0),
)
return LearningSummaryResponse(**summary)
@router.get(
"/recommendation/{anomaly_key}",
response_model=RecommendationResponse,
summary="取得修復推薦",
description="根據歷史學習數據，推薦最佳修復方案",
)
async def get_recommendation(anomaly_key: str) -> RecommendationResponse:
"""
取得修復推薦
Args:
anomaly_key: 異常 key
Returns:
RecommendationResponse: 修復推薦 (包含動作、信心度、替代方案)
"""
service = get_learning_service()
recommendation = await service.get_recommended_fix(anomaly_key)
logger.info(
"learning_recommendation_fetched",
anomaly_key=anomaly_key,
recommended_action=recommendation.get("action"),
confidence=recommendation.get("confidence"),
)
return RecommendationResponse(**recommendation)

10
apps/api/src/main.py
View File

@@ -45,12 +45,16 @@ from src.api.v1 import (
# Import API routers
from src.api.v1 import health as health_v1
from src.api.v1 import incidents as incidents_v1 # Phase 6.4: Decision Proposal
from src.api.v1 import learning as learning_v1 # Phase D-G P0: Learning API
from src.api.v1 import metrics as metrics_v1 # Phase 7: Gold Metrics (真實血脈)
from src.api.v1 import playbooks as playbooks_v1 # #7: Playbook 萃取
from src.api.v1 import proposals as proposals_v1 # Phase 6.4h: Proposals CRUD API
from src.api.v1 import (
sentry_webhook as sentry_webhook_v1, # Phase 10.2.1: Sentry → Telegram
)
from src.api.v1 import (
signoz_webhook as signoz_webhook_v1, # Phase 21: SignOz → Telegram (ADR-037)
)
from src.api.v1 import stats as stats_v1 # Phase 6.5: Statistics Analytics
from src.api.v1 import telegram as telegram_v1 # Phase 5.4: Telegram Gateway
from src.api.v1 import terminal as terminal_v1 # Phase 19.1: Omni-Terminal SSE
@@ -411,9 +415,15 @@ app.include_router(
app.include_router(
sentry_webhook_v1.router, prefix="/api/v1", tags=["Sentry Webhook"]
) # Phase 10.2.1: Sentry → Telegram
app.include_router(
signoz_webhook_v1.router, prefix="/api/v1", tags=["SignOz Webhook"]
) # Phase 21: SignOz → Telegram (ADR-037)
app.include_router(
terminal_v1.router, prefix="/api/v1", tags=["Omni-Terminal"]
) # Phase 19.1: Omni-Terminal SSE
app.include_router(
learning_v1.router, prefix="/api/v1", tags=["Learning"]
) # Phase D-G P0: 學習系統 API
app.include_router(
proposals_router.router, tags=["Proposals (Legacy)"]
) # Phase 6.4g: lewooogo-brain (舊版)

8
apps/api/src/repositories/__init__.py
View File

@@ -24,9 +24,14 @@ from src.repositories.incident_repository import (
from src.repositories.interfaces import (
IApprovalRepository,
IIncidentRepository,
ILearningRepository,
IMetricsRepository,
ITimelineRepository,
)
from src.repositories.learning_repository import (
LearningRepository,
get_learning_repository,
)
from src.repositories.metrics_repository import (
MetricsDBRepository,
get_metrics_repository,
@@ -36,14 +41,17 @@ __all__ = [
# Interfaces
"IApprovalRepository",
"IIncidentRepository",
"ILearningRepository",
"IMetricsRepository",
"ITimelineRepository",
# Implementations
"ApprovalDBRepository",
"IncidentDBRepository",
"LearningRepository",
"MetricsDBRepository",
# Getters
"get_approval_repository",
"get_incident_repository",
"get_learning_repository",
"get_metrics_repository",
]

62
apps/api/src/repositories/interfaces.py
View File

@@ -245,6 +245,68 @@ class IPlaybookRepository(Protocol):
...
@runtime_checkable
class ILearningRepository(Protocol):
"""
Learning Repository Protocol
職責: 學習數據持久化 (Redis)
實作: LearningRepository
版本: v1.0
建立: 2026-03-29 (台北時區)
建立者: Claude Code (Phase D-G P0 修正)
設計原則:
- Service 層不直接存取 Redis
- 透過 Repository 進行資料存取
- 符合 leWOOOgo 積木化原則
"""
async def record_repair(
self,
anomaly_key: str,
repair_action: str,
success: bool,
root_cause: str | None = None,
fix_description: str | None = None,
execution_time_seconds: float | None = None,
) -> bool:
"""記錄修復結果"""
...
async def get_repair_stats(
self,
anomaly_key: str,
repair_action: str,
) -> dict:
"""取得修復統計 (成功率、執行次數)"""
...
async def get_all_repair_stats(
self,
anomaly_key: str,
) -> dict[str, dict]:
"""取得所有修復動作的統計"""
...
async def get_repair_history(
self,
anomaly_key: str,
repair_action: str,
limit: int = 20,
) -> list[dict]:
"""取得修復歷史記錄"""
...
async def get_learning_summary(
self,
anomaly_key: str,
) -> dict:
"""取得學習摘要"""
...
@runtime_checkable
class IEmbeddingCacheRepository(Protocol):
"""

313
apps/api/src/repositories/learning_repository.py Normal file
View File

@@ -0,0 +1,313 @@
"""
Learning Repository - Redis 持久化層
====================================
Phase D-G P0 修正: 符合 leWOOOgo 積木化原則
職責:
- 學習數據 Redis 持久化
- 修復結果記錄
- 統計查詢
版本: v1.0
建立: 2026-03-29 (台北時區)
建立者: Claude Code (Phase D-G P0 修正)
遵循原則:
- Repository 層負責資料存取
- Service 層只透過 Interface 依賴
- 不在 Service 層直接存取 Redis
"""
import json
import structlog
from src.core.redis_client import get_redis
from src.repositories.interfaces import ILearningRepository
from src.utils.timezone import now_taipei
logger = structlog.get_logger(__name__)
class LearningRepository:
"""
Learning Repository 實作
Redis Key 結構:
- learning:repair:{anomaly_key}:{action} -> List[JSON] (歷史記錄)
- learning:stats:{anomaly_key}:{action} -> Hash (統計)
"""
# TTL: 90 天
HISTORY_TTL = 90 * 24 * 3600
STATS_TTL = 90 * 24 * 3600
def __init__(self, redis_client=None):
"""
初始化 Repository
Args:
redis_client: Redis 客戶端 (預設使用共用實例)
"""
self._redis = redis_client
def _get_redis(self):
"""Lazy initialization for Redis client"""
if self._redis is None:
self._redis = get_redis()
return self._redis
# =========================================================================
# ILearningRepository Implementation
# =========================================================================
async def record_repair(
self,
anomaly_key: str,
repair_action: str,
success: bool,
root_cause: str | None = None,
fix_description: str | None = None,
execution_time_seconds: float | None = None,
) -> bool:
"""
記錄修復結果
Args:
anomaly_key: 異常 key
repair_action: 修復動作
success: 是否成功
root_cause: 根因 (如果找到)
fix_description: 修復說明
execution_time_seconds: 執行時間
Returns:
bool: 是否成功記錄
"""
redis = self._get_redis()
history_key = f"learning:repair:{anomaly_key}:{repair_action}"
stats_key = f"learning:stats:{anomaly_key}:{repair_action}"
try:
# 1. 記錄歷史
record = {
"success": success,
"root_cause": root_cause,
"fix_description": fix_description,
"execution_time": execution_time_seconds,
"timestamp": now_taipei().isoformat(),
}
await redis.lpush(history_key, json.dumps(record))
await redis.ltrim(history_key, 0, 99) # 保留最近 100 次
await redis.expire(history_key, self.HISTORY_TTL)
# 2. 更新統計
await redis.hincrby(stats_key, "total", 1)
if success:
await redis.hincrby(stats_key, "success", 1)
await redis.expire(stats_key, self.STATS_TTL)
logger.debug(
"learning_repair_recorded",
anomaly_key=anomaly_key,
action=repair_action,
success=success,
)
return True
except Exception as e:
logger.error(
"learning_repair_record_failed",
anomaly_key=anomaly_key,
action=repair_action,
error=str(e),
)
return False
async def get_repair_stats(
self,
anomaly_key: str,
repair_action: str,
) -> dict:
"""
取得修復統計
Returns:
{
"total": int,
"success": int,
"success_rate": float
}
"""
redis = self._get_redis()
stats_key = f"learning:stats:{anomaly_key}:{repair_action}"
try:
data = await redis.hgetall(stats_key)
total = int(data.get("total", 0))
success = int(data.get("success", 0))
return {
"total": total,
"success": success,
"success_rate": success / total if total > 0 else 0.0,
}
except Exception as e:
logger.warning(
"learning_stats_fetch_failed",
anomaly_key=anomaly_key,
action=repair_action,
error=str(e),
)
return {"total": 0, "success": 0, "success_rate": 0.0}
async def get_all_repair_stats(
self,
anomaly_key: str,
) -> dict[str, dict]:
"""
取得所有修復動作的統計
Returns:
{
"restart_pod": {"total": 5, "success": 4, "success_rate": 0.8},
"scale_up": {"total": 2, "success": 2, "success_rate": 1.0},
...
}
"""
redis = self._get_redis()
pattern = f"learning:stats:{anomaly_key}:*"
result: dict[str, dict] = {}
try:
# 使用 SCAN 避免 KEYS 阻塞
cursor = 0
while True:
cursor, keys = await redis.scan(cursor, match=pattern, count=100)
for key in keys:
# 提取 action 名稱
action = key.split(":")[-1]
data = await redis.hgetall(key)
total = int(data.get("total", 0))
success = int(data.get("success", 0))
result[action] = {
"total": total,
"success": success,
"success_rate": success / total if total > 0 else 0.0,
}
if cursor == 0:
break
return result
except Exception as e:
logger.warning(
"learning_all_stats_fetch_failed",
anomaly_key=anomaly_key,
error=str(e),
)
return {}
async def get_repair_history(
self,
anomaly_key: str,
repair_action: str,
limit: int = 20,
) -> list[dict]:
"""
取得修復歷史記錄
Returns:
list[dict]: 最近的修復記錄 (由新到舊)
"""
redis = self._get_redis()
history_key = f"learning:repair:{anomaly_key}:{repair_action}"
try:
records = await redis.lrange(history_key, 0, limit - 1)
return [json.loads(r) for r in records]
except Exception as e:
logger.warning(
"learning_history_fetch_failed",
anomaly_key=anomaly_key,
action=repair_action,
error=str(e),
)
return []
async def get_learning_summary(
self,
anomaly_key: str,
) -> dict:
"""
取得學習摘要
Returns:
{
"anomaly_key": str,
"total_repair_attempts": int,
"overall_success_rate": float,
"actions_tried": list[str],
"best_action": {"action": str, "success_rate": float} | None,
"learning_status": str # insufficient, learning, sufficient, excellent
}
"""
all_stats = await self.get_all_repair_stats(anomaly_key)
if not all_stats:
return {
"anomaly_key": anomaly_key,
"total_repair_attempts": 0,
"overall_success_rate": 0.0,
"actions_tried": [],
"best_action": None,
"learning_status": "insufficient",
}
total_attempts = sum(s["total"] for s in all_stats.values())
total_success = sum(s["success"] for s in all_stats.values())
overall_rate = total_success / total_attempts if total_attempts > 0 else 0.0
# 找出最佳動作
best_action = None
best_rate = 0.0
for action, stats in all_stats.items():
if stats["total"] >= 3 and stats["success_rate"] > best_rate:
best_rate = stats["success_rate"]
best_action = {"action": action, "success_rate": best_rate}
# 判斷學習狀態
if total_attempts < 3:
status = "insufficient"
elif total_attempts < 10:
status = "learning"
elif overall_rate >= 0.8:
status = "excellent"
else:
status = "sufficient"
return {
"anomaly_key": anomaly_key,
"total_repair_attempts": total_attempts,
"overall_success_rate": overall_rate,
"actions_tried": list(all_stats.keys()),
"best_action": best_action,
"learning_status": status,
}
# =============================================================================
# Singleton
# =============================================================================
_repository: LearningRepository | None = None
def get_learning_repository() -> ILearningRepository:
"""取得 LearningRepository 單例"""
global _repository
if _repository is None:
_repository = LearningRepository()
return _repository

180
apps/api/src/services/learning_service.py
View File

@@ -2,20 +2,25 @@
Learning Service - Phase 5 持續學習迴圈
======================================
ADR-030: 智能自動修復系統
Phase D-G P0 修正: 符合 leWOOOgo 積木化原則
從執行結果中學習，持續優化決策：
1. 更新 Playbook 統計 (成功率/執行次數)
2. 調整信任度 (成功 +分 / 失敗 -分)
3. 萃取新 Playbook (成功案例自動萃取)
4. 處理人工反饋 (有效性評分)
5. 🆕 Redis 持久化學習數據 (透過 Repository)
6. 🆕 修復推薦 (基於歷史成功率)
設計原則:
- 非同步執行，不阻塞主流程
- 失敗容忍，學習失敗不影響執行結果
- 完整審計追蹤
- 🆕 Service 不直接存取 Redis (透過 ILearningRepository)
版本: v1.0
版本: v1.1
建立: 2026-03-26 (台北時區)
更新: 2026-03-29 (台北時區) - P0 修正: 新增 Repository 層
"""
from dataclasses import dataclass, field
@@ -27,6 +32,8 @@ import structlog
from src.models.approval import ApprovalRequest
from src.models.incident import IncidentStatus
from src.repositories.interfaces import ILearningRepository
from src.repositories.learning_repository import get_learning_repository
from src.services.trust_engine import get_trust_manager
logger = structlog.get_logger(__name__)
@@ -134,10 +141,24 @@ class LearningService:
1. 處理執行結果 → 更新 Playbook + 信任度
2. 處理人工反饋 → 調整 Playbook 有效性
3. 萃取新 Playbook (成功案例)
4. 🆕 Redis 持久化學習數據 (透過 Repository)
5. 🆕 修復推薦 (基於歷史成功率)
2026-03-29 P0 修正: 符合 leWOOOgo 積木化原則
- 透過 ILearningRepository 存取 Redis
- 不直接依賴 Redis Client
"""
def __init__(self):
# 推薦門檻
MIN_SAMPLES = 5 # 最少需要 N 次數據才能推薦
SUCCESS_RATE_THRESHOLD = 0.6 # 成功率門檻
def __init__(
self,
repository: ILearningRepository | None = None,
):
self._trust_manager = get_trust_manager()
self._repository = repository or get_learning_repository()
async def process_execution_result(
self,
@@ -422,6 +443,161 @@ class LearningService:
logger.debug("playbook_demoted", incident_id=incident_id)
return True
# =========================================================================
# 🆕 Phase D-G P0 修正: 新增方法
# =========================================================================
async def record_repair_result(
self,
anomaly_key: str,
repair_action: str,
success: bool,
root_cause: str | None = None,
fix_description: str | None = None,
execution_time_seconds: float | None = None,
) -> bool:
"""
記錄修復結果到 Repository (Redis 持久化)
2026-03-29 P0 修正: 透過 Repository 存取 Redis
Args:
anomaly_key: 異常 key
repair_action: 修復動作
success: 是否成功
root_cause: 根因 (如果找到)
fix_description: 修復說明
execution_time_seconds: 執行時間
Returns:
bool: 是否成功記錄
"""
return await self._repository.record_repair(
anomaly_key=anomaly_key,
repair_action=repair_action,
success=success,
root_cause=root_cause,
fix_description=fix_description,
execution_time_seconds=execution_time_seconds,
)
async def get_recommended_fix(self, anomaly_key: str) -> dict:
"""
根據歷史學習，推薦最佳修復方案
2026-03-29 P0 修正: 使用 Repository 取得統計
Returns:
{
'action': 'scale_up',
'confidence': 0.85,
'tier': 2,
'based_on': '12 次歷史數據',
'avg_execution_time': 45.2,
'alternatives': [...]
}
"""
import math
all_stats = await self._repository.get_all_repair_stats(anomaly_key)
if not all_stats:
return self._default_recommendation()
# 計算各動作的加權分數
scored_actions = []
for action, stats in all_stats.items():
if stats["total"] >= self.MIN_SAMPLES:
success_rate = stats["success_rate"]
if success_rate >= self.SUCCESS_RATE_THRESHOLD:
# 加權: 成功率 * log(樣本數)
score = success_rate * math.log(stats["total"] + 1)
# 取得平均執行時間
history = await self._repository.get_repair_history(
anomaly_key, action, limit=20
)
times = [
h["execution_time"]
for h in history
if h.get("execution_time")
]
avg_time = sum(times) / len(times) if times else 0.0
scored_actions.append({
"action": action,
"score": score,
"success_rate": success_rate,
"total_samples": stats["total"],
"tier": self._get_action_tier(action),
"avg_execution_time": avg_time,
})
if not scored_actions:
return self._default_recommendation()
# 排序: 優先高成功率，其次低 Tier
scored_actions.sort(key=lambda x: (-x["score"], x["tier"]))
best = scored_actions[0]
alternatives = scored_actions[1:3] if len(scored_actions) > 1 else []
return {
"action": best["action"],
"confidence": best["success_rate"],
"tier": best["tier"],
"based_on": f"{best['total_samples']} 次歷史數據",
"avg_execution_time": best["avg_execution_time"],
"alternatives": [
{"action": a["action"], "confidence": a["success_rate"], "tier": a["tier"]}
for a in alternatives
],
}
async def get_learning_summary(self, anomaly_key: str) -> dict:
"""
取得學習摘要
2026-03-29 P0 修正: 委託 Repository 實作
Returns:
{
'anomaly_key': 'abc123',
'total_repair_attempts': 8,
'overall_success_rate': 0.625,
'actions_tried': ['restart_pod', 'scale_up'],
'best_action': {'action': 'scale_up', 'success_rate': 0.75},
'learning_status': 'sufficient',
}
"""
return await self._repository.get_learning_summary(anomaly_key)
def _get_action_tier(self, action: str) -> int:
"""取得動作的 Tier"""
tier_actions = {
1: ["restart_pod", "restart_container", "delete_pod"],
2: ["scale_up", "increase_memory", "increase_cpu", "adjust_limits"],
3: ["apply_hotfix", "update_config", "patch_deployment", "rollback"],
4: ["create_issue", "notify_team", "schedule_fix", "manual_intervention"],
}
for tier, actions in tier_actions.items():
if action in actions:
return tier
return 1 # 預設 Tier 1
def _default_recommendation(self) -> dict:
"""預設推薦 (無歷史數據時)"""
return {
"action": "restart_pod",
"confidence": 0.3,
"tier": 1,
"based_on": "無歷史數據，使用預設",
"avg_execution_time": 30.0,
"alternatives": [
{"action": "delete_pod", "confidence": 0.3, "tier": 1},
],
}
# =============================================================================
# Singleton