Agents API

Agent types and base classes for building SWARM simulations.

BaseAgent

Abstract base class for all agents.

swarm.agents.base.BaseAgent

Bases: ABC

Abstract base class for agent policies.

Agents observe the environment and take actions based on their behavioral policy.

Source code in swarm/agents/base.py

class BaseAgent(ABC):
    """
    Abstract base class for agent policies.

    Agents observe the environment and take actions based on their
    behavioral policy.
    """

    def __init__(
        self,
        agent_id: str,
        agent_type: AgentType,
        roles: Optional[List[Role]] = None,
        config: Optional[Dict] = None,
        name: Optional[str] = None,
    ):
        """
        Initialize agent.

        Args:
            agent_id: Unique identifier
            agent_type: Behavioral archetype
            roles: List of roles this agent can fulfill
            config: Agent-specific configuration
            name: Human-readable label (defaults to agent_id)
        """
        self.agent_id = agent_id
        self.name = name or agent_id
        self.agent_type = agent_type
        self.roles = roles or [Role.WORKER]
        self.config = config or {}

        # Internal state
        self._memory: List[Dict] = []
        self._interaction_history: List[SoftInteraction] = []

    @property
    def primary_role(self) -> Role:
        """Get the agent's primary role."""
        return self.roles[0] if self.roles else Role.WORKER

    @abstractmethod
    def act(self, observation: Observation) -> Action:
        """
        Decide on an action given the current observation.

        Args:
            observation: Current view of the environment

        Returns:
            Action to take
        """
        pass

    @abstractmethod
    def accept_interaction(
        self,
        proposal: InteractionProposal,
        observation: Observation,
    ) -> bool:
        """
        Decide whether to accept a proposed interaction.

        Args:
            proposal: The interaction proposal
            observation: Current observation

        Returns:
            True to accept, False to reject
        """
        pass

    @abstractmethod
    def propose_interaction(
        self,
        observation: Observation,
        counterparty_id: str,
    ) -> Optional[InteractionProposal]:
        """
        Create an interaction proposal for a counterparty.

        Args:
            observation: Current observation
            counterparty_id: Target agent ID

        Returns:
            InteractionProposal or None if not proposing
        """
        pass

    def update_from_outcome(
        self,
        interaction: SoftInteraction,
        payoff: float,
    ) -> None:
        """
        Update internal state after an interaction completes.

        Args:
            interaction: The completed interaction
            payoff: Payoff received
        """
        self._interaction_history.append(interaction)
        self._memory.append({
            "type": "interaction_outcome",
            "interaction_id": interaction.interaction_id,
            "counterparty": (
                interaction.counterparty
                if interaction.initiator == self.agent_id
                else interaction.initiator
            ),
            "p": interaction.p,
            "payoff": payoff,
            "accepted": interaction.accepted,
            "timestamp": datetime.now().isoformat(),
        })

    def remember(self, memory_item: Dict) -> None:
        """Add an item to memory."""
        memory_item["timestamp"] = datetime.now().isoformat()
        self._memory.append(memory_item)

    def get_memory(self, limit: int = 100) -> List[Dict]:
        """Get recent memory items."""
        return self._memory[-limit:]

    def get_interaction_history(self, limit: int = 50) -> List[SoftInteraction]:
        """Get recent interaction history."""
        return self._interaction_history[-limit:]

    def compute_counterparty_trust(self, counterparty_id: str) -> float:
        """
        Compute trust score for a counterparty based on history.

        Args:
            counterparty_id: ID of the counterparty

        Returns:
            Trust score in [0, 1]
        """
        relevant = [
            i for i in self._interaction_history
            if (i.initiator == counterparty_id or i.counterparty == counterparty_id)
            and i.accepted
        ]

        if not relevant:
            return 0.5  # Neutral for unknown agents

        # Average p from past interactions
        avg_p = sum(i.p for i in relevant) / len(relevant)
        return avg_p

    def should_post(self, observation: Observation) -> bool:
        """Determine if agent should create a post."""
        return observation.can_post

    def should_vote(self, observation: Observation) -> bool:
        """Determine if agent should vote."""
        return observation.can_vote and len(observation.visible_posts) > 0

    def should_interact(self, observation: Observation) -> bool:
        """Determine if agent should initiate an interaction."""
        return observation.can_interact

    def should_claim_task(self, observation: Observation) -> bool:
        """Determine if agent should claim a task."""
        return observation.can_claim_task and len(observation.available_tasks) > 0

    def create_noop_action(self) -> Action:
        """Create a no-op action."""
        return Action(
            action_type=ActionType.NOOP,
            agent_id=self.agent_id,
        )

    def create_post_action(self, content: str) -> Action:
        """Create a post action."""
        return Action(
            action_type=ActionType.POST,
            agent_id=self.agent_id,
            content=content,
        )

    def create_reply_action(self, post_id: str, content: str) -> Action:
        """Create a reply action."""
        return Action(
            action_type=ActionType.REPLY,
            agent_id=self.agent_id,
            target_id=post_id,
            content=content,
        )

    def create_vote_action(self, post_id: str, direction: int) -> Action:
        """Create a vote action (+1 upvote, -1 downvote)."""
        return Action(
            action_type=ActionType.VOTE,
            agent_id=self.agent_id,
            target_id=post_id,
            vote_direction=direction,
        )

    def create_propose_action(
        self,
        counterparty_id: str,
        interaction_type: InteractionType,
        content: str = "",
        task_id: Optional[str] = None,
    ) -> Action:
        """Create an interaction proposal action."""
        return Action(
            action_type=ActionType.PROPOSE_INTERACTION,
            agent_id=self.agent_id,
            counterparty_id=counterparty_id,
            interaction_type=interaction_type,
            content=content,
            target_id=task_id or "",
        )

    def create_accept_action(self, proposal_id: str) -> Action:
        """Create an interaction acceptance action."""
        return Action(
            action_type=ActionType.ACCEPT_INTERACTION,
            agent_id=self.agent_id,
            target_id=proposal_id,
        )

    def create_reject_action(self, proposal_id: str) -> Action:
        """Create an interaction rejection action."""
        return Action(
            action_type=ActionType.REJECT_INTERACTION,
            agent_id=self.agent_id,
            target_id=proposal_id,
        )

    def create_claim_task_action(self, task_id: str) -> Action:
        """Create a task claim action."""
        return Action(
            action_type=ActionType.CLAIM_TASK,
            agent_id=self.agent_id,
            target_id=task_id,
        )

    def create_submit_output_action(self, task_id: str, content: str) -> Action:
        """Create a task output submission action."""
        return Action(
            action_type=ActionType.SUBMIT_OUTPUT,
            agent_id=self.agent_id,
            target_id=task_id,
            content=content,
        )

    def create_post_bounty_action(
        self,
        reward_amount: float,
        task_description: str = "",
        min_reputation: float = 0.0,
        deadline_epoch: Optional[int] = None,
    ) -> Action:
        """Create an action to post a bounty."""
        return Action(
            action_type=ActionType.POST_BOUNTY,
            agent_id=self.agent_id,
            content=task_description,
            metadata={
                "reward_amount": reward_amount,
                "min_reputation": min_reputation,
                "deadline_epoch": deadline_epoch,
            },
        )

    def create_place_bid_action(
        self,
        bounty_id: str,
        bid_amount: float,
        message: str = "",
    ) -> Action:
        """Create an action to place a bid on a bounty."""
        return Action(
            action_type=ActionType.PLACE_BID,
            agent_id=self.agent_id,
            target_id=bounty_id,
            content=message,
            metadata={"bid_amount": bid_amount},
        )

    def create_accept_bid_action(self, bounty_id: str, bid_id: str) -> Action:
        """Create an action to accept a bid."""
        return Action(
            action_type=ActionType.ACCEPT_BID,
            agent_id=self.agent_id,
            target_id=bounty_id,
            metadata={"bid_id": bid_id},
        )

    def create_reject_bid_action(self, bid_id: str) -> Action:
        """Create an action to reject a bid."""
        return Action(
            action_type=ActionType.REJECT_BID,
            agent_id=self.agent_id,
            target_id=bid_id,
        )

    def create_withdraw_bid_action(self, bid_id: str) -> Action:
        """Create an action to withdraw a bid."""
        return Action(
            action_type=ActionType.WITHDRAW_BID,
            agent_id=self.agent_id,
            target_id=bid_id,
        )

    def create_file_dispute_action(self, escrow_id: str, reason: str = "") -> Action:
        """Create an action to file a dispute."""
        return Action(
            action_type=ActionType.FILE_DISPUTE,
            agent_id=self.agent_id,
            target_id=escrow_id,
            content=reason,
        )

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(id={self.agent_id}, type={self.agent_type.value})"

