Introdução

A Gata Guru é uma aplicação moderna de chatbot desenvolvida para facilitar a análise de dados corporativos da Petlove através de interface conversacional com Inteligência Artificial. Esta solução representa um exemplo prático de como integrar tecnologias de IA com ferramentas de análise de dados em uma arquitetura escalável e bem estruturada.

Como funciona a Gata Guru?

O Gata Guru é um sistema modular de chat inteligente que permite aos usuários fazerem perguntas sobre os dados disponíveis no Catálogo de dados e DW da Petlove e receber respostas inteligentes geradas por IA. A Gata Guru também possui acesso ao Bigquery e tem permissão para rodar queries e extrais dados para responder as perguntas dos usuários.

Principais Funcionalidades

• 💬 Chat Inteligente: Interface conversacional responsiva em Streamlit
• 🧠 IA Avançada: Integração com modelos mais modernos da OpenAI
• 📊 Análise de Dados: Consultas automáticas no BigQuery
• 🔐 Segurança: Autenticação OAuth com domínio corporativo
• 📈 Visualizações: Dashboards e gráficos interativos
• 🔍 Catálogo de Dados: Acesso a metadados e documentação
• 🐛 Monitoramento: Tracking de erros com Honeybadger

Arquitetura da Aplicação

Visão Geral da Arquitetura

A Gata Guru foi desenvolvida seguindo princípios de arquitetura modular e Clean Code, onde cada componente tem responsabilidades bem definidas e pode ser desenvolvido, testado e mantido independentemente.

Componentes Principais

1. Gataguru System (chat/gataguru_system.py)

• Responsabilidade: Orquestração geral do sistema
• Funções: Inicialização, configuração e coordenação de componentes
• Características: Ponto central que integra todos os módulos

2. Conversation Manager (chat/conversation_manager.py)

• Responsabilidade: Processamento de mensagens e orquestração de agentes
• Funções: Execução de agentes especializados, Processamento de respostas, Gerenciamento de contexto de conversa, Montagem de conversas estruturadas

3. Agent Factory (chat/agent_factory.py)

• Responsabilidade: Criação e configuração de agentes especializados
• Tipos de Agentes:
• Intern Agent: Para tarefas básicas e consultas simples
• Expert Agent: Para análises complexas e especializadas
• Data Manager Agent: Para gerenciamento de dados e metadados

4. Agents Tools (chat/agents_tools.py)

• Responsabilidade: Ferramentas para interação com BigQuery e catálogo de dados
• Funcionalidades: Execução de consultas SQL no BigQuery, Acesso ao catálogo de dados corporativo, Modelo Pydantic para análise de dados estruturada

5. Interface de Compatibilidade (chat/custom_agents.py)

• Responsabilidade: Manter compatibilidade com código existente
• Funções: Wrappers para funções como a answers() para processar as mensagens na IA e assemble_conversation() para organizar os históricos de mensagens

Stack Tecnológica

Frontend e Interface

Streamlit (v1.32+)

• Propósito: Framework web para aplicações de dados em Python
• Vantagens:
• Desenvolvimento rápido de interfaces
• Integração nativa com ecossistema Python
• Componentes interativos prontos para uso
• Suporte a autenticação OAuth

CSS Customizado

• Tema personalizado para interface moderna
• Design responsivo para desktop e mobile
• Componentes visuais customizados

Backend e Processamento

Python 3.9+

• Linguagem principal do projeto
• Tipagem estática com typing-extensions
• Estruturas de dados com Pydantic

OpenAI GPT-4

• Modelo de linguagem para processamento de perguntas
• Geração de respostas inteligentes e contextualizadas
• Integração via API oficial da OpenAI

Armazenamento e Dados

Google BigQuery

• Data warehouse para consultas de dados corporativos
• Integração via Google Cloud SDK
• Service Account para autenticação segura

Data Catalog API — OpenMetaData

• Metadados e documentação de dados
• Descoberta de datasets disponíveis
• Governança de dados corporativos

Qualidade e Monitoramento

Honeybadger

• Error tracking em produção
• Monitoramento de performance
• Alertas automáticos para problemas

Pytest

• Framework de testes unitários
• 55 testes automatizados
• Cobertura de código abrangente

Ruff

• Linting e formatação de código
• Padrões de qualidade automatizados
• Integração com pre-commit hooks

Autenticação e Segurança

Google OAuth

• Autenticação com contas Google
• Domínio corporativo restrito (@petlove.com.br)
• Integração nativa com Streamlit

Authlib

