Esta é a primeira parte de uma série de várias partes sobre a construção de agentes com a API Assistant da OpenAI usando o Python SDK.

O que são agentes?

Do jeito que gosto de ver, um agente é, na verdade, apenas um software que aproveita um LLM (Large Language Model) e tenta imitar o comportamento humano. Isso significa que ele pode não apenas conversar e compreender a linguagem, mas também realizar ações que tenham impacto no mundo real. Essas ações normalmente são chamadas de ferramentas.

Nesta postagem do blog, exploraremos como construir um agente usando a API Assistant da OpenAI usando seu SDK Python. A Parte 1 será apenas o esqueleto do assistente. Ou seja, apenas a parte conversacional.

Optei por construir um aplicativo CLI propositalmente para ser independente de estrutura. Chamaremos propositalmente nossa implementação de Agente e nos referiremos à implementação do OpenAI SDK como Assistente para distinguir facilmente entre os dois.

Eu uso os termos ferramentas e funções de forma intercambiável quando se trata de funções que o Agente pode chamar. A Parte 2 abordará a chamada de função com mais detalhes.

Pré-requisitos

Para acompanhar este tutorial, você precisará do seguinte:

• Python3 instalado em sua máquina
• Uma chave de API OpenAI
• Conhecimento básico de programação Python

Conceitos do assistente OpenAI

Assistant : Um Assistant na API Assistants é uma entidade configurada para responder às mensagens do usuário. Ele usa instruções, um modelo escolhido e ferramentas para interagir com funções e fornecer respostas.

Thread : Um Thread representa uma conversa ou diálogo na API Assistants. Ele é criado para cada interação do usuário e pode conter diversas mensagens, servindo como contêiner para a conversa em andamento.

Mensagem : Uma mensagem é uma unidade de comunicação em um Thread. Ele contém texto (e potencialmente arquivos no futuro) e é usado para transmitir dúvidas do usuário ou respostas do assistente dentro de um Thread.

Run : Uma Run é uma instância do Assistant processando um Thread. Envolve ler o Thread, decidir se deve chamar ferramentas e gerar respostas com base na interpretação do modelo das mensagens do Thread.

Configurando o Ambiente de Desenvolvimento

O primeiro passo é criar um ambiente virtual usando venv e ativá-lo. Isso garantirá que nossas dependências sejam isoladas da instalação do sistema Python:

 python3 -m venv venv source venv/bin/activate

Vamos instalar nossa única dependência: o pacote openai :

 pip install openai

Crie um arquivo main.py Vamos preencher alguma lógica básica de tempo de execução para nosso aplicativo CLI:

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

Experimente executando python3 main.py :

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

Como você pode ver, a CLI aceita uma mensagem do usuário como entrada, e nosso assistente genial ainda não tem cérebro 🧠, então ele apenas repete a mensagem de volta. Ainda não é tão inteligente.

O agente

Agora começa a diversão 😁 (ou dores de cabeça 🤕). Fornecerei todas as importações necessárias para a aula final agora, para que você não fique pensando de onde as coisas estão vindo, já que mantive as importações fora dos exemplos de código por questões de brevidade. Vamos começar construindo uma classe Agent em um novo arquivo 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" )

No construtor da classe, inicializamos o cliente OpenAI como uma propriedade de classe, passando nossa chave de API OpenAI. A seguir, criamos uma propriedade de classe assistant que mapeia para nosso Assistente recém-criado. Armazenamos name e personality como propriedades de classe para uso posterior.

O argumento name que estamos passando para o método create serve apenas para identificar o Assistant no painel OpenAI, e a IA não está realmente ciente disso neste momento. Na verdade, você tem que passar o nome para as instructions que veremos mais tarde.

Você já pode definir instructions ao criar o Assistente, mas na verdade isso tornará seu Assistente menos flexível para alterações dinâmicas.

Você pode atualizar um Assistente chamando client.beta.assistants.update , mas há um lugar melhor para passar valores dinâmicos que veremos quando chegarmos às Execuções.

Observe que se você passar instructions aqui e novamente ao criar uma corrida, as instructions do Assistente serão substituídas pelas instructions da corrida. Eles não se complementam, então escolha um com base em suas necessidades: nível Assistant para instruções estáticas ou nível Run para instruções dinâmicas.

Para o modelo, escolhi o modelo gpt-4-turbo-preview para que possamos adicionar chamadas de função na parte 2 desta série. Você poderia usar gpt-3.5-turbo se quiser economizar algumas frações de centavo e ao mesmo tempo ter uma enxaqueca de pura frustração no futuro quando implementarmos ferramentas.

O GPT 3.5 é péssimo para chamar ferramentas; as horas que perdi tentando lidar com isso me permitem dizer isso. 😝 Vou deixar por isso mesmo e falaremos mais sobre isso mais tarde.

Criando um tópico, adicionando mensagens e recuperando a última mensagem

