Практикум: Создание агента-аналитика данных

70 минут Урок 30

Практикум: Создание агента-аналитика данных на базе Gemini 3

Приветствую, коллеги. Мы добрались до одного из самых захватывающих модулей курса. Сегодня мы не просто изучаем API, мы строим полноценного цифрового сотрудника.

Давайте честно: работа аналитика данных на 60% состоит из рутины. Загрузить CSV, очистить заголовки, проверить типы данных, построить базовые гистограммы, чтобы просто понять, на что мы смотрим. Gemini 3 обладает достаточным контекстным окном и логическими способностями, чтобы взять эту рутину на себя.

В этом уроке мы создадим Автономного Агента-Аналитика. Это не просто чат-бот, который «помнит» таблицу. Это система, которая:

Планирует действия (ReAct паттерн).
Пишет исполняемый код на Python (Pandas/Matplotlib).
Выполняет этот код в реальном времени.
Исправляет собственные ошибки, если код упал.
Интерпретирует результаты и строит графики.

Готовы? Давайте превратим сырые данные в инсайты.

Архитектура: Думай, Кодь, Смотри

Прежде чем писать код, давайте разберем архитектуру. Мы будем использовать паттерн, который часто называют Code Interpreter или Advanced Data Analysis.

Ключевое отличие от обычного RAG (Retrieval Augmented Generation) в том, что мы не скармливаем модели текст документа. Если у вас CSV на 500 МБ, ни одно контекстное окно не справится с этим эффективно (и дешево). Вместо этого мы даем агенту «инструмент» — возможность выполнять Python-код.

Цикл работы нашего агента:

Вход: Пользователь просит: «Покажи тренд продаж по месяцам».
Мысль (Thought): Агент решает: «Мне нужно загрузить файл, преобразовать колонку даты и сгруппировать продажи».
Действие (Action): Агент генерирует Python-скрипт.
Наблюдение (Observation): Мы (система) выполняем скрипт и возвращаем агенту результат (вывод print() или ошибку).
Ответ (Answer): Агент читает вывод и формулирует ответ пользователю.

Шаг 1: Подготовка среды и данных

Для начала нам нужен «подопытный» набор данных. Мы не будем усложнять и сгенерируем простой CSV с данными о продажах, но с типичными проблемами: даты в виде строк, возможные пропуски.

Также настроим клиент Gemini. Убедитесь, что у вас установлен `google-generativeai` последней версии.

python

import google.generativeai as genai
import pandas as pd
import io
import sys
from typing import Dict, Any

# Настройка API (замените на ваш ключ)
genai.configure(api_key="YOUR_API_KEY")

# 1. Создаем тестовый набор данных
csv_data = """
Date,Product,Category,Sales,Region
2023-01-01,Widget A,Gadgets,100,North
2023-01-02,Widget B,Gadgets,150,North
2023-01-05,Gadget X,Gadgets,200,South
2023-02-01,Widget A,Gadgets,120,North
2023-02-10,SuperTool,Tools,300,East
2023-03-01,Widget B,Gadgets,160,West
2023-03-15,SuperTool,Tools,310,East
N/A,Widget A,Gadgets,90,North
"""

# Сохраняем в файл для эмуляции реальной работы
filename = "sales_data.csv"
with open(filename, "w") as f:
    f.write(csv_data.strip())

print(f"Файл {filename} создан. Готов к анализу.")

Шаг 2: Инструмент выполнения кода (The Execution Engine)

Это сердце нашего агента. Gemini 3 умеет генерировать код, но выполнять его должны мы (или безопасная среда). В Enterprise-решениях критически важно использовать «песочницу» (Sandbox), например, Docker-контейнер или специализированные сервисы вроде E2B, чтобы агент случайно (или намеренно) не удалил системные файлы.

Для учебных целей мы создадим локальный исполнитель кода с использованием `exec()`, перехватывая стандартный вывод (stdout). Внимание: Не используйте `exec()` на продакшн-серверах без изоляции!

python

