Agents

Agents are AI-powered entities that execute tasks intelligently.

They bring reasoning and adaptability to your workflows through LLM integration.

Every agent provides durable conversation state - your AI interactions persist across failures and restarts.

What Agents Provide

Agents add intelligence and reasoning to your applications:

• Intelligent decision-making within tasks and workflows
• Natural language processing and generation with multiple LLM providers
• Conversation memory and context management with automatic persistence
• Specialized capabilities for different domains and use cases
• Multi-agent coordination patterns for complex collaborative operations

Key Features

LLM Integration

Work with OpenAI, Anthropic, Google, and other providers seamlessly:

from agnt5 import Agent

# OpenAI GPT-4
openai_agent = Agent(
    name="gpt4_assistant",
    model="gpt-4o",
    llm_provider="openai",
    api_key="your-openai-key"
)

# Anthropic Claude
claude_agent = Agent(
    name="claude_assistant", 
    model="claude-3-sonnet-20240229",
    llm_provider="anthropic",
    api_key="your-anthropic-key"
)

# Automatic model detection
auto_agent = Agent(
    name="smart_assistant",
    model="gpt-4o"  # Provider automatically detected
)

Memory Management

Maintain conversation history and learned context automatically:

from agnt5 import Agent, Memory

# Agent with persistent memory
memory = Memory(
    vector_store="pinecone",  # or "chroma", "weaviate"
    embedding_model="text-embedding-ada-002"
)

agent = Agent(
    name="memory_agent",
    model="gpt-4o",
    memory=memory,
    system_prompt="You remember previous conversations and learn from them."
)

# Conversations are automatically saved and retrieved
response1 = await agent.run("My name is John and I like Python programming")
response2 = await agent.run("What's my name and favorite language?")
# Agent remembers: "Your name is John and your favorite language is Python"

Specialization

Configure agents for specific roles and expertise:

# Customer service specialist
customer_agent = Agent(
    name="customer_service",
    model="gpt-4o",
    system_prompt="""
    You are a customer service representative for ACME Corp.
    Always be polite, helpful, and professional.
    If you can't resolve an issue, escalate to a human agent.
    """,
    temperature=0.7
)

# Data analysis specialist  
analyst_agent = Agent(
    name="data_analyst",
    model="gpt-4o",
    system_prompt="""
    You are a data analyst expert.
    Provide accurate statistical analysis and insights.
    Always show your calculations and reasoning.
    """, 
    temperature=0.1  # Low temperature for consistent analysis
)

# Creative writing specialist
creative_agent = Agent(
    name="creative_writer",
    model="gpt-4o",
    system_prompt="""
    You are a creative writer specializing in engaging content.
    Use vivid language and compelling narratives.
    """,
    temperature=0.9  # High temperature for creativity
)

Multi-Agent Coordination

Enable collaboration between multiple agents:

from agnt5 import Agent

class MultiAgentTeam:
    def __init__(self):
        self.researcher = Agent(
            name="researcher",
            model="gpt-4o",
            system_prompt="Research topics thoroughly and provide factual information."
        )
        
        self.writer = Agent(
            name="writer", 
            model="gpt-4o",
            system_prompt="Write engaging content based on research provided."
        )
        
        self.editor = Agent(
            name="editor",
            model="gpt-4o", 
            system_prompt="Edit and improve content for clarity and impact."
        )
    
    async def create_content(self, topic: str) -> dict:
        """Collaborative content creation with multiple agents."""
        # Research phase
        research = await self.researcher.run(f"Research comprehensive information about: {topic}")
        
        # Writing phase  
        draft = await self.writer.run(f"Write an article about {topic} using this research: {research}")
        
        # Editing phase
        final_content = await self.editor.run(f"Edit and improve this article: {draft}")
        
        return {
            "topic": topic,
            "research": research,
            "draft": draft,
            "final_content": final_content
        }

Interactive Capabilities

Support human-in-the-loop patterns:

