Это первая часть серии, состоящей из нескольких частей, посвященной созданию агентов с помощью OpenAI Assistant API с использованием Python SDK.

Что такое агенты?

Мне нравится смотреть на это так: агент на самом деле представляет собой просто часть программного обеспечения, использующую LLM (большую языковую модель) и пытающуюся имитировать человеческое поведение. Это означает, что он может не только разговаривать и понимать язык, но также выполнять действия, которые оказывают влияние на реальный мир. Эти действия обычно называются инструментами.

В этом сообщении блога мы рассмотрим, как создать агент с помощью OpenAI Assistant API, используя их Python SDK. Часть 1 будет просто скелетом помощника. То есть только разговорная часть.

Я решил создать приложение CLI, чтобы оно не зависело от платформы. Мы намеренно назовем нашу реализацию «Агентом», а реализацию OpenAI SDK — «Помощником», чтобы легко различать их.

Я использую термины «инструменты» и «функции» взаимозаменяемо, когда речь идет о функциях, которые может вызывать агент. Во второй части более подробно будет рассмотрен вызов функции.

Предварительные условия

Чтобы следовать этому руководству, вам понадобится следующее:

• Python3 установлен на вашем компьютере
• Ключ API OpenAI
• Базовые знания программирования Python

Концепции помощника OpenAI

Ассистент : Ассистент в API Assistants — это объект, настроенный для ответа на сообщения пользователя. Он использует инструкции, выбранную модель и инструменты для взаимодействия с функциями и предоставления ответов.

Тема : Тема представляет собой беседу или диалог в API Assistants. Он создается для каждого взаимодействия с пользователем и может содержать несколько сообщений, выступая в качестве контейнера для текущего разговора.

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

Запуск : Запуск — это экземпляр Помощника, обрабатывающего поток. Он включает в себя чтение потока, принятие решения о вызове инструментов и генерацию ответов на основе интерпретации моделью сообщений потока.

Настройка среды разработки

Первый шаг — создать виртуальную среду с помощью venv и активировать ее. Это гарантирует, что наши зависимости изолированы от установки системного Python:

 python3 -m venv venv source venv/bin/activate

Давайте установим нашу единственную зависимость: пакет openai :

 pip install openai

Создайте файл main.py Давайте добавим некоторую базовую логику времени выполнения для нашего приложения CLI:

 while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the assistant...") break print(f"Assistant: You said {user_input}")

Попробуйте это, запустив python3 main.py :

 python3 main.py User: hi Assistant: You said hi

Как видите, CLI принимает сообщение пользователя в качестве входных данных, а у нашего гениального помощника пока нет мозгов 🧠, поэтому он просто повторяет сообщение. Еще не такой умный.

Агент

Теперь начинается самое интересное 😁 (или головная боль 🤕). Я предоставлю весь импорт, необходимый для финального класса, прямо сейчас, чтобы вы не ломали голову над тем, откуда что взялось, поскольку для краткости я исключил импорт из примеров кода. Начнем с создания класса Agent в новом файле agent.py :

 import time import openai from openai.types.beta.threads.run import Run class Agent: def __init__(self, name: str, personality: str): self.name = name self.personality = personality self.client = openai.OpenAI(api_key="sk-*****") self.assistant = self.client.beta.assistants.create( name=self.name, model="gpt-4-turbo-preview" )

В конструкторе класса мы инициализируем клиент OpenAI как свойство класса, передавая наш ключ API OpenAI. Затем мы создаем свойство класса assistant , которое сопоставляется с нашим недавно созданным помощником. Мы сохраняем name и personality как свойства класса для дальнейшего использования.

Аргумент name , который мы передаем методу создания, предназначен только для идентификации Помощника на панели управления OpenAI, и на данный момент ИИ фактически не знает об этом. На самом деле вам нужно передать имя instructions , которые мы увидим позже.

Вы уже могли задать instructions при создании Ассистента, но на самом деле это сделает вашего Ассистента менее гибким к динамическим изменениям.

Вы можете обновить Assistant, вызвав client.beta.assistants.update , но есть лучшее место для передачи динамических значений, которые мы увидим, когда доберемся до Runs.

