Как создать агент с помощью OpenAI Assistant на Python. Часть 1: Диалог

Это первая часть серии, состоящей из нескольких частей, посвященной созданию агентов с помощью OpenAI Assistant API с использованием Python SDK. Что такое агенты? Мне нравится смотреть на это так: агент на самом деле представляет собой просто часть программного обеспечения, использующую LLM (большую языковую модель) и пытающуюся имитировать человеческое поведение. Это означает, что он может не только разговаривать и понимать язык, но также выполнять действия, которые оказывают влияние на реальный мир. Эти действия обычно называются инструментами. В этом сообщении блога мы рассмотрим, как создать агент с помощью OpenAI Assistant API, используя их Python SDK. Часть 1 будет просто скелетом помощника. То есть только разговорная часть. Я решил создать приложение CLI, чтобы оно не зависело от платформы. Мы намеренно назовем нашу реализацию «Агентом», а реализацию OpenAI SDK — «Помощником», чтобы легко различать их. Я использую термины и взаимозаменяемо, когда речь идет о функциях, которые может вызывать агент. Во второй части более подробно будет рассмотрен вызов функции. «инструменты» «функции» Предварительные условия Чтобы следовать этому руководству, вам понадобится следующее: Python3 установлен на вашем компьютере Ключ API OpenAI Базовые знания программирования Python Концепции помощника OpenAI : Ассистент в API Assistants — это объект, настроенный для ответа на сообщения пользователя. Он использует инструкции, выбранную модель и инструменты для взаимодействия с функциями и предоставления ответов. Ассистент : Тема представляет собой беседу или диалог в API Assistants. Он создается для каждого взаимодействия с пользователем и может содержать несколько сообщений, выступая в качестве контейнера для текущего разговора. Тема : Сообщение — это единица связи в потоке. Он содержит текст (и, возможно, файлы в будущем) и используется для передачи пользовательских запросов или ответов помощника в потоке. Сообщение : Запуск — это экземпляр Помощника, обрабатывающего поток. Он включает в себя чтение потока, принятие решения о вызове инструментов и генерацию ответов на основе интерпретации моделью сообщений потока. Запуск Настройка среды разработки Первый шаг — создать виртуальную среду с помощью и активировать ее. Это гарантирует, что наши зависимости изолированы от установки системного Python: venv python3 -m venv venv source venv/bin/activate Давайте установим нашу единственную зависимость: пакет : openai pip install openai Создайте файл Давайте добавим некоторую базовую логику времени выполнения для нашего приложения CLI: main.py 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 Аргумент , который мы передаем методу создания, предназначен только для идентификации Помощника на панели управления OpenAI, и на данный момент ИИ фактически не знает об этом. На самом деле вам нужно передать имя , которые мы увидим позже. name instructions Вы уже могли задать при создании Ассистента, но на самом деле это сделает вашего Ассистента менее гибким к динамическим изменениям. instructions Вы можете обновить Assistant, вызвав , но есть лучшее место для передачи динамических значений, которые мы увидим, когда доберемся до Runs. client.beta.assistants.update Обратите внимание: если вы передаете здесь и снова при создании запуска, Ассистента будут перезаписаны запуска. Они не дополняют друг друга, поэтому выбирайте один в зависимости от ваших потребностей: уровень помощника для статических инструкций или уровень выполнения для динамических инструкций. 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 ) Обратите внимание, что на данный момент добавлять сообщения можно только с ролью . Я считаю, что OpenAI планирует изменить это в будущем выпуске, поскольку это довольно ограничивает. user Теперь мы можем получить последнее сообщение в ветке: 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. Это все равно безмозгло. run_agent run_agent 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 Еще не очень умный. Ближе к попугаю 🦜, чем к хоббиту. В следующем разделе начинается настоящее веселье. Создание и опрос запуска При создании прогона вам необходимо периодически получать объект , чтобы проверять состояние прогона. Это называется опросом, и это отстой. Вам необходимо провести опрос, чтобы определить, что вашему агенту следует делать дальше. OpenAI планирует добавить поддержку потоковой передачи, чтобы упростить задачу. А пока я покажу вам, как настроить опрос, в следующем разделе. Run Обратите внимание на символ в следующих именах методов, который является стандартом 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 Помните, как я говорил вначале, что есть лучшее место для передачи динамических инструкций и данных? Это будет параметр при создании файла Run. В нашем случае мы могли бы получить завтраков из базы данных. Это позволит вам легко передавать различные соответствующие динамические данные каждый раз, когда вы хотите инициировать ответ. instructions 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.") 🥵 Уф, это много... Давайте распаковать. получает объект в качестве аргумента и извлекает текущий Run. Все доступные статусы можно найти в OpenAI. Мы просто будем использовать несколько из них, которые подходят для нашей текущей цели. _poll_run Run status документации Теперь мы запускаем цикл while, чтобы проверить статус завершения, обрабатывая несколько сценариев ошибок. Фактическая оплата API Assistant немного неясна, поэтому на всякий случай я решил отменить запуски через 2 минуты. Несмотря на то, что существует статус , когда OpenAI отменяет запуск через 10 минут. Если пробежка занимает более 2 минут, возможно, у вас все равно есть проблема. expired Поскольку я также не хочу опрашивать каждые несколько миллисекунд, я ограничиваю свой запрос, опрашивая только каждую 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/