async def interactive_support(user_query: str) -> dict:
    """Interactive customer support with human escalation."""
    agent = Agent(
        name="support_agent",
        model="gpt-4o",
        system_prompt="Provide customer support. If complex, recommend human agent."
    )
    
    # Initial AI response
    ai_response = await agent.run(user_query)
    
    # Check if human escalation is needed
    if "escalate" in ai_response.lower() or "human agent" in ai_response.lower():
        return {
            "ai_response": ai_response,
            "escalated": True,
            "human_required": True
        }
    else:
        return {
            "ai_response": ai_response,
            "escalated": False, 
            "resolved": True
        }

Creating Agents

Basic Agent

Create a simple agent with default settings:

from agnt5 import Agent

agent = Agent(
    name="general_assistant",
    model="gpt-4o"
)

# Simple interaction
response = await agent.run("Explain quantum computing in simple terms")
print(response)

Configured Agent

Create an agent with specific configuration:

from agnt5 import Agent, AgentConfig

config = AgentConfig(
    temperature=0.8,
    max_tokens=1000,
    top_p=0.9,
    presence_penalty=0.1,
    frequency_penalty=0.1
)

agent = Agent(
    name="configured_assistant",
    model="gpt-4o",
    config=config,
    system_prompt="You are a helpful AI assistant specialized in technical topics.",
    llm_provider="openai",
    api_key="your-api-key"
)

Agent with Tools

Create an agent that can use tools:

from agnt5 import Agent, tool

@tool
def calculate(expression: str) -> float:
    """Calculate mathematical expressions safely."""
    return eval(expression)  # In production, use a safe evaluator

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Implement web search logic
    return f"Search results for: {query}"

# Agent with tools
agent = Agent(
    name="tool_agent",
    model="gpt-4o",
    tools=[calculate, search_web],
    system_prompt="You can calculate and search for information to help users."
)

# Agent will automatically use tools when needed
response = await agent.run("What's the square root of 144 and find recent news about AI?")

Agent with Memory

Create an agent with persistent memory:

from agnt5 import Agent, Memory

memory = Memory(
    vector_store="chroma",
    embedding_model="text-embedding-ada-002",
    memory_size=1000  # Remember last 1000 interactions
)

agent = Agent(
    name="memory_agent",
    model="gpt-4o", 
    memory=memory,
    system_prompt="Remember our conversation history and provide personalized responses."
)

# First conversation
await agent.run("I'm working on a Python web app using FastAPI")

# Later conversation - agent remembers context
response = await agent.run("How can I add authentication to my app?")
# Agent knows you're using FastAPI and provides relevant FastAPI auth advice

Agent Interactions

Simple Interaction

Basic question-and-answer interaction:

agent = Agent(name="qa_agent", model="gpt-4o")

response = await agent.run("What are the benefits of renewable energy?")
print(response)

Conversation Flow

Maintain context across multiple interactions:

agent = Agent(
    name="conversation_agent",
    model="gpt-4o",
    system_prompt="You maintain context across our conversation."
)

# Build conversation context
response1 = await agent.run("I'm planning a trip to Japan")
response2 = await agent.run("What should I pack for winter?")  # Agent knows about Japan trip
response3 = await agent.run("How about food recommendations?")  # Agent knows about Japan winter trip

Structured Output

Get structured responses from agents:

import json
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    sentiment: str
    confidence: float
    key_topics: list[str]
    summary: str

agent = Agent(
    name="analysis_agent",
    model="gpt-4o",
    system_prompt="Analyze text and return structured JSON results."
)

text_to_analyze = "I love this new product! It's amazing and works perfectly."

response = await agent.run(f"""
Analyze this text and return a JSON object with sentiment, confidence, key_topics, and summary:
{text_to_analyze}
""")

# Parse structured response
result = AnalysisResult(**json.loads(response))
print(f"Sentiment: {result.sentiment}, Confidence: {result.confidence}")

Advanced Agent Features

Reflection and Self-Evaluation

Agents can reflect on and improve their responses:

agent = Agent(
    name="reflective_agent",
    model="gpt-4o",
    system_prompt="You provide thoughtful responses and can reflect on your work."
)

# Initial response
initial_response = await agent.run("Explain machine learning to a 10-year-old")

# Agent reflects on response quality
reflection = await agent.reflect_on_response(initial_response)
print(f"Reflection: {reflection}")