class PythonExecutor:
    def __init__(self, working_dir="."):
        self.locals = {}
        self.working_dir = working_dir
    
    def execute(self, code: str) -> str:
        """
        Выполняет Python код и возвращает результат (stdout) или ошибку.
        Состояние (переменные) сохраняется между вызовами в self.locals.
        """
        # Перехват stdout для получения результатов print()
        old_stdout = sys.stdout
        redirected_output = io.StringIO()
        sys.stdout = redirected_output
        
        try:
            # Оборачиваем выполнение, чтобы сохранить контекст переменных
            exec(code, {}, self.locals)
            result = redirected_output.getvalue()
            if not result:
                result = "Код выполнен успешно, но ничего не выведено. Используйте print(), чтобы увидеть результат."
            return result
        except Exception as e:
            return f"Error executing code: {str(e)}"
        finally:
            sys.stdout = old_stdout

# Тест исполнителя
executor = PythonExecutor()
print(executor.execute("import pandas as pd\ndf = pd.DataFrame({'a': [1, 2]})\nprint(df)"))
# Следующий вызов помнит 'df'
print(executor.execute("print(df['a'].sum())"))

Шаг 3: Определение инструмента для Gemini

Теперь нужно «объяснить» модели, что у нее есть этот инструмент. В Gemini API мы используем `tools` и Function Calling. Мы опишем функцию `run_python`, которую модель сможет вызывать.

Обратите внимание на описание (docstring). Чем точнее вы опишете, для чего нужен инструмент и как им пользоваться (например, «всегда используй print для вывода»), тем умнее будет вести себя агент.

python

tool_code_execution = {
    "function_declarations": [
        {
            "name": "run_python",
            "description": "Executes Python code to analyze data. Use pandas for data manipulation. ALWAYS access the file 'sales_data.csv'. The variable state persists between calls. Use print() to output any results you want to see.",
            "parameters": {
                "type": "OBJECT",
                "properties": {
                    "code": {
                        "type": "STRING",
                        "description": "The valid Python code to execute."
                    }
                },
                "required": ["code"]
            }
        }
    ]
}

Шаг 4: Системный промпт (Persona)

Настройка поведения агента делается через System Instruction. Для аналитика данных нам нужно задать несколько жестких правил:

Первый шаг — разведка. Никогда не угадывай названия колонок. Сначала загрузи файл и сделай `df.info()` или `df.head()`.
Устойчивость. Если код вернул ошибку, проанализируй её, исправь код и запусти снова.
Контекст. Помни, что `df` уже загружен в память после первого шага.

python

system_instruction = """
Ты - эксперт Data Analyst. Твоя задача - отвечать на вопросы пользователя, анализируя CSV файл 'sales_data.csv'.

ПРАВИЛА:
1. У тебя есть доступ к инструменту выполнения Python кода. ИСПОЛЬЗУЙ ЕГО.
2. В самом начале всегда проверь структуру данных: загрузи файл и выведи df.info() и df.head().
3. Если пользователь просит график, напиши код для его создания (matplotlib/seaborn), но помни, что ты не можешь показать его визуально в консоли. Вместо этого скажи, что график построен (или сохрани его в файл).
4. Никогда не выдумывай данные. Опирайся только на результаты выполнения кода.
5. Если код выдал ошибку, попробуй исправить её и запустить снова.
"""

Шаг 5: Сборка цикла (The Loop)

Самое сложное в агентах — это управление потоком разговора. Gemini API stateless (без сохранения состояния на сервере в базовом варианте), поэтому нам нужно самим управлять историей чата (`ChatSession`).

Когда модель решает вызвать функцию, она возвращает специальный объект `function_call`. Мы должны:

Отловить этот вызов.
Взять аргументы (код Python).
Передать их нашему `PythonExecutor`.
Вернуть результат обратно в модель как `function_response`.

Ниже представлен упрощенный цикл работы агента.

python

model = genai.GenerativeModel(
    model_name='gemini-1.5-pro-latest', # Используем мощную модель для кодинга
    tools=[tool_code_execution],
    system_instruction=system_instruction
)

# Инициализация чата
chat = model.start_chat(enable_automatic_function_calling=False)
executor = PythonExecutor() # Наш песочный ящик

