Приветствую, друзья! Сегодня я хочу рассказать вам одном занятном фреймворке — Swarm от OpenAI.

Если вы когда-либо задумывались о том, как создать систему, где несколько компонентов взаимодействуют между собой, передают задачи друг другу и делают это без вашего постоянного участия — Swarm для вас. Он основан на OpenAI's Chat Completions API, что может сделать его мощным инструментом в ваших руках.

Давайте уже к коду!

Хватит теории, переходим к практике.

# Установка из репозитория GitHub
pip install git+https://github.com/openai/swarm.git

Создаём наших первых агентов

Начнём с простого примера. Допустим, есть два агента:

• Агент А: дружелюбный помощник.

• Агент Б: поэт, говорящий только в хайку.

Импортируем необходимые классы:

from swarm import Swarm, Agent

Создадим клиента Swarm:

client = Swarm()

Теперь определим функцию, которая позволит Агенту А передать управление Агенту Б:

def transfer_to_agent_b():
    return agent_b

Создам наших агентов:

agent_a = Agent(
    name="Агент A",
    instructions="Ты дружелюбный помощник.",
    functions=[transfer_to_agent_b],
)

agent_b = Agent(
    name="Агент B",
    instructions="Отвечай только в стиле хайку.",
)

Запускаем диалог:

response = client.run(
    agent=agent_a,
    messages=[{"role": "user", "content": "Я хочу поговорить с Агентом B."}],
)

print(response.messages[-1]["content"])

И что же мы получим в ответ? Что-то вроде:

Звезды в небе спят,
Слушаю тебя сейчас,
Чем могу помочь?

Агент А, увидев, что пользователь хочет поговорить с Агентом Б, вызвал функцию transfer_to_agent_b(), которая вернула нам Агент Б. Swarm автоматически переключил контекст, и теперь мы общаемся с Агентом Б, который, как истинный поэт, отвечает в стиле хайку.

Добавляем немного контекста

Хорошо, а что если мы хотим, чтобы наши агенты знали имя пользователя? Для этого используем context_variables.

def greet_user(context_variables):
    user_name = context_variables.get("user_name", "друг")
    return f"Привет, {user_name}!"

agent_a = Agent(
    name="Агент А",
    instructions="Ты дружелюбный помощник.",
    functions=[greet_user, transfer_to_agent_b],
)

Теперь при запуске передадим имя пользователя:

response = client.run(
    agent=agent_a,
    messages=[{"role": "user", "content": "Я хочу поговорить с Агентом Б."}],
    context_variables={"user_name": "Анна"},
)

print(response.messages[1]["content"])  # Ответ от функции greet_user
print(response.messages[-1]["content"])  # Ответ от Агента Б

Вывод будет примерно таким:

Привет, Анна!
Солнце светит ярко,
Слушаю тебя внимательно,
Чем могу помочь?

Агент по поддержке клиентов

Теперь представим, что мы создаём систему поддержки клиентов. Нам нужен главный агент, который будет перенаправлять запросы в нужные отделы.

Создаём функции для передачи

def transfer_to_sales():
    return sales_agent

def transfer_to_support():
    return support_agent

def transfer_to_billing():
    return billing_agent

Определяем агентов отделов

sales_agent = Agent(
    name="Отдел продаж",
    instructions="Ты специалист отдела продаж. Помогаешь клиентам с покупками.",
)

support_agent = Agent(
    name="Служба поддержки",
    instructions="Ты сотрудник поддержки. Решаешь технические проблемы.",
)

billing_agent = Agent(
    name="Бухгалтерия",
    instructions="Ты сотрудник бухгалтерии. Отвечаешь на вопросы по оплате.",
)

Создаём главного агента

def triage_issue(context_variables, issue_description):
    issue_description = issue_description.lower()
    if "купить" in issue_description or "заказ" in issue_description:
        return Result(agent=sales_agent)
    elif "не работает" in issue_description or "ошибка" in issue_description:
        return Result(agent=support_agent)
    elif "оплата" in issue_description or "счёт" in issue_description:
        return Result(agent=billing_agent)
    else:
        return "Извините, я не совсем понял ваш запрос."

main_agent = Agent(
    name="Главный агент",
    instructions="Ты оператор колл-центра. Определяешь, в какой отдел направить клиента.",
    functions=[triage_issue],
)

Запускаем сценарий

response = client.run(
    agent=main_agent,
    messages=[{"role": "user", "content": "У меня проблема с оплатой заказа."}],
)

print(f"Клиент перенаправлен в: {response.agent.name}")
print(response.messages[-1]["content"])

Ожидаемый вывод:

Клиент перенаправлен в: Бухгалтерия
Здравствуйте! Чем могу помочь с вашей оплатой?

Обработка ошибок и дополнительные функции

Что если пользователь введёт что-то неожиданное? Swarm автоматически добавит сообщение об ошибке в диалог, и агент сможет попытаться снова.

Пример:

def check_order_status(order_id):
    if not order_id.isdigit():
        return "Пожалуйста, предоставьте корректный номер заказа."
    # Допустим, мы проверили статус заказа
    return f"Статус вашего заказа {order_id}: Отправлен."

support_agent = Agent(
    name="Служба поддержки",
    instructions="Ты сотрудник поддержки. Помогаешь клиентам с вопросами о заказах.",
    functions=[check_order_status],
)

Стриминг ответов

Swarm поддерживает стриминг ответов, что позволяет получать ответы по частям, как только они генерируются.

stream = client.run(
    agent=agent_b,
    messages=[{"role": "user", "content": "Расскажи стихотворение."}],
    stream=True,
)

for chunk in stream:
    print(chunk.get("content", ""), end="")

Заключение

• Логируйте всё: добавьте debug=True при запуске client.run(), чтобы видеть, что происходит под капотом.

• Используйте разные модели: можете задавать модель для каждого агента. Например, gpt-3.5-turbo для простых задач и gpt-4 для сложных.

• Контекст — наше всё: передавайте нужные данные через context_variables, чтобы агенты могли использовать их в своих функциях.

Если у вас возникли вопросы или вы хотите поделиться своими идеями, смело пишите в комментариях.

Ссылка на фреймворк

Все актуальные методы и инструменты DS и ML можно освоить на онлайн-курсах OTUS: в каталоге можно посмотреть список всех программ, а в календаре — записаться на открытые уроки.