# Agent self-evaluates
evaluation = await agent.self_evaluate(initial_response)
print(f"Self-evaluation score: {evaluation['score']}")

# Agent improves response
improved_response = await agent.improve_response(initial_response, reflection)
print(f"Improved response: {improved_response}")

Learning from Errors

Agents can learn from mistakes and improve:

async def error_learning_example():
    agent = Agent(
        name="learning_agent",
        model="gpt-4o",
        memory=Memory()  # Enable memory for learning
    )
    
    try:
        # Attempt a task that might fail
        response = await agent.run("Calculate the revenue for Q4 2023")
        # This fails because agent doesn't have access to revenue data
    except Exception as error:
        # Agent learns from the error
        learning_result = await agent.learn_from_error(
            error=str(error),
            context="User asked for revenue calculation without providing data"
        )
        
        # Try again with learned knowledge
        improved_response = await agent.run(
            "I need to calculate revenue but don't have the data. What should I ask for?"
        )
        return improved_response

Custom Agent Behaviors

Implement specialized agent behaviors:

class SpecializedAgent(Agent):
    """Custom agent with specialized behaviors."""
    
    def __init__(self, domain: str):
        super().__init__(
            name=f"{domain}_specialist",
            model="gpt-4o",
            system_prompt=f"You are an expert in {domain}."
        )
        self.domain = domain
        self.expertise_level = "expert"
    
    async def analyze_with_expertise(self, content: str) -> dict:
        """Analyze content using domain expertise."""
        prompt = f"""
        As an expert in {self.domain}, analyze this content:
        {content}
        
        Provide expert-level insights specific to {self.domain}.
        """
        
        analysis = await self.run(prompt)
        
        return {
            "domain": self.domain,
            "expertise_level": self.expertise_level,
            "analysis": analysis,
            "confidence": "high"
        }
    
    async def provide_recommendations(self, problem: str) -> list[str]:
        """Provide domain-specific recommendations."""
        prompt = f"""
        Given this problem in {self.domain}: {problem}
        
        Provide 3-5 specific recommendations as a JSON list.
        """
        
        response = await self.run(prompt)
        return json.loads(response)

# Usage
finance_agent = SpecializedAgent("finance")
recommendations = await finance_agent.provide_recommendations("How to reduce operational costs?")

Integration Patterns

Agents in Tasks

Use agents within durable tasks:

from agnt5 import task, Agent
from datetime import datetime

@task
def ai_powered_classification(text: str) -> dict:
    """Classify text using AI agent with durability."""
    agent = Agent(
        name="classifier",
        model="gpt-4o",
        system_prompt="Classify text into predefined categories."
    )
    
    classification = await agent.run(f"Classify this text: {text}")
    
    return {
        "text": text,
        "classification": classification,
        "timestamp": datetime.now().isoformat()
    }

# Task automatically retries on failure, preserving agent state
result = await ai_powered_classification("This is a customer complaint about billing")

Agents in Workflows

Orchestrate multiple agents within workflows:

from agnt5 import workflow, Agent

@workflow
async def content_review_workflow(content: str) -> dict:
    """Multi-agent content review workflow."""
    
    # Fact-checking agent
    fact_checker = Agent(
        name="fact_checker",
        model="gpt-4o",
        system_prompt="Verify facts and identify potential misinformation."
    )
    
    # Style reviewer agent
    style_reviewer = Agent(
        name="style_reviewer", 
        model="gpt-4o",
        system_prompt="Review content for style, tone, and readability."
    )
    
    # Legal compliance agent
    legal_reviewer = Agent(
        name="legal_reviewer",
        model="gpt-4o",
        system_prompt="Review content for legal compliance and potential issues."
    )
    
    # Run reviews in parallel
    fact_check = await fact_checker.run(f"Fact-check this content: {content}")
    style_review = await style_reviewer.run(f"Review style and tone: {content}")
    legal_review = await legal_reviewer.run(f"Review for legal compliance: {content}")
    
    return {
        "content": content,
        "fact_check": fact_check,
        "style_review": style_review,
        "legal_review": legal_review,
        "approved": all([
            "approved" in fact_check.lower(),
            "approved" in style_review.lower(), 
            "approved" in legal_review.lower()
        ])
    }

