"""TeamMember 实现 - Phase 10.1 团队成员实现,负责执行分配的任务。 """ from dataclasses import dataclass, field from typing import Any from enum import Enum class MemberStatus(Enum): """成员状态""" IDLE = "idle" BUSY = "busy" OFFLINE = "offline" @dataclass class MemberTask: """成员任务""" task_id: str description: str status: str = "pending" # pending, in_progress, completed, failed result: Any = None error: str | None = None class TeamMember: """团队成员 代表团队中的一个 Agent 成员,负责执行分配的任务。 """ def __init__(self, member_id: str, name: str, capabilities: list[str] | None = None): """ Args: member_id: 成员 ID name: 成员名称 capabilities: 成员能力列表 """ self.member_id = member_id self.name = name self.capabilities = capabilities or [] self.status = MemberStatus.IDLE self._tasks: dict[str, MemberTask] = {} self._metadata: dict[str, Any] = {} def assign_task(self, task_id: str, description: str) -> MemberTask: """接收任务分配 Args: task_id: 任务 ID description: 任务描述 Returns: 创建的任务对象 """ task = MemberTask(task_id=task_id, description=description) self._tasks[task_id] = task self.status = MemberStatus.BUSY return task def update_task_status( self, task_id: str, status: str, result: Any = None, error: str | None = None ) -> bool: """更新任务状态 Args: task_id: 任务 ID status: 新状态 result: 任务结果 error: 错误信息 Returns: 是否更新成功 """ if task_id not in self._tasks: return False task = self._tasks[task_id] task.status = status if result is not None: task.result = result if error is not None: task.error = error if status in ("completed", "failed"): self.status = MemberStatus.IDLE return True def get_task(self, task_id: str) -> MemberTask | None: """获取任务 Args: task_id: 任务 ID Returns: 任务对象或 None """ return self._tasks.get(task_id) def get_pending_tasks(self) -> list[MemberTask]: """获取待处理任务 Returns: 待处理任务列表 """ return [t for t in self._tasks.values() if t.status == "pending"] def get_active_task(self) -> MemberTask | None: """获取当前执行中的任务 Returns: 当前任务或 None """ for task in self._tasks.values(): if task.status == "in_progress": return task return None def get_completed_tasks(self) -> list[MemberTask]: """获取已完成任务 Returns: 已完成任务列表 """ return [t for t in self._tasks.values() if t.status == "completed"] def set_metadata(self, key: str, value: Any) -> None: """设置元数据 Args: key: 元数据键 value: 元数据值 """ self._metadata[key] = value def get_metadata(self, key: str) -> Any: """获取元数据 Args: key: 元数据键 Returns: 元数据值或 None """ return self._metadata.get(key) def get_status(self) -> dict[str, Any]: """获取成员状态 Returns: 状态字典 """ return { "member_id": self.member_id, "name": self.name, "status": self.status.value, "capabilities": self.capabilities, "task_count": len(self._tasks), "pending_count": len(self.get_pending_tasks()), "active_task": self.get_active_task().__dict__ if self.get_active_task() else None, }