Usage

from swarm.agents.base import BaseAgent, Action, Observation

class MyAgent(BaseAgent):
    def decide(self, observation: Observation) -> Action:
        # Implement decision logic
        pass

    def update(self, result) -> None:
        # Update internal state
        pass

Built-in Agent Types

HonestAgent

Cooperative agent that completes tasks diligently.

from swarm.agents.honest import HonestAgent

agent = HonestAgent(
    agent_id="honest_1",
    name="Alice",
    cooperation_threshold=0.7,
)

All agents accept an optional name parameter for human-readable display (defaults to agent_id).

OpportunisticAgent

Payoff-maximizing agent that cherry-picks high-value interactions.

from swarm.agents.opportunistic import OpportunisticAgent

agent = OpportunisticAgent(
    agent_id="opp_1",
    cherry_pick_threshold=0.6,
)

DeceptiveAgent

Builds trust through honest behavior, then exploits trusted relationships.

from swarm.agents.deceptive import DeceptiveAgent

agent = DeceptiveAgent(
    agent_id="dec_1",
    trust_building_epochs=5,
    exploitation_threshold=0.8,
)

AdversarialAgent

Actively disrupts the ecosystem by targeting honest agents.

from swarm.agents.adversarial import AdversarialAgent

agent = AdversarialAgent(
    agent_id="adv_1",
    target_selection="highest_reputation",
)

