Custom Agents Example
Custom Agents Example
Learn how to create and customize agents in AgenticFleet
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
-
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
-
Resource Management
async with agent: result = await agent.process(data)
-
Error Handling
try: result = await agent.process(data) except AgentError as e: logger.error(f"Agent error: {e}")
-
Testing
async def test_agent(): agent = CustomAgent() result = await agent.process( test_data ) assert result["status"] == "success"