SuperAGI Implementation for Autonomous AI Agents

SuperAGI for Autonomous AI Agents

SuperAGI — open-source платформа для создания и управления автономными агентами с web-интерфейсом, инструментами, векторной памятью и мониторингом. Предоставляет готовую инфраструктуру: агентный scheduler, marketplace инструментов (Toolkits), интеграции (GitHub, Jira, Slack, Email), телеметрию и контроль бюджета токенов.

Развёртывание SuperAGI

# Docker-развёртывание
git clone https://github.com/TransformerOptimus/SuperAGI.git
cd SuperAGI

# Настройка .env
cp config_template.yaml config.yaml
# Установить: OPENAI_API_KEY, DB credentials, Redis URL

docker-compose up -d

# Доступ к UI: http://localhost:3000

Создание агента через Python API

from superagi.client import SuperAGIClient
from superagi.models.agent import AgentConfig
from superagi.models.agent_schedule import AgentScheduleConfig

client = SuperAGIClient(
    base_url="http://localhost:8001",
    api_key="your-api-key",
)

# Конфигурация агента
agent_config = AgentConfig(
    name="Market Monitor Agent",
    description="Мониторит рыночные новости и генерирует дайджест",
    goals=[
        "Собрать последние новости по заданным тематикам",
        "Проанализировать влияние на бизнес",
        "Создать краткий дайджест для команды",
    ],
    instructions="Используй поиск и скрапинг для сбора актуальной информации. Будь конкретен, избегай общих фраз.",
    agent_type="TASK_QUEUE",
    model="gpt-4o",
    toolkits=["GoogleSearch", "WebScraper", "FileWriter", "SlackToolkit"],
    max_iterations=25,
    exit_criterion="TASK_COMPLETION",
    knowledge_base_ids=[kb.id],   # Привязка к корпоративной базе знаний
)

agent = client.agents.create(config=agent_config)

Запуск и мониторинг агента

# Запуск с параметрами
run = client.agent_runs.start(
    agent_id=agent.id,
    run_config={
        "goals": [
            "Собери новости о рынке SaaS за последние 7 дней",
            "Сфокусируйся на: funding rounds, M&A, product launches",
            "Создай дайджест в формате markdown и отправь в Slack #market-news",
        ],
    },
)

# Мониторинг прогресса
import time

while True:
    status = client.agent_runs.get_status(run.id)
    print(f"Статус: {status.state}, Итерации: {status.iterations}")

    if status.state in ("COMPLETED", "FAILED", "TERMINATED"):
        break

    time.sleep(10)

# Получение результата
result = client.agent_runs.get_result(run.id)
print(result.output)

# История выполненных шагов
steps = client.agent_runs.get_steps(run.id)
for step in steps:
    print(f"Шаг {step.step_id}: {step.tool_name} — {step.status}")

Кастомные Toolkits

# SuperAGI Toolkit — набор инструментов для агента
from superagi.tools.base_tool import BaseTool, BaseToolkit
from pydantic import BaseModel, Field

class CRMQueryInput(BaseModel):
    customer_name: str = Field(description="Имя клиента или компании")
    fields: list[str] = Field(description="Поля для получения", default=["all"])

class CRMQueryTool(BaseTool):
    name: str = "crm_query"
    description: str = "Поиск информации о клиенте в CRM системе"
    args_schema = CRMQueryInput

    def execute(self, customer_name: str, fields: list[str] = None) -> str:
        results = crm_api.search(name=customer_name, fields=fields)
        return str(results)

class CRMCreateTaskTool(BaseTool):
    name: str = "crm_create_task"
    description: str = "Создать задачу в CRM для клиента"

    def execute(self, customer_id: str, task_title: str, due_date: str) -> str:
        task = crm_api.create_task(
            customer_id=customer_id,
            title=task_title,
            due_date=due_date,
        )
        return f"Задача создана: {task['id']}"

class CRMToolkit(BaseToolkit):
    name: str = "CRMToolkit"
    description: str = "Инструменты для работы с CRM системой"

    def get_tools(self) -> list[BaseTool]:
        return [CRMQueryTool(), CRMCreateTaskTool()]

Расписание: агенты по триггерам

# Ежедневный отчёт в 9:00
schedule = client.agent_schedules.create(
    agent_id=agent.id,
    schedule_config=AgentScheduleConfig(
        start_time="2025-01-01T09:00:00",
        recurrence="DAILY",
        expiry_runs=30,    # 30 запусков максимум
        time_zone="Europe/Moscow",
    ),
)

# Webhook-триггер при новом лиде в CRM
# (интеграция через HTTP API SuperAGI)
@app.route("/webhook/new-lead", methods=["POST"])
def handle_new_lead():
    lead = request.json
    client.agent_runs.start(
        agent_id=lead_agent.id,
        run_config={"goals": [f"Проанализируй нового лида: {lead['company']}, {lead['contact']}"]},
    )
    return {"status": "started"}

Практический кейс: агент мониторинга конкурентов

Задача: еженедельный сбор информации о 8 конкурентах — новости, вакансии (сигнал о развитии), изменения цен, обновления продуктов.

Конфигурация:

• Toolkits: GoogleSearch, WebScraper, FileWriter
• Goals: 4 цели (по типу данных)
• Расписание: каждый понедельник 07:00

Рабочий процесс:

1. Агент ищет новости по каждому конкуренту
2. Скрапит страницы вакансий (по 5 конкурентам с открытым career-разделом)
3. Проверяет pricing pages на изменения
4. Генерирует структурированный markdown-отчёт
5. Сохраняет в корпоративный Notion

Результаты:

• Время подготовки еженедельного конкурентного отчёта: 3 часа → 12 минут ревью готового
• Пропущенных значимых событий: ~1–2/квартал (ранее 4–5)
• Нагрузка на аналитика: -5 часов/неделю

Когда выбирать SuperAGI

SuperAGI оправдан при наличии команды без глубокой экспертизы в LangChain/LangGraph — web-интерфейс ускоряет запуск. Для сложных production-систем с нестандартной логикой предпочтителен LangGraph с более явным контролем над выполнением.

Сроки

• Развёртывание и базовая конфигурация: 1–2 дня
• Настройка агентов через UI: 2–3 дня
• Кастомные toolkits: 3–5 дней
• Интеграция с корпоративными системами: 1–2 недели