Custom Agents Example

Learn how to create specialized agents with custom behaviors and capabilities.

Basic Custom Agent

Create a simple custom agent:

from agentic_fleet import Agent, AgentConfig
from agentic_fleet.behaviors import Behavior

class ResearchAgent(Agent):
    def __init__(
        self,
        name: str = "researcher",
        model: str = "gpt-4"
    ):
        config = AgentConfig(
            name=name,
            model=model,
            temperature=0.7,
            tools=["web_search", "file_reader"]
        )
        super().__init__(config)
    
    async def research(
        self,
        topic: str
    ) -> dict:
        """Conduct research on a topic."""
        results = await self.run(
            f"Research {topic} and provide summary"
        )
        return results

async def use_research_agent():
    # Create agent
    agent = ResearchAgent()
    
    # Use agent
    results = await agent.research(
        "quantum computing advances"
    )
    print("Research Results:", results)

# Run example
import asyncio
asyncio.run(use_research_agent())

Agent with Custom Behavior

Add specialized behaviors to agents:

from agentic_fleet.behaviors import Behavior
from typing import Dict, Any

class AnalysisBehavior(Behavior):
    """Adds data analysis capabilities."""
    
    async def analyze_data(
        self,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        # Implement analysis logic
        analysis = await self.process_data(data)
        return analysis
    
    async def process_data(
        self,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        # Process the data
        return {
            "summary": "Data analysis...",
            "insights": ["Insight 1", "Insight 2"]
        }

class AnalystAgent(Agent):
    def __init__(self, name: str = "analyst"):
        super().__init__(name=name)
        self.add_behavior(AnalysisBehavior())
    
    async def analyze(
        self,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        # Use behavior
        behavior = self.get_behavior(
            AnalysisBehavior
        )
        return await behavior.analyze_data(data)

async def use_analyst_agent():
    # Create agent
    agent = AnalystAgent()
    
    # Use agent
    data = {"values": [1, 2, 3, 4, 5]}
    analysis = await agent.analyze(data)
    print("Analysis:", analysis)

asyncio.run(use_analyst_agent())

Specialized Agent Types

Writing Agent

from agentic_fleet.tools import TextGenerator
from agentic_fleet.memory import Memory

class WritingAgent(Agent):
    def __init__(
        self,
        name: str = "writer",
        style: str = "professional"
    ):
        super().__init__(name=name)
        self.style = style
        self.memory = Memory()
        self.add_tool(TextGenerator())
    
    async def write_article(
        self,
        topic: str,
        length: str = "medium"
    ) -> str:
        # Generate article
        prompt = self.create_prompt(
            topic,
            length
        )
        
        article = await self.run(prompt)
        
        # Store in memory
        await self.memory.store(
            f"article_{topic}",
            article
        )
        
        return article
    
    def create_prompt(
        self,
        topic: str,
        length: str
    ) -> str:
        return f"""
        Write a {length} article about {topic}
        in a {self.style} style.
        """

async def use_writing_agent():
    # Create agent
    agent = WritingAgent(
        style="technical"
    )
    
    # Write article
    article = await agent.write_article(
        topic="AI Advances",
        length="short"
    )
    print("Article:", article)

asyncio.run(use_writing_agent())

Code Review Agent

from agentic_fleet.tools import (
    CodeAnalyzer,
    GitTool
)

class CodeReviewAgent(Agent):
    def __init__(
        self,
        name: str = "reviewer"
    ):
        super().__init__(name=name)
        self.add_tools([
            CodeAnalyzer(),
            GitTool()
        ])
    
    async def review_code(
        self,
        code: str,
        context: dict = None
    ) -> Dict[str, Any]:
        # Analyze code
        analysis = await self.use_tool(
            "code_analyzer",
            code=code
        )
        
        # Generate review
        review = await self.run(
            f"Review code: {code}\n"
            f"Analysis: {analysis}"
        )
        
        return {
            "analysis": analysis,
            "review": review,
            "suggestions": self.extract_suggestions(
                review
            )
        }
    
    def extract_suggestions(
        self,
        review: str
    ) -> list:
        # Extract improvement suggestions
        return ["Suggestion 1", "Suggestion 2"]

async def use_code_review_agent():
    # Create agent
    agent = CodeReviewAgent()
    
    # Review code
    code = """
    def add(a, b):
        return a + b
    """
    
    review = await agent.review_code(code)
    print("Code Review:", review)

asyncio.run(use_code_review_agent())

Agent with Custom Memory

from agentic_fleet.memory import BaseMemory
from typing import Optional

class CustomMemory(BaseMemory):
    def __init__(self):
        self.data = {}
    
    async def store(
        self,
        key: str,
        value: Any
    ):
        self.data[key] = value
    
    async def retrieve(
        self,
        key: str
    ) -> Optional[Any]:
        return self.data.get(key)
    
    async def clear(self):
        self.data.clear()

class MemoryAgent(Agent):
    def __init__(
        self,
        name: str = "memory_agent"
    ):
        super().__init__(name=name)
        self.memory = CustomMemory()
    
    async def remember(
        self,
        key: str,
        value: Any
    ):
        await self.memory.store(key, value)
    
    async def recall(
        self,
        key: str
    ) -> Optional[Any]:
        return await self.memory.retrieve(key)

async def use_memory_agent():
    # Create agent
    agent = MemoryAgent()
    
    # Store and retrieve
    await agent.remember("fact", "Important!")
    fact = await agent.recall("fact")
    print("Recalled:", fact)

asyncio.run(use_memory_agent())

Best Practices

  1. Clear Interface

    class WellDesignedAgent(Agent):
        """Agent with clear purpose and interface."""
        
        async def process(
            self,
            input_data: Dict[str, Any]
        ) -> Dict[str, Any]:
            """Process input data."""
            pass
    
  2. Resource Management

    async with agent:
        result = await agent.process(data)
    
  3. Error Handling

    try:
        result = await agent.process(data)
    except AgentError as e:
        logger.error(f"Agent error: {e}")
    
  4. Testing

    async def test_agent():
        agent = CustomAgent()
        result = await agent.process(
            test_data
        )
        assert result["status"] == "success"