• Biblioteca OAuth para Python
• Gerenciamento de tokens e sessões

Estrutura do Projeto

Fluxo de funcionamento

1. Autenticação do Usuário

2. Processamento de Perguntas

Benefícios da Arquitetura Modular

✅ Separação de Responsabilidades

Cada componente tem uma função específica e bem definida, facilitando manutenção e evolução.

✅ Facilidade de Testes

Componentes isolados são mais fáceis de testar individualmente, resultando em 55 testes automatizados.

✅ Escalabilidade

Novos agentes e ferramentas podem ser adicionados sem afetar componentes existentes.

✅ Reutilização

Componentes podem ser reutilizados em outros projetos ou aplicações.

✅ Debugging

Problemas são mais fáceis de isolar e corrigir devido à modularidade.

Monitoramento e Observabilidade

Honeybadger Integration

• Error tracking automático
• Contexto de usuário e sessão
• Alertas em tempo real
• Métricas de performance

Logging Estruturado

• Logs em português brasileiro
• Níveis de log apropriados
• Contexto rico para debugging

Qualidade de Código

Padrões de Desenvolvimento

• Docstrings em português brasileiro
• Tipagem estática com type hints
• Formatação automática com Ruff
• Pre-commit hooks para qualidade

Testes Automatizados

• 55 testes unitários
• Cobertura abrangente
• Integração com CI/CD
• Execução automática no commit

Traces

• Monitoramento dos agentes e uso das ferramentas direto na Openai

Conclusão

A Gata Guru representa um exemplo moderno de aplicação de IA para análise de dados, demonstrando como integrar diferentes tecnologias em uma arquitetura escalável e bem estruturada. A escolha de frameworks e tecnologias foi cuidadosamente planejada para:

• Facilitar o desenvolvimento com Streamlit e Python
• Garantir qualidade com testes e ferramentas de linting
• Proporcionar escalabilidade com arquitetura modular
• Assegurar segurança com autenticação OAuth
• Manter observabilidade com monitoramento robusto

A aplicação serve como referência para desenvolvimento de soluções similares, mostrando como combinar IA, análise de dados e boas práticas de engenharia de software em um projeto real e funcional.

Projeto desenvolvido pelo time de IA da Petlove, agradecimentos especiais ao time de Plataforma (Ray), SRE (Lucas) e Security (Marcelo)

Gata Guru: Uma Aplicação de Chat Inteligente com IA para Análise de Dados was originally published in Petlabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

Since we’ve started using OpenAI here ate Petlove, one of the main concerns was about keep tracking of its costs. We could not find good benchmarks and most of the information available was about cots per tokens and strategies to estimate data processing based on that.

OpenAI platform do offers a usage tracking, but it is very limited and do not bring any detail about usage trends by customer, model and other breaks.

If you want to know the spent by member or by model you have to dig into daily usage breakdown dropdown menus, searching day by day, or copy/pasting information to organize a more consolidated view.

Looking for help at the internet I could find some posts, like this from Mark Craddock which he explains good strategies and suggest using the API response to build your own dashboard.

Also, searching at OpenAI community resources, I could find several discussions about this need and in this topic I found a partial solution from Davell, which I mixed with the idea of attributing the right pricing info from OpenAI platform table.

The objective of this post is to bring more details on how to build that cost tracking using OpenAI API response and a bit of python code to make a simple cost tracking tool with more detailed view.

To start, we are using the following packages:

import json  # to work on request responses and storage data
import requests # call API
import datetime # parse dates
from dateutil.relativedelta import relativedelta # date manipulation
import dateutil.parser as dparser # parse dates from strings

import os # get environment vars
import pandas as pd # you probably wrote this one before I tell
import time # sleeper to avoid exceeding API limits

Then you can get your OpenAI credentials stored in env vars:

openai_org_id = os.getenv('OPENAI_ORG_ID')
openai_api_key = os.getenv('OPENAI_API_KEY')

Since I was running it locally, I have specified a working directory in which I’m saving the json files with the API extractions. So I start reading the json files stored in this working directory to get the last date extracted from the API:

files = []
[files.append(f) for f in os.listdir() if f.endswith('.json')]
files.sort()

fdates = {}

idate = datetime.date(2023, 7, 1) # initial date

for f in files:
    dates = []
    for i in f.split('_'):
        try:
            dparser.parse(i, fuzzy=True)
            
            if dparser.parse(i, fuzzy=True).date()>idate:
                
                first_day_of_month=dparser.parse(i, fuzzy=True).date()
                
            dates.append(dparser.parse(i, fuzzy=True)) 
            
        except ValueError:
            
            pass
        
    fdates[f] = dates
    