LLMAgent

LLM-powered agent with configurable persona.

from swarm.agents.llm_agent import LLMAgent
from swarm.agents.llm_config import LLMConfig

config = LLMConfig(
    provider="anthropic",
    model="claude-3-haiku-20240307",
    temperature=0.7,
)

agent = LLMAgent(
    agent_id="llm_1",
    config=config,
    persona="You are a helpful, collaborative agent.",
)

Agent Roles

Mixins that add specific behaviors to agents.

PosterRole

from swarm.agents.roles.poster import PosterRole, ContentStrategy

class ContentAgent(BaseAgent, PosterRole):
    def __init__(self, agent_id: str):
        super().__init__(agent_id)
        self.set_strategy(ContentStrategy(
            reply_priority=0.7,
            topics=["AI", "Safety"],
        ))

WorkerRole

from swarm.agents.roles.worker import WorkerRole

class WorkerAgent(BaseAgent, WorkerRole):
    pass

VerifierRole

from swarm.agents.roles.verifier import VerifierRole

class VerifierAgent(BaseAgent, VerifierRole):
    pass

Data Types

Observation

What an agent sees when making decisions.

Field	Type	Description
available_tasks	list	Tasks that can be claimed
visible_posts	list	Posts in the feed
agent_reputations	dict	Known agent reputations
own_reputation	float	Agent's current reputation
can_post	bool	Whether posting is allowed

Action

What an agent decides to do.

Field	Type	Description
action_type	ActionType	Type of action
target_id	str	Target of action
content	str	Content (for posts)
value	float	Value (for votes)

ActionType

from swarm.agents.base import ActionType

ActionType.CLAIM_TASK
ActionType.COMPLETE_TASK
ActionType.COLLABORATE
ActionType.POST
ActionType.REPLY
ActionType.VOTE
ActionType.WAIT