import numpy as np import math from typing import List, Dict, Any class MemorySlicingEngine: def __init__(self, alpha=0.3, beta=0.3, gamma=0.2, delta=0.2): self.alpha = alpha self.beta = beta self.gamma = gamma self.delta = delta def compute_relevance(self, memory_item: Dict[str, Any], child_task: Dict[str, Any]) -> float: # KeywordMatch: 关键词重合比率 kw_match = len(set(memory_item["keywords"]) & set(child_task["keywords"])) / max(1, len(child_task["keywords"])) # DepScore: 拓扑距离倒数 dep_score = 1.0 / (memory_item.get("distance_to_target", 5) + 1.0) # Temporal: 时间衰减因子 age = memory_item.get("age", 0) temporal_score = math.exp(-0.1 * age) # Semantic: 余弦相似度 semantic_sim = np.dot(memory_item["embedding"], child_task["embedding"]) / ( np.linalg.norm(memory_item["embedding"]) * np.linalg.norm(child_task["embedding"]) + 1e-9 ) # 综合关联打分 return (self.alpha * kw_match + self.beta * dep_score + self.gamma * temporal_score + self.delta * semantic_sim) def slice_memory(self, parent_memories: List[Dict[str, Any]], child_task: Dict[str, Any], theta=0.5) -> List[Dict[str, Any]]: sliced_memory = [] for item in parent_memories: r_score = self.compute_relevance(item, child_task) if r_score > theta: sliced_memory.append(item) return sliced_memory # [123, 172] class SpawnController: def __init__(self, threshold=0.7): self.threshold = threshold # 权重设置 self.weights = { "I_f": 0.30, # 文件依赖度权重 "C_c": 0.20, # 圈复杂度权重 "F_c": 0.25, # 测试级联失败权重 "O_c": 0.15, # 上下文饱和度权重 "U_c": 0.10 # 决策不确定度权重 } def evaluate_spawning_policy(self, metrics: Dict[str, float]) -> Dict[str, Any]: # metrics 包含归一化到 [0, 1] 之间的五个复杂度指标 spawn_score = sum(self.weights[k] * metrics[k] for k in self.weights) if spawn_score > self.threshold: # 寻找主导核心指标触发定向专家 dominant_metric = max(metrics, key=metrics.get) specialization = "Generalist" if dominant_metric == "I_f": specialization = "Refactoring Specialist" elif dominant_metric == "C_c": specialization = "Code Simplification Agent" elif dominant_metric == "F_c": specialization = "Testing & Debugging Expert" elif dominant_metric == "O_c": specialization = "Context Compression Agent" elif dominant_metric == "U_c": specialization = "Research & Analysis Agent" return {"decision": "spawn", "specialization": specialization, "score": spawn_score} return {"decision": "continue", "specialization": None, "score": spawn_score} # 测试运行自适应 Spawn if __name__ == "__main__": controller = SpawnController() # 模拟发生大规模测试失败且上下文面临饱和的环境指标 metrics = { "I_f": 0.4, "C_c": 0.5, "F_c": 0.9, # 高测试失败 "O_c": 0.8, # 上下文处于 40% 愚蠢区以上的高危态 "U_c": 0.3 } result = controller.evaluate_spawning_policy(metrics) print("Spawning Decision Outcome:", result)
夜雨聆风
