Agent Layer

Config-driven for simple agents · class-based for custom logic — swarm/agents/

LLMAgent — config-driven (most common case)
# swarm/agents/llm_agent.py

class LLMAgent:
    """Config-driven agent. Implements Agent protocol."""
    def __init__(
        self,
        name: str,
        provider: LLMProvider,
        system_prompt: str = "",
        tools: list[Tool] | None = None,
        temperature: float = 0.7,
        max_tokens: int = 4096,
    ) -> None: ...

    async def run(self, task: str, ctx: SwarmContext) -> AgentResult:
        messages = self._build_messages(task, ctx)
        content = await self.provider.complete(messages, tools=self.tools)
        return AgentResult(agent_name=self.name, content=content)

    def _build_messages(self, task, ctx) -> list[Message]:
        msgs = []
        if self.system_prompt:
            msgs.append(Message("system", self.system_prompt))
        msgs.extend(ctx.history[-10:])   # last 10 turns for context
        msgs.append(Message("user", task))
        return msgs
BaseAgent — subclass for custom logic
# swarm/agents/base.py

class BaseAgent:
    """Subclass this for custom agent behavior."""
    def __init__(self, name: str, provider: LLMProvider) -> None:
        self.name = name
        self.provider = provider

    async def run(self, task: str, ctx: SwarmContext) -> AgentResult:
        raise NotImplementedError

# Custom agent example — user subclasses this
class ResearchAgent(BaseAgent):
    async def run(self, task: str, ctx: SwarmContext) -> AgentResult:
        # custom pre-processing
        enriched = f"Research thoroughly: {task}"
        content = await self.provider.complete([Message("user", enriched)])
        ctx.state["research"] = content           # write to shared store
        return AgentResult(
            agent_name=self.name,
            content=content,
            confidence=0.9,
            metadata={"sources": ["web", "docs"]},
        )
Agent factory — from dict/YAML config
# swarm/agents/__init__.py

def agent_from_config(config: dict, provider: LLMProvider) -> LLMAgent:
    """Build an LLMAgent from a plain dict. Used by CLI + Swarm.from_config()."""
    return LLMAgent(
        name=config["name"],
        provider=provider,
        system_prompt=config.get("system_prompt", ""),
        tools=[tool_from_config(t) for t in config.get("tools", [])],
        temperature=config.get("temperature", 0.7),
    )

# Usage
agent = agent_from_config({
    "name": "researcher",
    "system_prompt": "You are a thorough research assistant.",
    "tools": [{"name": "search", "description": "Search the web"}],
}, provider=ClaudeProvider())