hubian
05950a3c84
核心组件： - Orchestrator: 意图理解、任务拆分、竞标管理、结果验证 - Worker: 竞标任务、执行交付 - TaskBoard: 状态管理、信息存储 - BidEvaluator: 竞标评估算法 - ExecutionMonitor: 执行监控、超时处理 - LLMClient: 大模型接口调用 功能特性： - 竞标机制：Agent主动竞争任务 - 动态调度：串行/并行任务智能调度 - 智能容错：超时切换、验证重试 - 质量保证：结果验证、历史追踪 Web界面：首页、请求列表、任务列表、Agent管理 API接口：请求/任务/Agent管理、测试接口 端口：19015
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] |