first_day += relativedelta(days=1) # adding 1 day to start from the day after last extraction
current_day = datetime.date.today() #end date

Next step is to prepare API headers and extract the list of members in the organization:

headers = {
    "method": "GET",
    "authority": "api.openai.com",
    "scheme": "https",
    "path": f"/v1/organizations/{openai_org_id}/users",
    "authorization": f"Bearer {openai_api_key}",
}

starttime = time.monotonic() #starting to count time because the following request counts for the usage limits

users_response = requests.get(f"https://api.openai.com/v1/organizations/{openai_org_id}/users", headers=headers)
users = users_response.json()["members"]["data"]

Now we create a iteration over members and dates which we want to extract the API data:

df_costs = pd.DataFrame() #empty dataframe to store the data extracted

for user in users:
    
    id_of_user = user["user"]["id"]

    current_date = first_day
    
    while current_date < current_day and True:
        
        # I was giving 15 seconds just to ensure the requests per minute would not break the limits
        time.sleep(15.0 - ((time.monotonic() - starttime) % 15.0))
        
        usage_headers = {
            "method": "GET",
            "authority": "api.openai.com",
            "authorization": f"Bearer {openai_api_key}",
            "openai-organization": openai_org_id,
        }
        
        usage_response = requests.get(f"https://api.openai.com/v1/usage?date={current_date}&user_public_id={id_of_user}", headers=usage_headers)
        user_data = usage_response.json()
     
        # This conditional is to avoid breaking if the response is empty
        if len(user_data['data'])==0:
            
            current_date += relativedelta(days=1)
            
            continue
            
        else:
            
            data = user_data["data"]
            
            # This condition is to parse whisper model response in the same structure as the GPT/Embeddings models
            if len(user_data['whisper_api_data'])>0:
                
                wad = []
                for w in user_data['whisper_api_data']:
                    wd={}
                    wd['aggregation_timestamp'] = w['timestamp']
                    wd['n_requests'] = w['num_requests']
                    wd['operation'] = 'audio-whisper'
                    wd['snapshot_id'] = w['model_id']
                    wd['n_context'] = w['num_seconds']
                    wd['n_context_tokens_total'] = w['num_seconds']
                    wd['n_generated'] = 0
                    wd['n_generated_tokens_total'] = 0
                    wad = wad + [wd]
                data = data + wad
                
            else:
                
                pass
              
            df = pd.DataFrame(data)
            df['local_timestamp'] = df['aggregation_timestamp'].apply(lambda x: datetime.datetime.fromtimestamp(x))
            # converting to openai tz
            df['system_timestamp'] = df['local_timestamp'].dt.tz_localize('America/Sao_Paulo').dt.tz_convert("UTC")
            df['user'] = user["user"]["name"].lower().replace(" ", "_")
            df['email'] = user["user"]["email"]

            df_costs = pd.concat([df_costs,df])

            current_date += relativedelta(days=1)

After running the extratcions you can take a look into you df_costs dataframe using df_costs.head()

Here we are building the model costs parameters:

model_costs = {
    "gpt-3.5-turbo": {"context": 0.0015, "generated": 0.002},
    "gpt-3.5-turbo-0301": {"context": 0.0015, "generated": 0.002},
    "gpt-3.5-turbo-0613": {"context": 0.0015, "generated": 0.002},
    "gpt-3.5-turbo-16k": {"context": 0.003, "generated": 0.004},
    "gpt-3.5-turbo-16k-0613": {"context": 0.003, "generated": 0.004},
    "gpt-4": {"context": 0.03, "generated": 0.06},
    "gpt-4-0314": {"context": 0.03, "generated": 0.06},
    "gpt-4-0613": {"context": 0.03, "generated": 0.06},
    "gpt-4-32k": {"context": 0.06, "generated": 0.12},
    "gpt-4-32k-0314": {"context": 0.06, "generated": 0.12},
    "gpt-4-32k-0613": {"context": 0.06, "generated": 0.12},
    "text-embedding-ada-002-v2": {"context": 0.0001, "generated": 0},
    "text-davinci:003": {"context": 0.02, "generated": 0.02},
    "whisper-1": {"context": 0.1, "generated": 0}, # costs are 0.006 per min (or 0.0001 per second). Multiplying by x1000 to use same rules as tokens
}

mc = pd.DataFrame(model_costs)
mc = mc.T.reset_index()