Agent Tool Integration

Create tools that use agents internally:

from agnt5 import tool, Agent
import json

@tool
def smart_summarizer(text: str, style: str = "professional") -> str:
    """Summarize text using AI with different styles."""
    agent = Agent(
        name="summarizer",
        model="gpt-4o",
        system_prompt=f"Summarize text in a {style} style."
    )
    
    return await agent.run(f"Summarize this text: {text}")

@tool  
def sentiment_analyzer(text: str) -> dict:
    """Analyze sentiment using AI."""
    agent = Agent(
        name="sentiment_agent",
        model="gpt-4o",
        system_prompt="Analyze sentiment and return structured data."
    )
    
    response = await agent.run(f"""
    Analyze sentiment of this text and return JSON:
    {text}
    
    Format: {{"sentiment": "positive/negative/neutral", "confidence": 0.0-1.0}}
    """)
    
    return json.loads(response)

# Other agents can use these tools
main_agent = Agent(
    name="main_agent",
    model="gpt-4o",
    tools=[smart_summarizer, sentiment_analyzer]
)

Common Patterns

Customer Support Agent

Intelligent customer service with escalation:

class CustomerSupportAgent:
    def __init__(self):
        self.agent = Agent(
            name="support_agent",
            model="gpt-4o",
            system_prompt="""
            You are a customer support agent for TechCorp.
            
            Guidelines:
            - Be helpful, polite, and professional
            - If you can't resolve an issue, recommend human escalation
            - Always ask for order/account numbers when relevant
            - Provide step-by-step solutions when possible
            """
        )
        self.escalation_threshold = 0.3
    
    async def handle_inquiry(self, customer_message: str, customer_id: str = None) -> dict:
        """Handle customer inquiry with potential escalation."""
        # Analyze inquiry complexity
        complexity_prompt = f"""
        Rate the complexity of this customer inquiry from 0-1:
        {customer_message}
        
        Return only a number between 0 and 1.
        """
        
        complexity_response = await self.agent.run(complexity_prompt)
        complexity = float(complexity_response.strip())
        
        if complexity > self.escalation_threshold:
            return {
                "response": "This inquiry requires specialized assistance. Let me connect you with a human agent.",
                "escalated": True,
                "complexity": complexity
            }
        
        # Handle with AI
        response = await self.agent.run(f"""
        Customer inquiry: {customer_message}
        Customer ID: {customer_id or "Not provided"}
        
        Provide a helpful response.
        """)
        
        return {
            "response": response,
            "escalated": False,
            "complexity": complexity
        }

Research and Analysis Agent

Comprehensive research with multiple sources:

from datetime import datetime

class ResearchAgent:
    def __init__(self):
        self.agent = Agent(
            name="research_agent",
            model="gpt-4o",
            tools=[web_search, academic_search, news_search],
            system_prompt="""
            You are a research agent that conducts thorough analysis.
            Always cite sources and provide balanced perspectives.
            """
        )
    
    async def comprehensive_research(self, topic: str) -> dict:
        """Conduct comprehensive research on a topic."""
        research_plan = await self.agent.run(f"""
        Create a research plan for: {topic}
        
        Include:
        1. Key questions to investigate
        2. Types of sources to search
        3. Analysis framework
        """)
        
        # Execute research based on plan
        findings = await self.agent.run(f"""
        Execute this research plan: {research_plan}
        
        Topic: {topic}
        
        Use available tools to gather information and provide comprehensive analysis.
        """)
        
        return {
            "topic": topic,
            "research_plan": research_plan,
            "findings": findings,
            "timestamp": datetime.now().isoformat()
        }

Content Creation Pipeline

Multi-agent content creation system:

class ContentCreationPipeline:
    def __init__(self):
        self.researcher = Agent(
            name="researcher",
            model="gpt-4o",
            system_prompt="Research topics and gather factual information."
        )
        
        self.writer = Agent(
            name="writer",
            model="gpt-4o", 
            system_prompt="Create engaging, well-structured content."
        )
        
        self.editor = Agent(
            name="editor",
            model="gpt-4o",
            system_prompt="Edit content for clarity, flow, and impact."
        )
        
        self.seo_optimizer = Agent(
            name="seo_optimizer",
            model="gpt-4o",
            system_prompt="Optimize content for search engines while maintaining quality."
        )
    
    async def create_article(self, topic: str, target_audience: str) -> dict:
        """Complete content creation pipeline."""
        # Research phase
        research = await self.researcher.run(f"""
        Research comprehensive information about: {topic}
        Target audience: {target_audience}
        
        Provide factual information, statistics, and key points.
        """)
        
        # Writing phase
        draft = await self.writer.run(f"""
        Write an engaging article about: {topic}
        Target audience: {target_audience}
        Research: {research}
        
        Create compelling, well-structured content.
        """)
        
        # Editing phase
        edited = await self.editor.run(f"""
        Edit this article for clarity and impact:
        {draft}
        
        Improve flow, readability, and engagement.
        """)
        
        # SEO optimization
        optimized = await self.seo_optimizer.run(f"""
        Optimize this article for SEO:
        {edited}
        
        Topic: {topic}
        
        Add relevant keywords and improve search visibility.
        """)
        
        return {
            "topic": topic,
            "target_audience": target_audience,
            "research": research,
            "draft": draft,
            "edited": edited,
            "final_content": optimized
        }

Performance Considerations

Model Selection

Choose appropriate models for different use cases:

# Fast responses for simple tasks
quick_agent = Agent(
    name="quick_responder",
    model="gpt-3.5-turbo",  # Faster, lower cost
    temperature=0.7
)

# High-quality responses for complex tasks  
quality_agent = Agent(
    name="quality_responder",
    model="gpt-4o",  # Higher quality, slower
    temperature=0.3
)

# Specialized models for specific domains
code_agent = Agent(
    name="code_assistant", 
    model="gpt-4o",  # Good for code
    system_prompt="You are a programming expert."
)

Caching and Memory

Optimize agent performance with smart caching:

from agnt5 import Agent, Memory

# Shared memory for similar tasks
shared_memory = Memory(
    vector_store="chroma",
    embedding_model="text-embedding-ada-002"
)

# Multiple agents sharing knowledge
agent1 = Agent(name="agent1", model="gpt-4o", memory=shared_memory)
agent2 = Agent(name="agent2", model="gpt-4o", memory=shared_memory)

# Knowledge learned by agent1 is available to agent2
await agent1.run("Remember: our company policy is X")
response = await agent2.run("What's our company policy?")  # Knows about policy X

Batch Processing

Process multiple requests efficiently:

import asyncio

async def batch_agent_processing(requests: list[str]) -> list[str]:
    """Process multiple requests with a single agent efficiently."""
    agent = Agent(name="batch_agent", model="gpt-4o")
    
    # Process in parallel with controlled concurrency
    semaphore = asyncio.Semaphore(5)  # Limit concurrent requests
    
    async def process_request(request):
        async with semaphore:
            return await agent.run(request)
    
    results = await asyncio.gather(*[
        process_request(req) for req in requests
    ])
    
    return results

Best Practices

Agent Design:

• Define specific roles and responsibilities
• Choose appropriate models for task complexity
• Provide clear system prompts and context
• Adjust temperature for creativity vs consistency

Memory Management:

• Store relevant conversation history
• Configure appropriate memory sizes
• Handle sensitive information properly
• Implement cleanup for long-running agents

Error Handling:

• Handle API failures gracefully
• Implement retry strategies
• Use fallback models when needed
• Track performance and error rates

Cost Optimization:

• Select cost-effective models
• Optimize prompts to reduce tokens
• Cache repeated queries
• Batch requests to reduce overhead

Next Steps

Agents provide intelligence and reasoning to your applications. Learn how they integrate with other AGNT5 components:

• Tasks - Create durable AI-powered functions
• Workflows - Orchestrate multi-agent operations
• Tools - Give agents access to external systems
• Entities - Maintain persistent agent state and memory

Ready to build your first intelligent agent? Check out the Quick Start Guide to get started with AI-powered applications.