Обратите внимание: если вы передаете instructions здесь и снова при создании запуска, instructions Ассистента будут перезаписаны instructions запуска. Они не дополняют друг друга, поэтому выбирайте один в зависимости от ваших потребностей: уровень помощника для статических инструкций или уровень выполнения для динамических инструкций.

В качестве модели я выбрал модель gpt-4-turbo-preview , чтобы мы могли добавить вызов функций во второй части этой серии статей. Вы можете использовать gpt-3.5-turbo , если хотите сэкономить несколько долей копейки, одновременно вызывая у себя мигрень чистого разочарования при внедрении инструментов.

GPT 3.5 ужасен при вызове инструментов; часы, которые я потерял, пытаясь справиться с этим, позволяют мне сказать это. 😝 На этом я оставлю, а об этом позже.

Создание темы, добавление сообщений и получение последнего сообщения

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

 class Agent: # ... (rest of code) def create_thread(self): self.thread = self.client.beta.threads.create()

И нам понадобится способ добавлять сообщения в эту ветку:

 class Agent: # ... (rest of code) def add_message(self, message): self.client.beta.threads.messages.create( thread_id=self.thread.id, role="user", content=message )

Обратите внимание, что на данный момент добавлять сообщения можно только с ролью user . Я считаю, что OpenAI планирует изменить это в будущем выпуске, поскольку это довольно ограничивает.

Теперь мы можем получить последнее сообщение в ветке:

 class Agent: # ... (rest of code) def get_last_message(self): return self.client.beta.threads.messages.list( thread_id=self.thread.id ).data[0].content[0].text.value

Затем мы создаем метод точки входа run_agent , чтобы проверить, что у нас есть на данный момент. В настоящее время метод run_agent просто возвращает последнее сообщение в потоке. На самом деле он не выполняет Run. Это все равно безмозгло.

 class Agent: # ... (rest of code) def run_agent(self): message = self.get_last_message() return message

Вернувшись в main.py , мы создаем агент и наш первый поток. Добавляем сообщение в ветку. Затем верните то же сообщение обратно пользователю, но на этот раз из этого живого потока.

 from agent import Agent agent = Agent(name="Bilbo Baggins", personality="You are the accomplished and renowned adventurer from The Hobbit. You act like you are a bit of a homebody, but you are always up for an adventure. You worry a bit too much about breakfast.") agent.create_thread() while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the agent...") break agent.add_message(user_input) answer = agent.run_agent() print(f"Assistant: {answer}")

Давайте запустим:

 python3 main.py User: hi Assistant: hi

Еще не очень умный. Ближе к попугаю 🦜, чем к хоббиту. В следующем разделе начинается настоящее веселье.

Создание и опрос запуска

При создании прогона вам необходимо периодически получать объект Run , чтобы проверять состояние прогона. Это называется опросом, и это отстой. Вам необходимо провести опрос, чтобы определить, что вашему агенту следует делать дальше. OpenAI планирует добавить поддержку потоковой передачи, чтобы упростить задачу. А пока я покажу вам, как настроить опрос, в следующем разделе.

Обратите внимание на символ _ в следующих именах методов, который является стандартом Python и указывает, что метод предназначен для внутреннего использования и к нему не следует обращаться напрямую из внешнего кода.

Сначала давайте создадим вспомогательный метод _create_run для создания Run и обновим run_agent для вызова этого метода:

 class Agent: # ... (rest of code) def get_breakfast_count_from_db(self): return 1 def _create_run(self): count = self.get_breakfast_count_from_db() return self.client.beta.threads.runs.create( thread_id=self.thread.id, assistant_id=self.assistant.id, instructions=f""" Your name is: {self.name} Your personality is: {self.personality} Metadata related to this conversation: {{ "breakfast_count": {count} }} """, ) def run_agent(self): run = self._create_run() # add this line message = self.get_last_message() return message

Обратите внимание, как мы передаем thread.id и assistant.id для создания запуска.

