8a538bb950
- Renamed `check_environment` to `check_api_key_configured` for clarity, simplifying the API key validation logic. - Removed the blocking behavior of the API key check during application startup, allowing the app to run while providing a prompt for configuration. - Updated `LocalAgentApp` to accept an `api_configured` parameter, enabling conditional messaging for API key setup. - Enhanced the `SandboxRunner` to support backup management and improved execution result handling with detailed metrics. - Integrated data governance strategies into the `HistoryManager`, ensuring compliance and improved data management. - Added privacy settings and metrics tracking across various components to enhance user experience and application safety.
253 lines
8.8 KiB
Python
253 lines
8.8 KiB
Python
"""
|
||
任务复用度量指标收集模块
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, List
|
||
from dataclasses import dataclass, asdict
|
||
|
||
|
||
@dataclass
|
||
class ReuseEvent:
|
||
"""复用事件记录"""
|
||
timestamp: str
|
||
original_task_id: str # 被复用的任务 ID
|
||
new_task_id: Optional[str] # 新任务 ID（如果执行了）
|
||
similarity_score: float # 相似度分数
|
||
user_action: str # 用户操作：accepted/rejected/rollback/failed
|
||
differences_count: int # 差异数量
|
||
critical_differences: int # 关键差异数量
|
||
execution_success: Optional[bool] # 执行是否成功（如果执行了）
|
||
|
||
|
||
class ReuseMetrics:
|
||
"""复用指标管理器"""
|
||
|
||
def __init__(self, workspace_path: Path):
|
||
self.workspace = workspace_path
|
||
self.metrics_file = workspace_path / "reuse_metrics.json"
|
||
self._events: List[ReuseEvent] = []
|
||
self._load()
|
||
|
||
def _load(self):
|
||
"""加载指标数据"""
|
||
if self.metrics_file.exists():
|
||
try:
|
||
with open(self.metrics_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
self._events = [ReuseEvent(**event) for event in data]
|
||
except Exception as e:
|
||
print(f"[警告] 加载复用指标失败: {e}")
|
||
self._events = []
|
||
|
||
def _save(self):
|
||
"""保存指标数据"""
|
||
try:
|
||
self.metrics_file.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(self.metrics_file, 'w', encoding='utf-8') as f:
|
||
data = [asdict(event) for event in self._events]
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"[警告] 保存复用指标失败: {e}")
|
||
|
||
def record_reuse_offered(
|
||
self,
|
||
original_task_id: str,
|
||
similarity_score: float,
|
||
differences_count: int,
|
||
critical_differences: int
|
||
):
|
||
"""记录复用建议被提供"""
|
||
event = ReuseEvent(
|
||
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
original_task_id=original_task_id,
|
||
new_task_id=None,
|
||
similarity_score=similarity_score,
|
||
user_action='offered',
|
||
differences_count=differences_count,
|
||
critical_differences=critical_differences,
|
||
execution_success=None
|
||
)
|
||
self._events.append(event)
|
||
self._save()
|
||
return event
|
||
|
||
def record_reuse_accepted(
|
||
self,
|
||
original_task_id: str,
|
||
similarity_score: float,
|
||
differences_count: int,
|
||
critical_differences: int
|
||
):
|
||
"""记录用户接受复用"""
|
||
event = ReuseEvent(
|
||
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
original_task_id=original_task_id,
|
||
new_task_id=None,
|
||
similarity_score=similarity_score,
|
||
user_action='accepted',
|
||
differences_count=differences_count,
|
||
critical_differences=critical_differences,
|
||
execution_success=None
|
||
)
|
||
self._events.append(event)
|
||
self._save()
|
||
return event
|
||
|
||
def record_reuse_rejected(
|
||
self,
|
||
original_task_id: str,
|
||
similarity_score: float,
|
||
differences_count: int,
|
||
critical_differences: int
|
||
):
|
||
"""记录用户拒绝复用"""
|
||
event = ReuseEvent(
|
||
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
original_task_id=original_task_id,
|
||
new_task_id=None,
|
||
similarity_score=similarity_score,
|
||
user_action='rejected',
|
||
differences_count=differences_count,
|
||
critical_differences=critical_differences,
|
||
execution_success=None
|
||
)
|
||
self._events.append(event)
|
||
self._save()
|
||
return event
|
||
|
||
def record_reuse_execution(
|
||
self,
|
||
original_task_id: str,
|
||
new_task_id: str,
|
||
success: bool
|
||
):
|
||
"""记录复用后的执行结果"""
|
||
# 查找最近的 accepted 事件并更新
|
||
for event in reversed(self._events):
|
||
if (event.original_task_id == original_task_id and
|
||
event.user_action == 'accepted' and
|
||
event.new_task_id is None):
|
||
event.new_task_id = new_task_id
|
||
event.execution_success = success
|
||
self._save()
|
||
return event
|
||
|
||
# 如果没找到，创建新记录
|
||
event = ReuseEvent(
|
||
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
original_task_id=original_task_id,
|
||
new_task_id=new_task_id,
|
||
similarity_score=0.0,
|
||
user_action='executed',
|
||
differences_count=0,
|
||
critical_differences=0,
|
||
execution_success=success
|
||
)
|
||
self._events.append(event)
|
||
self._save()
|
||
return event
|
||
|
||
def record_reuse_rollback(
|
||
self,
|
||
original_task_id: str,
|
||
new_task_id: str
|
||
):
|
||
"""记录复用后回滚（用户撤销/重做）"""
|
||
event = ReuseEvent(
|
||
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
original_task_id=original_task_id,
|
||
new_task_id=new_task_id,
|
||
similarity_score=0.0,
|
||
user_action='rollback',
|
||
differences_count=0,
|
||
critical_differences=0,
|
||
execution_success=False
|
||
)
|
||
self._events.append(event)
|
||
self._save()
|
||
return event
|
||
|
||
def get_statistics(self) -> Dict:
|
||
"""获取统计数据"""
|
||
if not self._events:
|
||
return {
|
||
'total_offered': 0,
|
||
'total_accepted': 0,
|
||
'total_rejected': 0,
|
||
'total_executed': 0,
|
||
'total_rollback': 0,
|
||
'acceptance_rate': 0.0,
|
||
'rejection_rate': 0.0,
|
||
'success_rate': 0.0,
|
||
'failure_rate': 0.0,
|
||
'rollback_rate': 0.0,
|
||
'avg_similarity': 0.0,
|
||
'avg_differences': 0.0,
|
||
'avg_critical_differences': 0.0
|
||
}
|
||
|
||
offered = [e for e in self._events if e.user_action == 'offered']
|
||
accepted = [e for e in self._events if e.user_action == 'accepted']
|
||
rejected = [e for e in self._events if e.user_action == 'rejected']
|
||
executed = [e for e in self._events if e.execution_success is not None]
|
||
rollback = [e for e in self._events if e.user_action == 'rollback']
|
||
|
||
total_offered = len(offered)
|
||
total_accepted = len(accepted)
|
||
total_rejected = len(rejected)
|
||
total_executed = len(executed)
|
||
total_rollback = len(rollback)
|
||
|
||
# 计算成功和失败
|
||
successful = [e for e in executed if e.execution_success]
|
||
failed = [e for e in executed if not e.execution_success]
|
||
|
||
# 计算率
|
||
acceptance_rate = total_accepted / total_offered if total_offered > 0 else 0.0
|
||
rejection_rate = total_rejected / total_offered if total_offered > 0 else 0.0
|
||
success_rate = len(successful) / total_executed if total_executed > 0 else 0.0
|
||
failure_rate = len(failed) / total_executed if total_executed > 0 else 0.0
|
||
rollback_rate = total_rollback / total_executed if total_executed > 0 else 0.0
|
||
|
||
# 平均值
|
||
all_events = offered + accepted + rejected
|
||
avg_similarity = sum(e.similarity_score for e in all_events) / len(all_events) if all_events else 0.0
|
||
avg_differences = sum(e.differences_count for e in all_events) / len(all_events) if all_events else 0.0
|
||
avg_critical_differences = sum(e.critical_differences for e in all_events) / len(all_events) if all_events else 0.0
|
||
|
||
return {
|
||
'total_offered': total_offered,
|
||
'total_accepted': total_accepted,
|
||
'total_rejected': total_rejected,
|
||
'total_executed': total_executed,
|
||
'total_rollback': total_rollback,
|
||
'acceptance_rate': acceptance_rate,
|
||
'rejection_rate': rejection_rate,
|
||
'success_rate': success_rate,
|
||
'failure_rate': failure_rate,
|
||
'rollback_rate': rollback_rate,
|
||
'avg_similarity': avg_similarity,
|
||
'avg_differences': avg_differences,
|
||
'avg_critical_differences': avg_critical_differences
|
||
}
|
||
|
||
def get_recent_events(self, count: int = 20) -> List[ReuseEvent]:
|
||
"""获取最近的事件"""
|
||
return self._events[-count:] if self._events else []
|
||
|
||
|
||
# 全局单例
|
||
_metrics: Optional[ReuseMetrics] = None
|
||
|
||
|
||
def get_reuse_metrics(workspace_path: Path) -> ReuseMetrics:
|
||
"""获取复用指标管理器单例"""
|
||
global _metrics
|
||
if _metrics is None:
|
||
_metrics = ReuseMetrics(workspace_path)
|
||
return _metrics
|
||
|