Создание вашего первого AI агента: Пошаговое руководство

1
2
3
4
5
6
7
8
9

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

# Создание виртуального окружения
python -m venv venv
source venv/bin/activate  # На Windows: venv\Scripts\activate

# Установка необходимых пакетов
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_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:
        """Обработка ввода пользователя и определение необходимого действия"""
        system_prompt = """Вы - агент-помощник исследователь. 
        Получив ввод пользователя, определите, какое действие нужно предпринять.
        Доступные действия: web_search, summarize, analyze
        Верните только название действия."""
        
        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]:
        """Простая функция веб-поиска (макет реализации)"""
        # В реальной реализации вы бы использовали настоящий API поиска
        print(f"🔍 Поиск по запросу: {query}")
        
        # Макет результатов поиска
        mock_results = [
            f"Статья 1 о {query}: Lorem ipsum dolor sit amet...",
            f"Исследовательская работа о {query}: Consectetur adipiscing elit...",
            f"Новостная статья о {query}: Sed do eiusmod tempor..."
        ]
        
        return mock_results
    
    def summarize_content(self, content: List[str]) -> str:
        """Суммирование исследовательского контента с помощью ИИ"""
        combined_content = "\n".join(content)
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system", 
                    "content": "Суммируйте следующий контент в ясной, структурированной форме."
                },
                {"role": "user", "content": combined_content}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def analyze_information(self, summary: str) -> str:
        """Анализ информации и предоставление инсайтов"""
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system", 
                    "content": "Проанализируйте следующую информацию и предоставьте ключевые инсайты, тренды и выводы."
                },
                {"role": "user", "content": summary}
            ],
            temperature=0.4
        )
        
        return response.choices[0].message.content
    
    def act(self, action: str, query: str) -> str:
        """Выполнение определенного действия"""
        if action == "web_search":
            results = self.web_search(query)
            summary = self.summarize_content(results)
            analysis = self.analyze_information(summary)
            
            # Сохранение в памяти
            self.memory.append({
                "query": query,
                "results": results,
                "summary": summary,
                "analysis": analysis,
                "timestamp": time.time()
            })
            
            return f"""
## Результаты исследования по теме: {query}

### Краткое содержание:
{summary}

### Анализ:
{analysis}
            """
        
        return "Запрошено неизвестное действие."
    
    def run(self, user_input: str) -> str:
        """Основной цикл выполнения агента"""
        print(f"🤖 Агент получил: {user_input}")
        
        # Восприятие окружения (ввод пользователя)
        action = self.perceive(user_input)
        print(f"🧠 Агент решил: {action}")
        
        # Действие на основе восприятия
        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):
        """Добавление нового опыта в память"""
        experience["timestamp"] = datetime.now().isoformat()
        self.short_term_memory.append(experience)
        
        # Перенос в долгосрочную память, если кратковременная заполнена
        if len(self.short_term_memory) > 10:
            self.transfer_to_long_term()
    
    def transfer_to_long_term(self):
        """Перенос опыта в долгосрочную память"""
        self.long_term_memory.extend(self.short_term_memory)
        self.short_term_memory = []
        self.save_memory()
    
    def save_memory(self):
        """Сохранение памяти в файл"""
        with open(self.memory_file, 'w', encoding='utf-8') as f:
            json.dump(self.long_term_memory, f, indent=2, ensure_ascii=False)
    
    def load_memory(self) -> list:
        """Загрузка памяти из файла"""
        if os.path.exists(self.memory_file):
            with open(self.memory_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return []
    
    def recall_similar(self, query: str, top_k: int = 3) -> list:
        """Вспоминание похожего прошлого опыта"""
        # Простое сопоставление ключевых слов (в продакшене используйте эмбеддинги)
        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))
        
        # Возврат топ k наиболее релевантных воспоминаний
        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("🤖 Агент помощник исследователь")
    print("=" * 40)
    
    agent = ResearchAgent()
    memory = AgentMemory()
    
    while True:
        try:
            user_input = input("\n💬 Что вы хотели бы исследовать? (или 'quit' для выхода): ")
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("👋 До свидания!")
                break
            
            if not user_input.strip():
                continue
            
            # Проверка на наличие похожих прошлых исследований
            similar_memories = memory.recall_similar(user_input)
            if similar_memories:
                print(f"🧠 Я нашел {len(similar_memories)} похожих прошлых исследований.")
            
            # Обработка запроса
            result = agent.run(user_input)
            print(result)
            
            # Сохранение в память
            memory.add_to_memory({
                "query": user_input,
                "result": result,
                "action_taken": "research"
            })
            
        except KeyboardInterrupt:
            print("\n👋 До свидания!")
            break
        except Exception as e:
            print(f"❌ Ошибка: {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):
        """Тест корректной инициализации агента"""
        self.assertIsNotNone(self.agent.client)
        self.assertIsInstance(self.agent.tools, dict)
        self.assertEqual(len(self.agent.memory), 0)
    
    def test_web_search(self):
        """Тест функциональности веб-поиска"""
        results = self.agent.web_search("искусственный интеллект")
        self.assertIsInstance(results, list)
        self.assertGreater(len(results), 0)
    
    def test_perceive_function(self):
        """Тест функции восприятия"""
        action = self.agent.perceive("Расскажи мне о машинном обучении")
        self.assertIn(action.lower(), ["web_search", "summarize", "analyze"])

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

1
2
3
4
5
6
7
8

# Убедитесь, что ваше виртуальное окружение активировано
source venv/bin/activate

# Запуск агента
python main.py

# Запуск тестов
python test_agent.py

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

def calculate(self, expression: str) -> str:
    """Функция безопасного калькулятора"""
    try:
        # Используйте eval осторожно в продакшене!
        result = eval(expression)
        return f"Результат: {result}"
    except:
        return "Неверное вычисление"

def get_weather(self, location: str) -> str:
    """Получение информации о погоде (макет)"""
    return f"Погода в {location}: Солнечно, 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]:
    """Настоящий веб-поиск используя 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:
        # Добавление контекста из истории разговоров
        context = "\n".join(self.conversation_history[-5:])  # Последние 5 сообщений
        
        full_input = f"Контекст: {context}\nПользователь: {user_input}"
        result = self.run(full_input)
        
        # Обновление истории разговоров
        self.conversation_history.append(f"Пользователь: {user_input}")
        self.conversation_history.append(f"Агент: {result}")
        
        return result