Помните, как я говорил вначале, что есть лучшее место для передачи динамических инструкций и данных? Это будет параметр instructions при создании файла Run. В нашем случае мы могли бы получить count завтраков из базы данных. Это позволит вам легко передавать различные соответствующие динамические данные каждый раз, когда вы хотите инициировать ответ.

Теперь ваш агент знает, как мир меняется вокруг него, и может действовать соответствующим образом. Мне нравится иметь в своих инструкциях объект метаданных JSON, который сохраняет соответствующий динамический контекст. Это позволяет мне передавать данные менее подробно и в формате, который очень хорошо понимает LLM.

Пока не запускайте это; это не сработает, потому что мы не ждем завершения выполнения при получении последнего сообщения, поэтому оно все равно будет последним сообщением пользователя.

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

 class Agent: # ... (rest of code) def _retrieve_run(self, run: Run): return self.client.beta.threads.runs.retrieve( run_id=run.id, thread_id=self.thread.id)

Обратите внимание, что нам нужно передать как run.id так и thread.id , чтобы найти конкретный запуск.

Добавьте метод _poll_run в наш класс агента:

 class Agent: # ... (rest of code) def _cancel_run(self, run: Run): self.client.beta.threads.runs.cancel( run_id=run.id, thread_id=self.thread.id) def _poll_run(self, run: Run): status = run.status start_time = time.time() while status != "completed": if status == 'failed': raise Exception(f"Run failed with error: {run.last_error}") if status == 'expired': raise Exception("Run expired.") time.sleep(1) run = self._retrieve_run(run) status = run.status elapsed_time = time.time() - start_time if elapsed_time > 120: # 2 minutes self._cancel_run(run) raise Exception("Run took longer than 2 minutes.")

🥵 Уф, это много... Давайте распаковать.

_poll_run получает объект Run в качестве аргумента и извлекает текущий status Run. Все доступные статусы можно найти в документации OpenAI. Мы просто будем использовать несколько из них, которые подходят для нашей текущей цели.

Теперь мы запускаем цикл while, чтобы проверить статус завершения, обрабатывая несколько сценариев ошибок. Фактическая оплата API Assistant немного неясна, поэтому на всякий случай я решил отменить запуски через 2 минуты.

Несмотря на то, что существует статус expired , когда OpenAI отменяет запуск через 10 минут. Если пробежка занимает более 2 минут, возможно, у вас все равно есть проблема.

Поскольку я также не хочу опрашивать каждые несколько миллисекунд, я ограничиваю свой запрос, опрашивая только каждую 1 секунду, пока не достигну 2-минутной отметки и не отменю прогон. Вы можете настроить это так, как считаете нужным.

На каждой итерации после задержки мы снова получаем статус «Выполнение».

Теперь давайте подключим все это к нашему методу run_agent . Вы заметите, что сначала мы создаем запуск с помощью _create_run затем опрашиваем с помощью _poll_run , пока не получим ответ или не возникнет ошибка, и, наконец, когда опрос завершен, мы извлекаем последнее сообщение из потока, которое теперь будет от агента.

Затем мы возвращаем сообщение в наш цикл выполнения, чтобы его можно было отправить обратно пользователю.

 class Agent: # ... (rest of code) def run_agent(self): run = self._create_run() self._poll_run(run) # add this line message = self.get_last_message() return message

Вуаля, теперь, когда вы снова запустите своего агента, вы получите ответ от нашего дружелюбного агента:

 python3 main.py User: hi Assistant: Hello there! What adventure can we embark on today? Or perhaps, before we set out, we should think about breakfast. Have you had yours yet? I've had mine, of course – can't start the day without a proper breakfast, you know. User: how many breakfasts have you had? Assistant: Ah, well, I've had just 1 breakfast today. But the day is still young, and there's always room for a second, isn't there? What about you? How can I assist you on this fine day?

Во второй части мы добавим возможность нашему Агенту вызывать инструменты.

Полный код вы можете найти на моем GitHub .

Спасибо за ваше чтение. Рад услышать любые мысли и отзывы в комментариях. Следуйте за мной в Linkedin, чтобы увидеть больше такого контента: https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/