155 lines
4.4 KiB
Python
155 lines
4.4 KiB
Python
|
"""
|
||
|
|
竞标评估器 - 评估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]
|