Building Your First AI Agent: A Step-by-Step Guide

1
2
3
4
5
6
7
8
9

mkdir my-first-ai-agent
cd my-first-ai-agent

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install required packages
pip install openai python-dotenv requests beautifulsoup4

1

OPENAI_API_KEY=your_api_key_here

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

# agent.py
import os
import requests
from bs4 import BeautifulSoup
from openai import OpenAI
from dotenv import load_dotenv
import time
from typing import List, Dict

# Load environment variables
load_dotenv()

class ResearchAgent:
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.memory = []
        self.tools = {
            "web_search": self.web_search,
            "summarize": self.summarize_content,
            "analyze": self.analyze_information
        }
    
    def perceive(self, user_input: str) -> str:
        """Process user input and determine the action needed"""
        system_prompt = """You are a research assistant agent. 
        Given user input, determine what action to take.
        Available actions: web_search, summarize, analyze
        Return only the action name."""
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_input}
            ],
            temperature=0.1
        )
        
        return response.choices[0].message.content.strip()
    
    def web_search(self, query: str) -> List[str]:
        """Simple web search function (mock implementation)"""
        # In a real implementation, you'd use a proper search API
        print(f"🔍 Searching for: {query}")
        
        # Mock search results
        mock_results = [
            f"Article 1 about {query}: Lorem ipsum dolor sit amet...",
            f"Research paper on {query}: Consectetur adipiscing elit...",
            f"News article about {query}: Sed do eiusmod tempor..."
        ]
        
        return mock_results
    
    def summarize_content(self, content: List[str]) -> str:
        """Summarize research content using AI"""
        combined_content = "\n".join(content)
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system", 
                    "content": "Summarize the following content in a clear, structured way."
                },
                {"role": "user", "content": combined_content}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def analyze_information(self, summary: str) -> str:
        """Analyze and provide insights on the information"""
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system", 
                    "content": "Analyze the following information and provide key insights, trends, and conclusions."
                },
                {"role": "user", "content": summary}
            ],
            temperature=0.4
        )
        
        return response.choices[0].message.content
    
    def act(self, action: str, query: str) -> str:
        """Execute the determined action"""
        if action == "web_search":
            results = self.web_search(query)
            summary = self.summarize_content(results)
            analysis = self.analyze_information(summary)
            
            # Store in memory
            self.memory.append({
                "query": query,
                "results": results,
                "summary": summary,
                "analysis": analysis,
                "timestamp": time.time()
            })
            
            return f"""
## Research Results for: {query}

### Summary:
{summary}

### Analysis:
{analysis}
            """
        
        return "Unknown action requested."
    
    def run(self, user_input: str) -> str:
        """Main agent execution loop"""
        print(f"🤖 Agent received: {user_input}")
        
        # Perceive the environment (user input)
        action = self.perceive(user_input)
        print(f"🧠 Agent decided to: {action}")
        
        # Act based on perception
        result = self.act(action, user_input)
        
        return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# memory.py
import json
import os
from datetime import datetime