def agent_step(user_input, max_turns=5):
    # Отправляем сообщение пользователя
    response = chat.send_message(user_input)
    
    turns = 0
    while turns < max_turns:
        # Проверяем, хочет ли модель вызвать функцию
        part = response.parts[0]
        
        if fn := part.function_call:
            print(f"\n[AGENT] 🤖 Пишет код...\n----------\n{fn.args['code']}\n----------")
            
            # 1. Выполняем код
            execution_result = executor.execute(fn.args['code'])
            print(f"[SYSTEM] ⚙️ Результат выполнения:\n{execution_result[:200]}... (обрезано)")
            
            # 2. Возвращаем результат модели
            # Важно: нужно правильно сформировать ответ для API
            response = chat.send_message(
                genai.protos.Content(
                    parts=[genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name='run_python',
                            response={'result': execution_result}
                        )
                    )]
                )
            )
            turns += 1
        else:
            # Если функции нет, значит модель дала текстовый ответ
            return response.text
            
    return "Превышен лимит шагов агента. Задача слишком сложная или возникла циклическая ошибка."

# ЗАПУСК
query = "Проанализируй файл. Какие категории товаров принесли больше всего выручки?"
print(f"USER: {query}")
final_answer = agent_step(query)
print(f"\nAGENT ANSWER: {final_answer}")

Разбор полетов: Что произошло?

Если вы запустите этот код, вы увидите магию в действии:

Первый проход: Агент сгенерирует код: `import pandas as pd; df = pd.read_csv('sales_data.csv'); print(df.info())`.
Реакция системы: Наш `PythonExecutor` выполнит это и вернет структуру таблицы (колонки Date, Product, Sales...).
Второй проход: Агент «увидит» структуру, поймет, что нужно сгруппировать по `Category` и просуммировать `Sales`. Он сгенерирует новый код.
Финал: Получив суммы, он сформирует текстовый ответ: «Категория Tools принесла больше всего выручки...».

Это и есть суть автономного агента: он сам определяет шаги на основе обратной связи от среды.

Упражнение

Модифицируйте системный промпт и инструмент PythonExecutor так, чтобы агент мог сохранять графики. Задача: 1. Добавьте в 'run_python' возможность (в описании), что код может сохранять файлы PNG. 2. Попросите агента: 'Построй круговую диаграмму распределения продаж по регионам и сохрани её'.

Решение:

1. Обновите описание инструмента: '...If creating charts, save them as distinct filenames (e.g., chart.png) using plt.savefig()'.
2. В executor добавьте import matplotlib.pyplot as plt в locals или просто убедитесь, что библиотека установлена в среде.
3. Запрос агенту: 'Построй круговую диаграмму продаж по регионам'.

Агент должен сгенерировать код:
plt.figure()
df.groupby('Region')['Sales'].sum().plot.pie()
plt.savefig('region_sales.png')
print('График сохранен в region_sales.png')

Обработка ошибок и самокоррекция

Одна из самых сильных сторон LLM-агентов — способность исправлять собственный код. Представьте, что в нашем CSV файле в колонке `Sales` есть значение 'N/A' (мы специально добавили его в шаге 1).

Если агент попытается сделать `df['Sales'].sum()`, Pandas может выдать ошибку (TypeError) или склеить строки, если не преобразовать типы.

Когда `executor` вернет текст ошибки (Traceback), Gemini 3 прочитает его в следующем шаге диалога. Благодаря системному промпту и обучению, модель поймет: «Ага, колонка Sales имеет тип object, нужно сделать pd.to_numeric c errors='coerce'». Она перепишет код и запустит его снова. Вам не нужно писать `try-except` блоки на каждый чих — агент сам пишет обработчики по ситуации.

Вопрос

Почему при создании агента-аналитика мы используем подход Code Execution, а не просто скармливаем содержимое CSV файла в промпт?

Потому что Gemini не умеет читать CSV текст.

Потому что выполнение кода дешевле, точнее для математики и не зависит от размера файла (контекстного окна) так сильно, как чтение raw text.

Потому что Python код выполняется быстрее, чем генерация токенов.

Это требование безопасности Google Cloud.

Чтобы агент мог удалить файл после чтения.

Заключение

Мы только что создали прототип того, что в Enterprise-сегменте продается за большие деньги. Ваш агент умеет:

Работать с данными любого объема (через Pandas).
Исправлять свои ошибки.
Отвечать на бизнес-вопросы, переводя естественный язык в SQL или Python-код.

В следующем уроке мы разберем, как масштабировать это решение: добавим постоянную память (Vector Store) для хранения документации по специфичным корпоративным данным, чтобы агент понимал, что такое KPI «Gross Margin» в контексте вашей компании.

Предыдущий урок Следующий урок