Depois de criarmos um agente, precisaremos iniciar uma conversa.

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

E vamos querer uma maneira de adicionar mensagens a esse tópico:

 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 )

Observe que no momento só é possível adicionar mensagens com a função user . Acredito que a OpenAI planeja mudar isso em uma versão futura, pois isso é bastante limitante.

Agora, podemos obter a última mensagem do tópico:

 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

A seguir, criamos um método run_agent de ponto de entrada para testar o que temos até agora. Atualmente, o método run_agent apenas retorna a última mensagem do thread. Na verdade, ele não executa uma corrida. Ainda não tem cérebro.

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

De volta a main.py , criamos o agente e nosso primeiro thread. Adicionamos uma mensagem ao tópico. Em seguida, retorne a mesma mensagem ao usuário, mas desta vez, vindo daquele tópico ativo.

 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}")

Vamos executá-lo:

 python3 main.py User: hi Assistant: hi

Ainda não é muito inteligente. Mais perto de um papagaio 🦜 do que de um hobbit. Na próxima seção, a verdadeira diversão começa.

Criando e pesquisando uma execução

Ao criar uma execução, você precisa recuperar periodicamente o objeto Run para verificar o status da execução. Isso se chama votação e é uma droga. Você precisa fazer uma pesquisa para determinar o que seu agente deve fazer a seguir. A OpenAI planeja adicionar suporte para streaming para tornar isso mais simples. Enquanto isso, mostrarei como configurar a votação na próxima seção.

Observe o _ nos nomes dos métodos a seguir, que é o padrão em Python para indicar que o método se destina ao uso interno e não deve ser acessado diretamente por código externo.

Primeiro, vamos criar um método auxiliar _create_run para criar um Run e atualizar run_agent para chamar este método:

 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

Observe como passamos thread.id e assistant.id para criar uma execução.

Lembra como eu disse no início que havia um lugar melhor para passar instruções e dados dinâmicos? Esse seria o parâmetro instructions ao criar o Run. No nosso caso, poderíamos fazer com que a count do café da manhã fosse obtida de um banco de dados. Isso permitirá que você transmita facilmente diferentes dados dinâmicos relevantes sempre que desejar acionar uma resposta.

Agora, seu agente está ciente das mudanças no mundo ao seu redor e pode agir de acordo. Gosto de ter um objeto JSON de metadados em minhas instruções que mantenha o contexto dinâmico relevante. Isso me permite transmitir dados sendo menos detalhado e em um formato que o LLM entende muito bem.

Não execute isso ainda; não funcionará porque não estamos aguardando a conclusão da execução quando recebermos a última mensagem, portanto, ainda será a última mensagem do usuário.

Vamos resolver isso construindo nosso mecanismo de votação. Primeiro, precisaremos de uma maneira de recuperar uma execução de forma fácil e repetida, então vamos adicionar um método _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)

Observe como precisamos passar run.id e thread.id para encontrar uma execução específica.

Adicione um método _poll_run à nossa classe Agent:

 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.")

🥵 Ufa, é muita coisa... Vamos desempacotar.

_poll_run recebe um objeto Run como argumento e extrai o status atual da execução. Todos os status disponíveis podem ser encontrados nos documentos OpenAI. Usaremos apenas alguns que atendam ao nosso propósito atual.

Agora executamos um loop while para verificar o status de concluído enquanto lidamos com alguns cenários de erro. O faturamento real da API Assistant é um pouco obscuro, então, para garantir, optei por cancelar minhas execuções após 2 minutos.

Mesmo que haja um status expired para quando o OpenAI cancela a execução após 10 minutos. Se uma corrida demorar mais de 2 minutos, você provavelmente terá um problema de qualquer maneira.

Como também não quero fazer pesquisas a cada poucos milissegundos, acelero minha solicitação pesquisando apenas a cada 1 segundo até atingir a marca de 2 minutos e cancelar minha execução. Você pode ajustar isso para o que achar melhor.

Cada iteração após o atraso, buscamos novamente o status Run.

Agora, vamos inserir tudo isso em nosso método run_agent . Você notará que primeiro criamos a execução com _create_run depois pesquisamos com _poll_run até obtermos uma resposta ou um erro for gerado e, finalmente, quando a pesquisa terminar, recuperamos a última mensagem do thread que agora será do agente.

Em seguida, retornamos a mensagem ao nosso loop de tempo de execução, para que ela possa ser enviada de volta ao usuário.

 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

Voilà, agora quando você executar seu agente novamente, receberá uma resposta de nosso amigável Agente:

 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?

Na parte 2, adicionaremos a capacidade de nosso Agente chamar ferramentas.

Você pode encontrar o código completo no meu GitHub .

Obrigado pela sua leitura. Fico feliz em ouvir quaisquer opiniões e comentários nos comentários. Siga-me no Linkedin para mais conteúdos como este: https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/