hubian
05950a3c84
核心组件： - Orchestrator: 意图理解、任务拆分、竞标管理、结果验证 - Worker: 竞标任务、执行交付 - TaskBoard: 状态管理、信息存储 - BidEvaluator: 竞标评估算法 - ExecutionMonitor: 执行监控、超时处理 - LLMClient: 大模型接口调用 功能特性： - 竞标机制：Agent主动竞争任务 - 动态调度：串行/并行任务智能调度 - 智能容错：超时切换、验证重试 - 质量保证：结果验证、历史追踪 Web界面：首页、请求列表、任务列表、Agent管理 API接口：请求/任务/Agent管理、测试接口 端口：19015
238 lines
8.1 KiB
Python
238 lines
8.1 KiB
Python
"""
|
||
任务公告板 - 任务状态管理、信息存储
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import threading
|
||
from typing import Dict, List, Optional, Any
|
||
from pathlib import Path
|
||
from .models import Task, Bid, Execution, Attempt, TaskStatus, TaskGraph, UserRequest, AgentProfile
|
||
|
||
|
||
class TaskBoard:
|
||
"""任务公告板 - 存储和管理所有任务状态"""
|
||
|
||
def __init__(self, storage_path: str = None):
|
||
self.storage_path = Path(storage_path or Path.home() / ".multi_agent" / "task_board.json")
|
||
self.storage_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 内存存储
|
||
self.user_requests: Dict[str, UserRequest] = {}
|
||
self.tasks: Dict[str, Task] = {}
|
||
self.bids: Dict[str, Dict[str, List[Bid]]] = {} # task_id -> {bid_id: Bid}
|
||
self.executions: Dict[str, Execution] = {}
|
||
self.attempts: Dict[str, List[Attempt]] = {} # task_id -> [Attempt]
|
||
self.agents: Dict[str, AgentProfile] = {}
|
||
|
||
# 锁
|
||
self._lock = threading.Lock()
|
||
|
||
# 加载存储
|
||
self._load()
|
||
|
||
def _load(self):
|
||
"""从文件加载"""
|
||
if self.storage_path.exists():
|
||
try:
|
||
with open(self.storage_path, 'r') as f:
|
||
data = json.load(f)
|
||
# 这里可以恢复数据，简化处理，只保留基本结构
|
||
except Exception:
|
||
pass
|
||
|
||
def _save(self):
|
||
"""保存到文件"""
|
||
try:
|
||
data = {
|
||
'user_requests': {k: v.to_dict() for k, v in self.user_requests.items()},
|
||
'tasks': {k: v.to_dict() for k, v in self.tasks.items()},
|
||
'agents': {k: v.to_dict() for k, v in self.agents.items()},
|
||
'saved_at': time.time()
|
||
}
|
||
with open(self.storage_path, 'w') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"保存失败: {e}")
|
||
|
||
# === 用户请求管理 ===
|
||
|
||
def create_request(self, content: str) -> UserRequest:
|
||
"""创建用户请求"""
|
||
request = UserRequest(content=content)
|
||
with self._lock:
|
||
self.user_requests[request.id] = request
|
||
self._save()
|
||
return request
|
||
|
||
def get_request(self, request_id: str) -> Optional[UserRequest]:
|
||
"""获取用户请求"""
|
||
return self.user_requests.get(request_id)
|
||
|
||
def update_request(self, request: UserRequest):
|
||
"""更新用户请求"""
|
||
with self._lock:
|
||
self.user_requests[request.id] = request
|
||
self._save()
|
||
|
||
def list_requests(self, limit: int = 50) -> List[UserRequest]:
|
||
"""列出用户请求"""
|
||
requests = list(self.user_requests.values())
|
||
requests.sort(key=lambda r: r.created_at, reverse=True)
|
||
return requests[:limit]
|
||
|
||
# === 任务管理 ===
|
||
|
||
def add_task(self, task: Task):
|
||
"""添加任务"""
|
||
with self._lock:
|
||
self.tasks[task.id] = task
|
||
self.bids[task.id] = {}
|
||
self.attempts[task.id] = []
|
||
self._save()
|
||
|
||
def get_task(self, task_id: str) -> Optional[Task]:
|
||
"""获取任务"""
|
||
return self.tasks.get(task_id)
|
||
|
||
def update_task_status(self, task_id: str, status: TaskStatus):
|
||
"""更新任务状态"""
|
||
with self._lock:
|
||
if task_id in self.tasks:
|
||
self.tasks[task_id].status = status
|
||
self._save()
|
||
|
||
def get_tasks_by_status(self, status: TaskStatus) -> List[Task]:
|
||
"""获取指定状态的任务"""
|
||
return [t for t in self.tasks.values() if t.status == status]
|
||
|
||
def list_tasks(self, limit: int = 100) -> List[Task]:
|
||
"""列出所有任务"""
|
||
tasks = list(self.tasks.values())
|
||
tasks.sort(key=lambda t: t.created_at, reverse=True)
|
||
return tasks[:limit]
|
||
|
||
# === 竞标管理 ===
|
||
|
||
def add_bid(self, bid: Bid):
|
||
"""添加竞标"""
|
||
with self._lock:
|
||
if bid.task_id in self.bids:
|
||
self.bids[bid.task_id][bid.id] = bid
|
||
|
||
def get_bids(self, task_id: str) -> List[Bid]:
|
||
"""获取任务的竞标列表"""
|
||
if task_id in self.bids:
|
||
return list(self.bids[task_id].values())
|
||
return []
|
||
|
||
def clear_bids(self, task_id: str):
|
||
"""清除任务的竞标"""
|
||
with self._lock:
|
||
if task_id in self.bids:
|
||
self.bids[task_id] = {}
|
||
|
||
# === 执行管理 ===
|
||
|
||
def add_execution(self, execution: Execution):
|
||
"""添加执行记录"""
|
||
with self._lock:
|
||
self.executions[execution.id] = execution
|
||
|
||
def get_execution(self, execution_id: str) -> Optional[Execution]:
|
||
"""获取执行记录"""
|
||
return self.executions.get(execution_id)
|
||
|
||
def update_execution(self, execution: Execution):
|
||
"""更新执行记录"""
|
||
with self._lock:
|
||
self.executions[execution.id] = execution
|
||
self._save()
|
||
|
||
# === 尝试记录管理 ===
|
||
|
||
def add_attempt(self, attempt: Attempt):
|
||
"""添加尝试记录"""
|
||
with self._lock:
|
||
if attempt.task_id in self.attempts:
|
||
self.attempts[attempt.task_id].append(attempt)
|
||
|
||
def get_attempts(self, task_id: str) -> List[Attempt]:
|
||
"""获取任务的尝试记录"""
|
||
if task_id in self.attempts:
|
||
return self.attempts[task_id]
|
||
return []
|
||
|
||
def get_attempt_count(self, task_id: str) -> int:
|
||
"""获取任务的尝试次数"""
|
||
return len(self.get_attempts(task_id))
|
||
|
||
# === Agent管理 ===
|
||
|
||
def register_agent(self, agent: AgentProfile):
|
||
"""注册Agent"""
|
||
with self._lock:
|
||
self.agents[agent.id] = agent
|
||
self._save()
|
||
|
||
def unregister_agent(self, agent_id: str):
|
||
"""注销Agent"""
|
||
with self._lock:
|
||
self.agents.pop(agent_id, None)
|
||
self._save()
|
||
|
||
def get_agent(self, agent_id: str) -> Optional[AgentProfile]:
|
||
"""获取Agent"""
|
||
return self.agents.get(agent_id)
|
||
|
||
def list_agents(self) -> List[AgentProfile]:
|
||
"""列出所有Agent"""
|
||
return list(self.agents.values())
|
||
|
||
def update_agent_stats(
|
||
self,
|
||
agent_id: str,
|
||
success: bool,
|
||
duration: float,
|
||
quality_score: float = 0.0
|
||
):
|
||
"""更新Agent统计"""
|
||
with self._lock:
|
||
if agent_id in self.agents:
|
||
agent = self.agents[agent_id]
|
||
agent.total_tasks += 1
|
||
if success:
|
||
agent.success_tasks += 1
|
||
# 更新平均完成时间
|
||
prev_total_time = agent.avg_completion_time * (agent.success_tasks - 1)
|
||
agent.avg_completion_time = (prev_total_time + duration) / agent.success_tasks
|
||
# 更新平均质量评分
|
||
prev_total_score = agent.avg_quality_score * (agent.success_tasks - 1)
|
||
agent.avg_quality_score = (prev_total_score + quality_score) / agent.success_tasks
|
||
self._save()
|
||
|
||
def find_capable_agents(self, task: Task) -> List[AgentProfile]:
|
||
"""找到能处理该任务的Agent"""
|
||
capable = []
|
||
for agent in self.agents.values():
|
||
# 检查能力匹配
|
||
# 简化：所有Agent都可以尝试所有任务
|
||
# 实际应该根据task.input_data中的required_capabilities匹配
|
||
capable.append(agent)
|
||
return capable
|
||
|
||
# === 统计信息 ===
|
||
|
||
def get_stats(self) -> Dict:
|
||
"""获取统计信息"""
|
||
return {
|
||
'total_requests': len(self.user_requests),
|
||
'total_tasks': len(self.tasks),
|
||
'tasks_by_status': {
|
||
status.value: len([t for t in self.tasks.values() if t.status == status])
|
||
for status in TaskStatus
|
||
},
|
||
'total_agents': len(self.agents),
|
||
'total_executions': len(self.executions),
|
||
'total_bids': sum(len(b) for b in self.bids.values())
|
||
} |