long_articles_agentic_supply/supply/core/agent/base.py

"""Agent 抽象基类 —— 领域核心

定义 Agent 的契约：每个 Agent 拥有一组 Tool、一份 Memory、一个 run() 方法。
业务方继承 Agent 实现具体智能体逻辑，框架层通过 AgentOrchestrator 调度执行。
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional

from supply.core.agent.tool import Tool
from supply.core.agent.memory import Memory


# ==================== Agent 状态 ====================


class AgentStatus(str, Enum):
    IDLE = "idle"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    CANCELLED = "cancelled"


# ==================== Agent 上下文 ====================


@dataclass
class AgentContext:
    """Agent 执行上下文 —— 携带本次调用的会话状态

    每次 run() 传入一个新的 AgentContext，包含：
      - session_id: 会话标识，用于 Memory 存取和日志追踪
      - working_memory: 临时工作区，Agent 可在执行过程中读写
      - metadata: 业务方注入的额外参数（topic, user_id 等）

    与 trace_id 的关系：
      trace_id 是分布式追踪 ID（跨 Agent 调用），session_id 是单次 Agent 会话 ID。
      trace_id 可作为 session_id 使用，也可独立设置。
    """

    session_id: str = ""
    trace_id: str = ""
    working_memory: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)


# ==================== Agent 执行结果 ====================


@dataclass
class AgentResult:
    """Agent 执行结果"""

    status: AgentStatus = AgentStatus.IDLE
    data: Any = None
    error: Optional[str] = None
    # 执行过程中产生的日志事件（由 orchestator 统一推送 SLS）
    events: List[Dict[str, Any]] = field(default_factory=list)
    # 工具调用记录（用于审计和调试）
    tool_calls: List[Dict[str, Any]] = field(default_factory=list)

    @property
    def success(self) -> bool:
        return self.status == AgentStatus.SUCCESS

    @classmethod
    def ok(cls, data: Any = None, **kwargs) -> "AgentResult":
        return cls(status=AgentStatus.SUCCESS, data=data, **kwargs)

    @classmethod
    def fail(cls, error: str, **kwargs) -> "AgentResult":
        return cls(status=AgentStatus.FAILED, error=error, **kwargs)


# ==================== Agent 抽象基类 ====================


class Agent(ABC):
    """Agent 抽象基类

    每个 Agent 是独立的智能体单元，拥有：
      - name / description: 标识与描述
      - tools: 可调用的工具集合
      - memory: 记忆系统（会话记忆 / 向量记忆）

    使用方式：

        class ContentWriter(Agent):
            name = "content_writer"
            description = "长文写作智能体"

            def __init__(self):
                self._tools = [WebSearchTool(), FactCheckTool()]
                self._memory = ConversationMemory(max_turns=10)

            @property
            def tools(self) -> list[Tool]:
                return self._tools

            @property
            def memory(self) -> Memory:
                return self._memory

            async def run(self, ctx: AgentContext) -> AgentResult:
                # 1. 从 memory 获取历史
                # 2. 调用 LLM 规划
                # 3. 执行 tool 调用
                # 4. 更新 memory
                ...
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Agent 唯一标识"""
        ...

    @property
    @abstractmethod
    def description(self) -> str:
        """Agent 功能描述"""
        ...

    @property
    @abstractmethod
    def tools(self) -> List[Tool]:
        """Agent 可调用的工具列表"""
        ...

    @property
    @abstractmethod
    def memory(self) -> Memory:
        """Agent 的记忆系统"""
        ...

    @abstractmethod
    async def run(self, ctx: AgentContext) -> AgentResult:
        """执行 Agent 主逻辑

        Args:
            ctx: 执行上下文（session、trace、metadata）

        Returns:
            AgentResult: 执行结果（success/fail + data）
        """
        ...

    def __repr__(self) -> str:
        return f"<Agent name={self.name}>"