""" 竞标评估器 - 评估Agent竞标,选择最佳执行者 """ from typing import Dict, List from .models import Bid, Task, AgentProfile class BidEvaluator: """竞标评估器""" def __init__(self): self.weights = { 'capability': 0.3, # 能力匹配度 'confidence': 0.2, # 自信度 'time_efficiency': 0.2, # 时间效率 'approach_quality': 0.2, # 方案质量 'historical': 0.1 # 历史表现 } def evaluate_bid(self, bid: Bid, agent: AgentProfile, max_time: int = 300) -> float: """ 评估单个竞标 Args: bid: 竞标书 agent: Agent档案 max_time: 最大可接受时间 Returns: 综合得分 0-1 """ scores = {} # 1. 能力匹配度 scores['capability'] = bid.capability_match # 2. 自信度 scores['confidence'] = bid.confidence # 3. 时间效率(时间越短得分越高) scores['time_efficiency'] = max(0, 1 - bid.estimated_time / max_time) # 4. 方案质量 scores['approach_quality'] = self._rate_approach(bid.approach) # 5. 历史表现 scores['historical'] = agent.success_rate # 加权求和 final_score = sum( self.weights[k] * v for k, v in scores.items() ) return final_score def _rate_approach(self, approach: str) -> float: """评估方案质量""" score = 0.5 # 有备选方案加分 if '备选' in approach or 'alternative' in approach.lower(): score += 0.1 # 有错误处理加分 if '错误' in approach or '异常' in approach or 'error' in approach.lower(): score += 0.1 # 有详细描述加分 if len(approach) > 100: score += 0.1 # 有步骤分解加分 if '步骤' in approach or 'step' in approach.lower(): score += 0.1 return min(score, 1.0) def select_best_bid( self, bids: List[Bid], agents: Dict[str, AgentProfile] ) -> tuple: """ 选择最佳竞标 Args: bids: 竞标列表 agents: Agent档案字典 Returns: (best_bid, best_agent, scores_dict) """ if not bids: return (None, None, {}) scores = {} for bid in bids: agent = agents.get(bid.agent_id) if agent: score = self.evaluate_bid(bid, agent) scores[bid.id] = { 'score': score, 'bid': bid, 'agent': agent, 'details': { 'capability': bid.capability_match, 'confidence': bid.confidence, 'time_efficiency': max(0, 1 - bid.estimated_time / 300), 'approach_quality': self._rate_approach(bid.approach), 'historical': agent.success_rate } } if not scores: return (None, None, {}) # 找最高分 best_bid_id = max(scores.keys(), key=lambda k: scores[k]['score']) best = scores[best_bid_id] return (best['bid'], best['agent'], scores) def get_backup_agents( self, bids: List[Bid], agents: Dict[str, AgentProfile], selected_agent_id: str ) -> List[AgentProfile]: """ 获取备选Agent列表(按得分排序) Args: bids: 竞标列表 agents: Agent档案字典 selected_agent_id: 已选择的AgentID Returns: 备选Agent列表 """ backup = [] for bid in bids: if bid.agent_id != selected_agent_id: agent = agents.get(bid.agent_id) if agent: score = self.evaluate_bid(bid, agent) backup.append((agent, score)) # 按得分降序排序 backup.sort(key=lambda x: x[1], reverse=True) return [a for a, s in backup]