391 lines
13 KiB
Python
391 lines
13 KiB
Python
"""PitfallDetector - 任务避坑预警
|
||
|
||
新任务启动时检索历史失败经验，匹配当前计划步骤，自动预警。
|
||
基于 ExperienceStore 中存储的失败经验，将失败步骤与当前计划步骤
|
||
进行关键词匹配，计算失败率并按严重程度返回预警列表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Any, Protocol
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class WarningLevel(str, Enum):
|
||
"""预警级别"""
|
||
|
||
HIGH = "high"
|
||
MEDIUM = "medium"
|
||
LOW = "low"
|
||
|
||
|
||
@dataclass
|
||
class PitfallWarning:
|
||
"""避坑预警
|
||
|
||
Attributes:
|
||
step_name: 计划步骤名称
|
||
warning_level: 预警级别（HIGH/MEDIUM/LOW）
|
||
failure_rate: 历史失败率（0.0 ~ 1.0）
|
||
historical_failures: 历史失败原因列表
|
||
suggestion: 优化建议
|
||
"""
|
||
|
||
step_name: str
|
||
warning_level: WarningLevel
|
||
failure_rate: float
|
||
historical_failures: list[str] = field(default_factory=list)
|
||
suggestion: str = ""
|
||
|
||
|
||
class ExperienceStoreProtocol(Protocol):
|
||
"""ExperienceStore 协议接口，用于类型标注"""
|
||
|
||
async def search(
|
||
self,
|
||
query: str,
|
||
top_k: int = 5,
|
||
task_type: str | None = None,
|
||
search_multiplier: int = 5,
|
||
) -> list[Any]:
|
||
...
|
||
|
||
|
||
# 预警级别阈值
|
||
_HIGH_THRESHOLD = 0.5
|
||
_MEDIUM_THRESHOLD = 0.2
|
||
|
||
|
||
class PitfallDetector:
|
||
"""避坑检测器
|
||
|
||
新任务启动时检索历史失败经验，匹配当前计划步骤，自动预警。
|
||
|
||
使用方式：
|
||
detector = PitfallDetector(experience_store)
|
||
warnings = await detector.check_pitfalls(
|
||
task_type="code_review",
|
||
planned_steps=[plan_step1, plan_step2, ...],
|
||
)
|
||
|
||
匹配逻辑：
|
||
1. 检索同类任务的失败经验
|
||
2. 从失败经验中提取失败步骤
|
||
3. 将失败步骤与当前计划步骤进行关键词匹配
|
||
4. 计算失败率并分配预警级别
|
||
|
||
预警级别：
|
||
- HIGH: failure_rate >= 0.5（历史高失败率步骤）
|
||
- MEDIUM: failure_rate >= 0.2（有失败记录但频率低）
|
||
- LOW: 有任何失败记录
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
experience_store: ExperienceStoreProtocol,
|
||
similarity_threshold: float = 0.3,
|
||
max_search_results: int = 50,
|
||
):
|
||
"""
|
||
Args:
|
||
experience_store: 经验存储实例（ExperienceStore 或 InMemoryExperienceStore）
|
||
similarity_threshold: 步骤名称关键词匹配的最小相似度阈值
|
||
max_search_results: 从经验存储检索的最大结果数
|
||
"""
|
||
self._store = experience_store
|
||
self._similarity_threshold = similarity_threshold
|
||
self._max_search_results = max_search_results
|
||
|
||
async def check_pitfalls(
|
||
self,
|
||
task_type: str,
|
||
planned_steps: list[Any],
|
||
) -> list[PitfallWarning]:
|
||
"""检查计划步骤中的潜在陷阱
|
||
|
||
Args:
|
||
task_type: 任务类型
|
||
planned_steps: 计划步骤列表（PlanStep 对象或具有 name/description 属性的对象）
|
||
|
||
Returns:
|
||
按严重程度排序的预警列表（HIGH → MEDIUM → LOW）
|
||
"""
|
||
if not planned_steps:
|
||
return []
|
||
|
||
# 1. 检索同类任务的所有经验（包含成功和失败，用于计算步骤级失败率）
|
||
all_experiences = await self._search_experiences(task_type)
|
||
if not all_experiences:
|
||
logger.debug(f"No experiences found for task_type={task_type}")
|
||
return []
|
||
|
||
# 2. 从经验中提取步骤级别的失败统计
|
||
step_failure_stats = self._extract_step_failure_stats(all_experiences)
|
||
|
||
# 3. 匹配当前计划步骤并生成预警
|
||
warnings = self._match_and_warn(planned_steps, step_failure_stats)
|
||
|
||
# 4. 按严重程度排序（HIGH → MEDIUM → LOW），同级别按失败率降序
|
||
warnings.sort(key=lambda w: (_warning_level_order(w.warning_level), -w.failure_rate))
|
||
|
||
if warnings:
|
||
logger.info(
|
||
f"PitfallDetector found {len(warnings)} warnings for task_type={task_type}: "
|
||
f"{sum(1 for w in warnings if w.warning_level == WarningLevel.HIGH)} HIGH, "
|
||
f"{sum(1 for w in warnings if w.warning_level == WarningLevel.MEDIUM)} MEDIUM, "
|
||
f"{sum(1 for w in warnings if w.warning_level == WarningLevel.LOW)} LOW"
|
||
)
|
||
|
||
return warnings
|
||
|
||
async def _search_experiences(self, task_type: str) -> list[Any]:
|
||
"""检索指定任务类型的所有经验（包含成功和失败）"""
|
||
try:
|
||
results = await self._store.search(
|
||
query=task_type,
|
||
top_k=self._max_search_results,
|
||
task_type=task_type,
|
||
)
|
||
return results
|
||
except (RuntimeError, ValueError, KeyError) as e:
|
||
logger.error(f"Failed to search experiences for pitfall detection: {e}")
|
||
return []
|
||
|
||
def _extract_step_failure_stats(
|
||
self, failed_experiences: list[Any]
|
||
) -> dict[str, _StepFailureStats]:
|
||
"""从失败经验中提取步骤级别的失败统计
|
||
|
||
steps_summary 可以是 str 或 list[dict]：
|
||
- list[dict]: 每个字典包含 step_name, outcome, duration_seconds, error
|
||
- str: 退化为整体统计
|
||
|
||
Returns:
|
||
以步骤名称为 key 的失败统计字典
|
||
"""
|
||
stats: dict[str, _StepFailureStats] = {}
|
||
|
||
for exp in failed_experiences:
|
||
steps_summary = exp.steps_summary
|
||
|
||
# 如果 steps_summary 是字符串，无法提取步骤级信息
|
||
if isinstance(steps_summary, str):
|
||
continue
|
||
|
||
if not isinstance(steps_summary, list):
|
||
continue
|
||
|
||
for step in steps_summary:
|
||
if not isinstance(step, dict):
|
||
continue
|
||
|
||
step_name = step.get("step_name", "")
|
||
if not step_name:
|
||
continue
|
||
|
||
outcome = step.get("outcome", "")
|
||
error = step.get("error", "")
|
||
|
||
if step_name not in stats:
|
||
stats[step_name] = _StepFailureStats(
|
||
step_name=step_name,
|
||
total_occurrences=0,
|
||
failure_occurrences=0,
|
||
failure_reasons=[],
|
||
optimization_tips=[],
|
||
)
|
||
|
||
s = stats[step_name]
|
||
s.total_occurrences += 1
|
||
|
||
if outcome in ("failure", "failed", "error"):
|
||
s.failure_occurrences += 1
|
||
if error:
|
||
s.failure_reasons.append(error)
|
||
|
||
# 收集优化建议 — only add to steps that are part of this experience
|
||
if hasattr(exp, 'optimization_tips') and exp.optimization_tips:
|
||
experience_steps = set(exp.steps) if hasattr(exp, 'steps') and exp.steps else set()
|
||
for step_name, s in stats.items():
|
||
if experience_steps and step_name in experience_steps:
|
||
s.optimization_tips.extend(exp.optimization_tips)
|
||
|
||
return stats
|
||
|
||
def _match_and_warn(
|
||
self,
|
||
planned_steps: list[Any],
|
||
step_failure_stats: dict[str, _StepFailureStats],
|
||
) -> list[PitfallWarning]:
|
||
"""将计划步骤与失败统计进行匹配，生成预警"""
|
||
warnings: list[PitfallWarning] = []
|
||
|
||
for step in planned_steps:
|
||
step_name = getattr(step, "name", "")
|
||
step_description = getattr(step, "description", "")
|
||
|
||
if not step_name:
|
||
continue
|
||
|
||
# 查找最佳匹配的失败步骤
|
||
best_match: _StepFailureStats | None = None
|
||
best_similarity = 0.0
|
||
|
||
for stats_step_name, stats in step_failure_stats.items():
|
||
similarity = _compute_name_similarity(
|
||
step_name, step_description, stats_step_name
|
||
)
|
||
if similarity > best_similarity:
|
||
best_similarity = similarity
|
||
best_match = stats
|
||
|
||
# 相似度低于阈值，跳过
|
||
if best_match is None or best_similarity < self._similarity_threshold:
|
||
continue
|
||
|
||
# 计算失败率
|
||
failure_rate = (
|
||
best_match.failure_occurrences / best_match.total_occurrences
|
||
if best_match.total_occurrences > 0
|
||
else 0.0
|
||
)
|
||
|
||
# 分配预警级别
|
||
warning_level = _determine_warning_level(failure_rate)
|
||
|
||
# 生成建议
|
||
suggestion = _build_suggestion(best_match, failure_rate)
|
||
|
||
warning = PitfallWarning(
|
||
step_name=step_name,
|
||
warning_level=warning_level,
|
||
failure_rate=round(failure_rate, 4),
|
||
historical_failures=best_match.failure_reasons[:5], # 最多保留 5 条
|
||
suggestion=suggestion,
|
||
)
|
||
warnings.append(warning)
|
||
|
||
return warnings
|
||
|
||
|
||
# ── 内部辅助类 ──────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class _StepFailureStats:
|
||
"""步骤级别的失败统计（内部使用）"""
|
||
|
||
step_name: str
|
||
total_occurrences: int
|
||
failure_occurrences: int
|
||
failure_reasons: list[str]
|
||
optimization_tips: list[str]
|
||
|
||
|
||
# ── 辅助函数 ──────────────────────────────────────────────
|
||
|
||
|
||
def _compute_name_similarity(
|
||
step_name: str, step_description: str, historical_step_name: str
|
||
) -> float:
|
||
"""计算步骤名称的关键词重叠相似度
|
||
|
||
基于关键词集合的 Jaccard 相似度，同时考虑 step_name 和 step_description。
|
||
|
||
Args:
|
||
step_name: 当前计划步骤名称
|
||
step_description: 当前计划步骤描述
|
||
historical_step_name: 历史步骤名称
|
||
|
||
Returns:
|
||
相似度分数（0.0 ~ 1.0）
|
||
"""
|
||
# 提取关键词：将名称拆分为词，过滤掉常见停用词
|
||
current_keywords = _extract_keywords(f"{step_name} {step_description}")
|
||
historical_keywords = _extract_keywords(historical_step_name)
|
||
|
||
if not current_keywords or not historical_keywords:
|
||
return 0.0
|
||
|
||
# Jaccard 相似度
|
||
intersection = current_keywords & historical_keywords
|
||
union = current_keywords | historical_keywords
|
||
|
||
if not union:
|
||
return 0.0
|
||
|
||
return len(intersection) / len(union)
|
||
|
||
|
||
_STOP_WORDS = frozenset({
|
||
"a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "for",
|
||
"of", "with", "by", "from", "is", "are", "was", "were", "be", "been",
|
||
"being", "have", "has", "had", "do", "does", "did", "will", "would",
|
||
"could", "should", "may", "might", "can", "shall", "not", "no",
|
||
})
|
||
|
||
|
||
def _extract_keywords(text: str) -> frozenset[str]:
|
||
"""从文本中提取关键词集合
|
||
|
||
转小写、按空白/下划线/连字符拆分、过滤停用词和单字符词。
|
||
"""
|
||
# 统一分隔符
|
||
normalized = text.lower().replace("_", " ").replace("-", " ")
|
||
words = normalized.split()
|
||
return frozenset(
|
||
w for w in words
|
||
if len(w) > 1 and w not in _STOP_WORDS
|
||
)
|
||
|
||
|
||
def _determine_warning_level(failure_rate: float) -> WarningLevel:
|
||
"""根据失败率确定预警级别
|
||
|
||
- HIGH: failure_rate >= 0.5
|
||
- MEDIUM: failure_rate >= 0.2
|
||
- LOW: 有任何失败记录
|
||
"""
|
||
if failure_rate >= _HIGH_THRESHOLD:
|
||
return WarningLevel.HIGH
|
||
if failure_rate >= _MEDIUM_THRESHOLD:
|
||
return WarningLevel.MEDIUM
|
||
return WarningLevel.LOW
|
||
|
||
|
||
def _warning_level_order(level: WarningLevel) -> int:
|
||
"""预警级别排序值（越小越严重）"""
|
||
return {
|
||
WarningLevel.HIGH: 0,
|
||
WarningLevel.MEDIUM: 1,
|
||
WarningLevel.LOW: 2,
|
||
}[level]
|
||
|
||
|
||
def _build_suggestion(stats: _StepFailureStats, failure_rate: float) -> str:
|
||
"""根据失败统计生成优化建议"""
|
||
parts: list[str] = []
|
||
|
||
if failure_rate >= _HIGH_THRESHOLD:
|
||
parts.append(f"该步骤历史失败率高达 {failure_rate:.0%}，建议重点关注")
|
||
elif failure_rate >= _MEDIUM_THRESHOLD:
|
||
parts.append(f"该步骤历史失败率为 {failure_rate:.0%}，需注意风险")
|
||
else:
|
||
parts.append(f"该步骤有少量失败记录（失败率 {failure_rate:.0%}）")
|
||
|
||
if stats.failure_reasons:
|
||
unique_reasons = list(dict.fromkeys(stats.failure_reasons))[:3]
|
||
reasons_str = "、".join(unique_reasons)
|
||
parts.append(f"常见失败原因：{reasons_str}")
|
||
|
||
if stats.optimization_tips:
|
||
unique_tips = list(dict.fromkeys(stats.optimization_tips))[:2]
|
||
tips_str = "；".join(unique_tips)
|
||
parts.append(f"建议：{tips_str}")
|
||
|
||
return "。".join(parts)
|