Spaces:
Sleeping
Sleeping
File size: 3,802 Bytes
3bfb250 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Base agent abstract class for ScrapeRL agents."""
from abc import ABC, abstractmethod
from typing import Any
from app.core.action import Action
from app.core.observation import Observation
class BaseAgent(ABC):
"""
Abstract base class for all agents in the ScrapeRL system.
Each agent specializes in a specific aspect of the scraping workflow:
- Planning and goal decomposition
- Navigation and URL prioritization
- Data extraction
- Verification and validation
- Memory operations
Agents communicate through message passing and coordinate via
the AgentCoordinator.
"""
def __init__(self, agent_id: str, config: dict[str, Any] | None = None):
"""
Initialize the agent.
Args:
agent_id: Unique identifier for this agent instance.
config: Optional configuration dictionary for the agent.
"""
self.agent_id = agent_id
self.config = config or {}
self._message_queue: list[dict[str, Any]] = []
self._action_history: list[Action] = []
@abstractmethod
async def act(self, observation: Observation) -> Action:
"""
Select an action based on the current observation.
This is the main decision-making method. The agent analyzes
the observation and returns the best action to take.
Args:
observation: The current state observation from the environment.
Returns:
The action to execute.
"""
pass
@abstractmethod
async def plan(self, observation: Observation) -> list[Action]:
"""
Create a plan of actions based on the current observation.
Unlike act() which returns a single action, plan() creates
a sequence of actions to achieve a goal.
Args:
observation: The current state observation from the environment.
Returns:
A list of planned actions in execution order.
"""
pass
async def explain(self, action: Action) -> str:
"""
Explain why this action was chosen.
Args:
action: The action to explain.
Returns:
A human-readable explanation of the action choice.
"""
return action.reasoning or "No explanation provided"
def receive_message(self, message: dict[str, Any]) -> None:
"""
Receive a message from another agent.
Args:
message: The message dictionary containing sender, type, and content.
"""
self._message_queue.append(message)
def get_pending_messages(self) -> list[dict[str, Any]]:
"""
Get all pending messages and clear the queue.
Returns:
List of pending messages.
"""
messages = self._message_queue.copy()
self._message_queue.clear()
return messages
def record_action(self, action: Action) -> None:
"""
Record an action in the agent's history.
Args:
action: The action that was executed.
"""
self._action_history.append(action)
def get_action_history(self) -> list[Action]:
"""
Get the history of actions taken by this agent.
Returns:
List of past actions.
"""
return self._action_history.copy()
def reset(self) -> None:
"""Reset the agent state for a new episode."""
self._message_queue.clear()
self._action_history.clear()
def __repr__(self) -> str:
"""String representation of the agent."""
return f"{self.__class__.__name__}(agent_id={self.agent_id!r})"
|