class AgentMemory:
    def __init__(self, memory_file="agent_memory.json"):
        self.memory_file = memory_file
        self.short_term_memory = []
        self.long_term_memory = self.load_memory()
    
    def add_to_memory(self, experience: dict):
        """Add new experience to memory"""
        experience["timestamp"] = datetime.now().isoformat()
        self.short_term_memory.append(experience)
        
        # Transfer to long-term memory if short-term is full
        if len(self.short_term_memory) > 10:
            self.transfer_to_long_term()
    
    def transfer_to_long_term(self):
        """Transfer experiences to long-term memory"""
        self.long_term_memory.extend(self.short_term_memory)
        self.short_term_memory = []
        self.save_memory()
    
    def save_memory(self):
        """Save memory to file"""
        with open(self.memory_file, 'w') as f:
            json.dump(self.long_term_memory, f, indent=2)
    
    def load_memory(self) -> list:
        """Load memory from file"""
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r') as f:
                return json.load(f)
        return []
    
    def recall_similar(self, query: str, top_k: int = 3) -> list:
        """Recall similar past experiences"""
        # Simple keyword matching (in production, use embeddings)
        relevant_memories = []
        query_words = set(query.lower().split())
        
        for memory in self.long_term_memory:
            memory_words = set(memory.get('query', '').lower().split())
            overlap = len(query_words & memory_words)
            if overlap > 0:
                relevant_memories.append((overlap, memory))
        
        # Return top k most relevant memories
        relevant_memories.sort(key=lambda x: x[0], reverse=True)
        return [mem[1] for mem in relevant_memories[:top_k]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# main.py
from agent import ResearchAgent
from memory import AgentMemory

def main():
    print("🤖 Research Assistant Agent")
    print("=" * 40)
    
    agent = ResearchAgent()
    memory = AgentMemory()
    
    while True:
        try:
            user_input = input("\n💬 What would you like to research? (or 'quit' to exit): ")
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("👋 Goodbye!")
                break
            
            if not user_input.strip():
                continue
            
            # Check if we have similar past research
            similar_memories = memory.recall_similar(user_input)
            if similar_memories:
                print(f"🧠 I found {len(similar_memories)} similar past research(es).")
            
            # Process the request
            result = agent.run(user_input)
            print(result)
            
            # Save to memory
            memory.add_to_memory({
                "query": user_input,
                "result": result,
                "action_taken": "research"
            })
            
        except KeyboardInterrupt:
            print("\n👋 Goodbye!")
            break
        except Exception as e:
            print(f"❌ Error: {e}")

if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# test_agent.py
import unittest
from agent import ResearchAgent

class TestResearchAgent(unittest.TestCase):
    def setUp(self):
        self.agent = ResearchAgent()
    
    def test_agent_initialization(self):
        """Test if agent initializes correctly"""
        self.assertIsNotNone(self.agent.client)
        self.assertIsInstance(self.agent.tools, dict)
        self.assertEqual(len(self.agent.memory), 0)
    
    def test_web_search(self):
        """Test web search functionality"""
        results = self.agent.web_search("artificial intelligence")
        self.assertIsInstance(results, list)
        self.assertGreater(len(results), 0)
    
    def test_perceive_function(self):
        """Test perception function"""
        action = self.agent.perceive("Tell me about machine learning")
        self.assertIn(action.lower(), ["web_search", "summarize", "analyze"])

if __name__ == '__main__':
    unittest.main()

1
2
3
4
5
6
7
8

# Make sure your virtual environment is activated
source venv/bin/activate

# Run the agent
python main.py

# Run tests
python test_agent.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def calculate(self, expression: str) -> str:
    """Safe calculator function"""
    try:
        # Use eval carefully in production!
        result = eval(expression)
        return f"Result: {result}"
    except:
        return "Invalid calculation"

def get_weather(self, location: str) -> str:
    """Get weather information (mock)"""
    return f"Weather in {location}: Sunny, 22°C"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import requests

def real_web_search(self, query: str) -> List[str]:
    """Real web search using DuckDuckGo API"""
    url = f"https://api.duckduckgo.com/?q={query}&format=json"
    response = requests.get(url)
    data = response.json()
    
    results = []
    for result in data.get('RelatedTopics', [])[:5]:
        if 'Text' in result:
            results.append(result['Text'])
    
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class ConversationAgent(ResearchAgent):
    def __init__(self):
        super().__init__()
        self.conversation_history = []
    
    def run_with_context(self, user_input: str) -> str:
        # Add context from conversation history
        context = "\n".join(self.conversation_history[-5:])  # Last 5 messages
        
        full_input = f"Context: {context}\nUser: {user_input}"
        result = self.run(full_input)
        
        # Update conversation history
        self.conversation_history.append(f"User: {user_input}")
        self.conversation_history.append(f"Agent: {result}")
        
        return result