And now we can apply the cost rates over the API consumption extracted into df_costs:

df_costs=df_costs.merge(mc, left_on='snapshot_id', right_on='index', how='left')
df_costs['context_costs']=(df_costs['n_context_tokens_total']/1000)*df_costs['context']
df_costs['generated_costs']=(df_costs['n_generated_tokens_total']/1000)*df_costs['generated']
df_costs['total_costs']=df_costs['context_costs']+df_costs['generated_costs']

The next block of code is to show how I was concatenating the current extraction to past extractions in my local working directory:

try:
    past_df = pd.read_json(files[len(files)-1])
    df_costs = pd.concat([df_costs, past_df])
except FileNotFoundError:
    pass
df_costs.reset_index(drop='level_0', inplace=True)

# ensuring datetime formats
df_costs['local_timestamp'] = pd.to_datetime(df_costs['local_timestamp'])
df_costs['system_timestamp'] = pd.to_datetime(df_costs['system_timestamp'])

# saving a json file    
with open(f"openai_costs_{idate}_to_{current_date - datetime.timedelta(days=1)}.json", "w") as f:
    json.dump(json.loads(df_costs.to_json(orient='columns', date_format='iso')), f)

This is the end of the data extraction step.

Now that we have the data, let’s explore how to make some charts using them. First lets import the packages for dataviz:

import matplotlib
import matplotlib.pyplot as plt
import seaborn as sns
%matplotlib inline

Then we create a function to aggregate data and compute the indicators:

def summary_costs(x):
    names = {
        'requests': x['n_requests'].sum()
        , 'context_tokens': x['n_context_tokens_total'].sum()
        , 'generated_tokens': x['n_generated_tokens_total'].sum()
        , 'context_costs': x['context_costs'].sum()
        , 'generated_costs': x['generated_costs'].sum()
        , 'total_costs': x['total_costs'].sum()
    }
    
    return pd.Series(names, index=['requests'
                                   , 'context_tokens'
                                   , 'generated_tokens'
                                   , 'context_costs'
                                   , 'generated_costs'
                                   , 'total_costs'
                                  ])
daily_cost = df_costs.groupby([pd.Grouper(key='local_timestamp', freq='D')]).apply(summary_costs).reset_index()
daily_cost['month_cumulated'] = daily_cost.groupby(pd.Grouper(key='local_timestamp', freq='M'))['total_costs'].cumsum()
daily_cost['day'] = daily_cost['local_timestamp'].apply(lambda x: x.day)
daily_cost['month'] = daily_cost['local_timestamp'].apply(lambda x: x.month)
daily_plot = daily_cost.pivot_table(index='day', columns='month', values='month_cumulated', aggfunc='sum')

This is the chart for Accumulated spent by day — Current x Past Month:

plt.figure(figsize=(10,10))
sns.lineplot(data=daily_plot, legend=True, sizes=(20, 2000))
plt.title('Cumulated spent by day - Current x Past Month')
plt.show()

And here we organize the data of the current month to check the consumption by user and model:

user_cost = df_costs.loc[df_costs['local_timestamp'].dt.month==datetime.date.today().month].groupby(['email'], group_keys=False).apply(summary_costs).sort_values('total_costs', ascending=False)[['total_costs']]
snap_cost = df_costs.loc[df_costs['local_timestamp'].dt.month==datetime.date.today().month].pivot_table(index='email', columns='snapshot_id', values='total_costs', aggfunc='sum')
user_snap_cost = user_cost.merge(snap_cost, left_index=True, right_index=True, how='left')

This is the chart for the Total spent by User/Model — Current Month:

plt.figure(figsize=(10,10))
user_snap_cost.drop('total_costs', axis=1).plot(kind='barh', stacked=True, color=['blue', 'red', 'purple', 'orange', 'yellow', 'skyblue', 'green'])
plt.gca().invert_yaxis()
plt.title('Total Spent by User/Model - Current Month')
plt.ylabel('User')
plt.xlabel('Cost USD')
plt.show()

Wrapping up

We can use API requests from openai to build our own cost tracking, customizing the way you want to track. In the scripts showed above we are using a local tracking, but you can use this script to build a pipeline (maybe using Airflow) to run this extraction on a daily basis and using the ata to populate any BI tool or fancy dashboard.

Hope you enjoyed and this could help you keeping track of your OpenAI costs.

You can find the full code in my github: https://github.com/bruno-curta/openai-costs-tracking

How to track OpenAI costs effectively? was originally published in Petlabs on Medium, where people are continuing the conversation by highlighting and responding to this story.