Создаем многоагентные системы с Swarm от OpenAI

Приветствую, друзья! Сегодня я хочу рассказать вам одном занятном фреймворке — 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: в каталоге можно посмотреть список всех программ, а в календаре — записаться на открытые уроки.