Inúmeras empresas de IA estão agora introduzindo sistemas de business intelligence baseados em agentes que podem analisar instantaneamente seus dados e fornecer insights abrangentes sem a necessidade de consultas SQL complexas ou análise manual de planilhas. Por exemplo, um sistema de análise de vendas com RAG pode responder a perguntas como “Qual foi nosso produto mais vendido no último trimestre?” em segundos, em vez de horas.
Um sistema de análise de vendas com RAG resolve esses problemas usando agentes de IA que podem raciocinar, tomar decisões e coordenar ferramentas externas de forma autônoma, permitindo consultas em linguagem natural em todo o seu ecossistema de dados. Em vez de aprender esquemas de banco de dados ou fórmulas de planilhas, o agente decompõe sua consulta, determina a melhor fonte de dados, realiza análises direcionadas e fornece respostas abrangentes e baseadas em dados com citações adequadas.
Por exemplo, você pode usar um LLM diferente, um formato de documento personalizado ou ajustar como os agentes analisam e se comportam para produzir saídas de negócios específicas. Neste tutorial, você aprenderá a construir seu próprio sistema de análise de vendas com RAG usando a API LLM unificada da Novita AI, o framework de agentes do LangChain e capacidades avançadas de processamento de documentos. Você criará um sistema que roteia automaticamente consultas para fontes de dados ideais e entrega insights de negócios acionáveis.
O que é um sistema de análise de vendas baseado em RAG?
Um sistema de análise de vendas baseado em RAG usa Retrieval-Augmented Generation para fornecer respostas precisas e fundamentadas a perguntas sobre vendas no varejo, combinando o poder de Grandes Modelos de Linguagem com uma camada de recuperação que pode acessar seus dados reais de negócios.
Este sistema geralmente envolve:
- O agente de IA analisa a consulta e determina se ela requer análise de dados estruturados (SQL), análise de documentos ou ambos.
- O agente roteia inteligentemente as consultas para a fonte de dados mais apropriada – bancos de dados SQL para análise quantitativa ou repositórios de documentos para insights qualitativos.
- O agente utiliza várias ferramentas para coletar informações, incluindo agentes Pandas para analisar arquivos CSV, agentes SQL para consultar bancos de dados e armazenamentos vetoriais para recuperar documentos. O agente não pode realizar tarefas essenciais no fluxo de trabalho sem essas ferramentas.
- O agente processa as informações coletadas usando técnicas especializadas, análise estatística para dados numéricos e busca semântica para documentos de texto.
- Após analisar os dados, o agente cria uma resposta estruturada que inclui resumos, cálculos e citações adequadas.
Ferramentas que você precisará
Antes de entrarmos na parte prática deste artigo, vamos configurar as ferramentas necessárias.
Novita AI
Para construir nosso sistema de análise baseado em RAG, precisaremos de acesso a grandes modelos de linguagem (LLMs) poderosos e modelos de embedding. A Novita AI oferece APIs de alto desempenho e acessíveis que fornecem acesso aos mais recentes grandes modelos de linguagem (LLMs), modelos de embedding e muito mais através de uma única interface unificada.
LangChain
O LangChain é um framework de código aberto projetado para construir aplicações com LLMs. Com o LangChain, você pode criar um fluxo de trabalho agêntico que raciocina passo a passo, usa ferramentas e interage com APIs. Para nosso sistema de análise de vendas, utilizaremos o LangChain para estruturar o processo de análise, empregar ferramentas como agentes SQL e processadores de documentos, e sintetizar todos os dados em insights estruturados.
Streamlit
Usaremos o Streamlit para construir a interface web interativa. É perfeito para prototipagem rápida e cria UIs de aparência profissional com código mínimo, tornando o tutorial simples de acompanhar.
FAISS
Para recuperação de documentos baseada em vetores, o FAISS fornece recursos rápidos de busca por similaridade, permitindo encontrar rapidamente trechos de documentos relevantes com base nas consultas dos usuários.
SQLAlchemy & PyMySQL
Estes lidam com nossas operações de banco de dados SQL e conectividade MySQL, permitindo que nosso sistema consulte dados estruturados de negócios diretamente.
pandas
Essencial para manipulação e análise de dados CSV. Nosso sistema usa agentes pandas para realizar cálculos complexos em dados tabulares.
Visão geral da arquitetura do sistema
Nosso sistema RAG processa inteligentemente as consultas dos usuários através de um fluxo de trabalho em várias etapas que determina automaticamente a melhor fonte de dados e método de processamento para cada pergunta. Em vez de forçar os usuários a saber onde seus dados residem, o sistema determina a abordagem ideal nos bastidores.
Processamento de consulta
A etapa inicial em nosso processo de trabalho com nosso Agente é o Processamento de Consulta. Neste caso, o Agente verificará a consulta feita pelo usuário e desenvolverá um plano personalizado de processamento que se encaixe no contexto da consulta. Em seguida, determinará o tipo de dados contra os quais a pergunta deve ser analisada, ou seja, dados estruturados, análise de documentos, ou ambos.
Seleção de fonte de dados
Quando um usuário envia uma pergunta, o sistema emprega uma abordagem direta: tenta resolver o problema consultando primeiro o banco de dados SQL. A Seleção de Fonte de Dados faz sentido porque dados estruturados geralmente fornecem as respostas mais precisas para perguntas quantitativas. O agente SQL pode executar rapidamente consultas e retornar números exatos com cálculos adequados.
Análise de documentos
Quando o agente SQL retorna resultados vazios ou não consegue encontrar informações relevantes, o sistema muda automaticamente para análise de documentos. O caminho de processamento de documentos varia dependendo do tipo de arquivo. Arquivos CSV e Excel são carregados em DataFrames Pandas para análise de dados complexos, enquanto PDFs, arquivos Word e documentos de texto são divididos em chunks e indexados usando FAISS para busca semântica.
Gerenciamento de fontes
É essencial citar adequadamente todas as fontes para garantir que as informações fornecidas possam ser rastreadas até uma autoridade confiável. Esta etapa no fluxo de trabalho ajuda a construir confiança com o público, permitindo que eles verifiquem quaisquer alegações ou dados apresentados no relatório final.
Síntese de resposta
Para conectar os pontos entre todas as informações coletadas, o Agente reúne insights de todas as fontes de dados analisadas para formar uma resposta coerente. Sem esta fase do fluxo de trabalho, a saída poderia ser um relatório desconexo.
Implementando o fluxo de trabalho de análise com RAG
Até agora, você entendeu o que é o sistema de análise de vendas baseado em RAG. Agora é hora de implementá-lo. Para começar, realizaremos algumas etapas principais:
Instalação e configuração
Antes de prosseguir, vamos estabelecer a estrutura completa de arquivos para nosso sistema de análise RAG. Você precisará criar os seguintes arquivos no diretório do seu projeto:
| rag-analytics-system/ ├── .env # Variáveis de ambiente ├── requirements.txt # Dependências Python ├── query_processor.py # Classe QueryProcessor principal ├── main.py # Interface Streamlit |
Configurando dependências
Primeiro, atualize o arquivo requirements.txt com todas as dependências necessárias:
| langchain==0.3.26 langchain-openai==0.3.25 python-dotenv==1.1.1 SQLAlchemy==2.0.41 pandas==2.3.0 PyPDF2==3.0.1 faiss-cpu==1.11.0 PyMySQL==1.1.1 cryptography==45.0.4 langchain-experimental==0.3.4 streamlit==1.46.0 openpyxl==3.1.5 python-docx==1.2.0 |
Em seguida, faça login na Novita AI. Após o login, navegue até a página Gerenciar Chaves de API, clique no botão Adicionar Nova Chave e insira o Nome da Chave.
A Novita AI fornecerá créditos gratuitos para testar vários modelos após o cadastro, então não se preocupe em comprar créditos antes de começar a construir ou experimentar. Adicione suas credenciais ao arquivo .env, como a Chave de API da Novita que você criou anteriormente:
| NOVITA_API_KEY=“sua_chave_novita_api_aqui” |
Agora, instale todas as dependências necessárias para o projeto:
| # Crie um ambiente virtual python -m venv rag_analytics_env source rag_analytics_env/bin/activate # No Windows: rag_analytics_env\Scripts\activate # Instale as dependências pip install -r requirements.txt |
Construindo o Processador de Consultas
Todo o fluxo de trabalho de análise RAG é encapsulado em uma classe modular chamada QueryProcessor. Esta classe gerenciará o processo de roteamento de consultas, realização de análise SQL, processamento de documentos, síntese de resultados e execução do loop completo de análise.
O QueryProcessor serve como o orquestrador central do nosso sistema de análise. Em vez de construir ferramentas separadas para diferentes tipos de dados, esta abordagem unificada nos permite rotear inteligentemente as consultas para o método de análise mais apropriado, seja bancos de dados SQL, arquivos CSV ou documentos não estruturados.
Primeiro, importaremos as bibliotecas necessárias para o projeto no arquivo query_processor.py. Essas importações nos fornecem tudo o que é necessário para integração com modelos de linguagem, conectividade com banco de dados, processamento de documentos e armazenamento vetorial:
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_experimental.agents.agent_toolkits import create_csv_agent, create_pandas_dataframe_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.utilities import SQLDatabase
from langchain.prompts import PromptTemplate
from langchain_community.tools import ReadFileTool
from langchain.schema import Document
import os
from dotenv import load_dotenv
from sqlalchemy.engine import Engine
import glob
import pandas as pd
load_dotenv()
Em seguida, inicializamos o agente de análise com acesso a um LLM via API da Novita e várias ferramentas de processamento de dados. O método de inicialização configura todos os componentes principais que precisaremos: o modelo de linguagem para entender consultas, embeddings para busca de similaridade de documentos e divisão de texto para processar documentos grandes eficientemente.
class QueryProcessor:
def __init__(self, documents_folder: str, sql_engine: Engine):
self.documents_folder = documents_folder
self.sql_engine = sql_engine
self.llm = ChatOpenAI(
model="google/gemma-3-27b-it",
temperature=0,
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "google"
}
)
self.embeddings = OpenAIEmbeddings(
model="baai/bge-m3",
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "baai"
}
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
self.sql_agent = None
self._prepare_data_sources()
O construtor inicializa vários componentes principais. Usamos uma temperatura de 0 para o LLM para garantir respostas consistentes e factuais, em vez de variações criativas. O divisor de texto é configurado com chunks sobrepostos para manter a continuidade do contexto ao processar documentos grandes. O método _prepare_data_sources() é chamado no final para configurar nossos agentes especializados.
Criando integração de múltiplas fontes de dados
O próximo passo envolve criar métodos que inicializam diferentes agentes de processamento de dados. Cada agente é especializado em lidar com tipos específicos de dados e tarefas de análise, assim como ter diferentes especialistas para vários tipos de perguntas.
def _prepare_data_sources(self):
"""Prepare both SQL and document data sources"""
# Prepare SQL agent
self._prepare_sql_agent()
# Prepare document agent
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""Initialize SQL agent"""
# Convert SQLAlchemy Engine to LangChain SQLDatabase
db = SQLDatabase(self.sql_engine)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
self.sql_agent = create_sql_agent(
llm=self.llm,
toolkit=toolkit,
agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
def _prepare_document_agent(self):
"""Initialize document agent using ReadFileTool and vector store"""
# Get all supported files
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"\
Found {len(supported_files)} supported files in {self.documents_folder}")
if supported_files:
# Create ReadFileTool
read_file_tool = ReadFileTool()
# Create tools list
tools = [read_file_tool]
# Create the prompt template for the react agent
prompt = PromptTemplate.from_template("""
You are a helpful assistant that can answer questions about business documents.
You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought: I should read the relevant documents to find the answer
{agent_scratchpad}
""")
agent = create_react_agent(self.llm, tools, prompt)
self.document_agent = AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
verbose=True
)
print("\
Document agent initialized successfully")
else:
print("\
No documents found in the specified folder.")
O método _prepare_sql_agent() cria um agente especializado para consultas ao banco de dados. Usamos o tipo de agente ZERO_SHOT_REACT_DESCRIPTION, o que significa que o agente pode raciocinar sobre consultas SQL sem precisar de exemplos específicos, tornando-o adaptável a diferentes esquemas de banco de dados. O método _prepare_document_agent() verifica tipos de arquivo suportados e cria um agente flexível que pode ler vários formatos de documento. O template de prompt ReAct (Raciocínio e Ação) guia o agente através de um processo de pensamento estruturado, semelhante à forma como um analista humano abordaria a análise de documentos.
Processamento de documentos e armazenamento vetorial
Em seguida, precisamos lidar com como o fluxo de trabalho processa diferentes tipos de documento. Esta seção cria métodos que lidam com arquivos CSV/Excel usando agentes pandas e documentos não estruturados usando armazenamento vetorial. O insight principal aqui é que diferentes tipos de dados requerem estratégias de processamento distintas.
CSV_PROMPT_PREFIX = """
IMPORTANT: You are working with a pandas DataFrame called 'df' that has been loaded with the actual data.
DO NOT create sample data or make up data. Use ONLY the actual DataFrame 'df' that is available to you.
First, explore the DataFrame by:
1. Setting pandas display options to show all columns: pd.set_option('display.max_columns', None)
2. Check the shape of the DataFrame: print(df.shape)
3. Get the column names: print(df.columns.tolist())
4. Check the data types: print(df.dtypes)
5. Look at the first few rows: print(df.head())
6. Then answer the question using the actual data in the DataFrame.
"""
CSV_PROMPT_SUFFIX = """
- **CRITICAL**: Use ONLY the actual data in the DataFrame. Do NOT create sample data or use fictional data.
- **ALWAYS** before giving the Final Answer, try another method to verify your results.
- Then reflect on the answers of the two methods you did and ask yourself if it answers correctly the original question.
- If you are not sure, try another method.
- FORMAT 4 FIGURES OR MORE WITH COMMAS.
- If the methods tried do not give the same result, reflect and try again until you have two methods that have the same result.
- If you still cannot arrive to a consistent result, say that you are not sure of the answer.
- If you are sure of the correct answer, create a beautiful and thorough response using Markdown.
- **DO NOT MAKE UP AN ANSWER OR USE PRIOR KNOWLEDGE, ONLY USE THE RESULTS OF THE CALCULATIONS YOU HAVE DONE**.
- **ALWAYS**, as part of your "Final Answer", explain how you got to the answer on a section that starts with: "\\
\\
Explanation:\\
".
- In the explanation, mention the column names that you used to get to the final answer.
- Show your work by displaying relevant DataFrame operations and their results.
"""
def _process_document_query(self, query: str) -> str:
"""Process document query using CSV/Excel agents for tabular data, and LLM for unstructured data."""
try:
print(f"\\
Processing query: {query}")
# Get all supported files
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"Found {len(supported_files)} supported files in {self.documents_folder}")
if not supported_files:
return "No documents found to search through."
# Check for CSV/Excel files first
csv_files = [f for f in supported_files if f.endswith('.csv')]
excel_files = [f for f in supported_files if f.endswith(('.xlsx', '.xls'))]
if csv_files:
csv_file = csv_files[0]
print(f"\\
Processing CSV file: {csv_file}")
try:
# First try with pandas DataFrame agent
df = pd.read_csv(csv_file)
print(f"CSV loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# Create the agent with improved configuration
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Avoid token limits with large DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Process the query with our custom prompt
print(f"Processing query with agent: {query}")
# Improved prompt that ensures agent uses the actual DataFrame
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error with pandas DataFrame agent: {str(e)}")
print("Trying alternative CSV agent...")
# Fallback to CSV agent
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"Error processing CSV file: {str(e2)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing CSV file: {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Processing Excel file: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# Create the agent with improved configuration
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Avoid token limits with large DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Process the query with our custom prompt
print(f"Processing query with agent: {query}")
# Improved prompt that ensures agent uses the actual DataFrame
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error processing Excel file: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing Excel file: {str(e)}"
# For unstructured text files
print("\\
Processing unstructured text files...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Error reading file {file_path}: {str(e)}")
continue
if not all_content:
return "Could not read any documents."
# Process unstructured text using vector store
print("Processing text with vector store...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Generated context length: {len(context)}")
response = self.llm.invoke(
f"""Based on the following context, answer the question: {query}
\\
Context:\\
{context}\\
\\
Answer:"""
)
return response.content
except Exception as e:
print(f"Error processing document query: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing document query: {str(e)}"
A seção de processamento CSV inclui engenharia de prompt detalhada para garantir análise precisa. Os prompts de prefixo e sufixo são cruciais porque impedem que o agente alucine dados ou forneça resultados incorretos. A etapa de verificação, onde o agente tenta múltiplos métodos, ajuda a garantir a precisão, assim como um analista cuidadoso verificaria seus cálculos. Priorizamos agentes DataFrame do pandas em vez de agentes CSV porque eles fornecem manipulação de dados mais robusta e melhor desempenho com grandes conjuntos de dados. O mecanismo de fallback garante que, se uma abordagem falhar, temos métodos alternativos para processar os dados. Agora vamos completar o método _process_document_query para lidar com arquivos Excel e documentos não estruturados:
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Processing Excel file: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# Create the agent with improved configuration
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Avoid token limits with large DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Process the query with our custom prompt
print(f"Processing query with agent: {query}")
# Improved prompt that ensures agent uses the actual DataFrame
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error processing Excel file: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing Excel file: {str(e)}"
# For unstructured text files
print("\\
Processing unstructured text files...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Error reading file {file_path}: {str(e)}")
continue
if not all_content:
return "Could not read any documents."
# Process unstructured text using vector store
print("Processing text with vector store...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Generated context length: {len(context)}")
response = self.llm.invoke(
f"""Based on the following context, answer the question: {query}
\\
Context:\\
{context}\\
\\
Answer:"""
)
return response.content
except Exception as e:
print(f"Error processing document query: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing document query: {str(e)}"
Integração com banco de dados SQL
Agora que lidamos com o processamento de documentos, precisamos criar o sistema inteligente de roteamento de consultas que decide entre análise SQL e de documentos. Esta é a inteligência central do nosso sistema – determinar qual fonte de dados tem maior probabilidade de conter a resposta para uma determinada pergunta.
def process_query(self, query: str) -> str:
"""
Process query using agents to intelligently decide between SQL and documents
"""
# First try SQL agent
try:
print("\\
Trying SQL Agent...")
sql_result = self.sql_agent.run(query)
no_answer_phrases = [
"no results", "i don't know", "unknown", "not sure", "cannot answer", "don't have", "no data", "n/a"
]
if sql_result and not any(phrase in sql_result.lower() for phrase in no_answer_phrases) and sql_result.strip():
return f"From SQL Database: {sql_result}"
else:
print("SQL Agent could not answer, trying documents...")
except Exception as e:
print(f"SQL Agent Error: {str(e)}")
print("Falling back to documents...")
# If SQL agent fails or returns no results, try document processing
try:
print("\\
Processing documents...")
doc_result = self._process_document_query(query)
if doc_result:
return f"From Documents: {doc_result}"
else:
print("Document processing returned no results")
except Exception as e:
print(f"Document Processing Error: {str(e)}")
return "Could not find relevant information in either SQL database or documents."
A lógica de roteamento de consultas segue um sistema de prioridade: bancos de dados SQL são tentados primeiro porque normalmente contêm dados estruturados e quantitativos que podem responder a perguntas de negócios de forma rápida e precisa. Se o agente SQL retornar respostas pouco claras ou negativas (detectadas através da nossa lista no_answer_phrases), o sistema automaticamente recorre ao processamento de documentos. Esta abordagem imita como um analista humano trabalharia – primeiro verificando fontes de dados estruturadas, depois recorrendo a documentos e relatórios quando o banco de dados não tem as informações necessárias.
Criando a interface Streamlit
Agora vamos construir a interface do usuário que torna nosso sistema de análise acessível aos usuários de negócios. A interface Streamlit oferece uma experiência intuitiva baseada em chat que oculta a complexidade técnica enquanto entrega poderosas capacidades de análise. Adicione os seguintes trechos de código ao arquivo main.py:
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# Load environment variables
load_dotenv()
#st.text_input
# Initialize session state
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# Set page config
st.set_page_config(
page_title="Document Analysis Chatbot",
page_icon="🤖",
layout="wide"
)
# Custom CSS for button and title styling
st.markdown("""
<style>
.stButton > button {
background-color: #23D57C;
color: white;
border: none;
border-radius: 8px;
padding: 0.5rem 1rem;
font-weight: 600;
transition: all 0.3s ease;
}
.stButton > button:hover {
background-color: #1fb36b;
box-shadow: 0 4px 8px rgba(35, 213, 124, 0.3);
transform: translateY(-2px);
}
.stButton > button:active {
background-color: #1a9960;
transform: translateY(0px);
}
h1 {
color: #23D57C !important;
font-weight: 700;
}
</style>
""", unsafe_allow_html=True)
# Title and description
st.title("Document Analysis Chatbot using Novita")
Esta configuração inicial cria uma interface de aparência profissional com estilo personalizado. O gerenciamento de estado da sessão garante que os usuários não percam o histórico da conversa ou precisem reinicializar suas fontes de dados ao interagir com a interface. O CSS personalizado fornece feedback visual e mantém uma experiência de marca consistente.
Integração final do sistema
Vamos completar a interface Streamlit com a inicialização da fonte de dados e a funcionalidade de chat. Esta seção final consolida todos os nossos componentes em uma aplicação amigável que analistas de negócios podem utilizar sem exigir conhecimento técnico.
# Check if data source is initialized
if st.session_state.processor is None:
# Center the data source configuration
st.markdown("<br><br>", unsafe_allow_html=True)
# Create centered columns
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 Get Started")
st.write("Initialize your data source to start chatting with your documents")
# SQL Database Configuration (hidden)
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# Documents Folder Configuration
st.write("**Documents Folder Path:**")
documents_folder = st.text_input(
"Documents Folder Path",
placeholder="Enter path to your documents folder (e.g., docs)",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# Center the button
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("Initialize Data Source", use_container_width=True):
try:
# Validate and resolve documents folder path
if not documents_folder:
st.error("Please provide a documents folder path")
else:
# Convert to absolute path
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"Documents folder not found: {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"Path is not a directory: {abs_documents_folder}")
else:
# Initialize query processor with a dummy SQL engine
# Create SQL engine
connection_string = f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}"
sql_engine = create_engine(connection_string)
st.session_state.processor = QueryProcessor(abs_documents_folder, sql_engine)
st.success("Data source initialized successfully!")
st.rerun()
except Exception as e:
st.error(f"Error initializing data source: {str(e)}")
else:
# Show data source status in sidebar
with st.sidebar:
st.header("📊 Data Source")
st.success("✅ Data source initialized")
if st.button("Reset Data Source"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# Main chat interface
st.header("Chat Interface")
# Display chat messages
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# Chat input
if prompt := st.chat_input("Ask a question about your documents"):
# Add user message to chat history
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# Process query
try:
response = st.session_state.processor.process_query(prompt)
# Add assistant response to chat history
st.session_state.messages.append({"role": "assistant", "content": response})
with st.chat_message("assistant"):
st.write(response)
except Exception as e:
st.error(f"Error processing query: {str(e)}")
# Add a clear chat button only if there are messages
if st.session_state.messages:
if st.button("Clear Chat"):
st.session_state.messages = []
st.rerun()
O processo de inicialização inclui tratamento robusto de erros e validação de caminho para prevenir erros comuns do usuário. A barra lateral fornece informações de status claras, e a interface de chat segue padrões familiares que os usuários esperam de assistentes de IA modernos. O sistema mantém o histórico da conversa, permitindo que os usuários construam perguntas anteriores e criem um fluxo de trabalho analítico natural. A funcionalidade de limpar chat permite que os usuários comecem de novo quando necessário.
Executando o sistema de análise
Agora que construímos o sistema de análise de vendas com RAG, vamos testá-lo e ver como ele se sai em consultas de negócios do mundo real.
Primeiro, crie um arquivo data_generator.py e copie o código do script Python aqui para gerar dados de amostra que forneceremos ao sistema, e execute o script com o comando:
| python data_generator.py |
Isso criará uma pasta sample_documents com:
- large_sales_dataset.csv - 10.000 registros de vendas
- business_strategy_2024.txt - Documento estratégico de negócios
- sales_meeting_notes.txt - Notas de reunião e itens de ação
Em seguida, inicie a aplicação Streamlit:
| streamlit run main.py |
A aplicação será aberta no seu navegador em http://localhost:8501.
Uma vez inicializado, você pode testar o sistema inserindo o caminho dos seus documentos como sample_documents e fazer várias perguntas de negócios:
| # Consultas de exemplo para testar o sistema topic_1 = “Qual foi o valor total de vendas para eletrônicos em 2024?” topic_2 = “Quais são nossas principais iniciativas estratégicas para experiência do cliente?” topic_3 = “Qual representante de vendas teve o maior total de vendas, e qual foi o valor?” |
Conclusão
Ao longo deste tutorial, construímos um sistema de análise baseado em RAG que demonstra como a IA moderna pode transformar fluxos de trabalho de business intelligence. Ao combinar múltiplas abordagens de processamento de dados sob uma única interface inteligente, criamos uma ferramenta que pode lidar com todo o espectro de perguntas de negócios que os analistas enfrentam diariamente. Os principais pontos fortes do sistema incluem:
- Roteamento Inteligente de Consultas que determina automaticamente a melhor fonte de dados para cada pergunta
- Suporte a múltiplos formatos para lidar com bancos de dados SQL, arquivos CSV, planilhas Excel e documentos de texto
- Tratamento Robusto de Erros para fornecer mecanismos de fallback e mensagens de erro claras
- Interação baseada em chat que não requer conhecimento técnico
- Verificação e validação embutidas para garantir resultados confiáveis
Você pode estender esta aplicação para incorporar fontes adicionais, técnicas de análise mais avançadas e lógica de negócios personalizada. Agora que você tem esse conhecimento, vá em frente e tente construir seus próprios agentes com a Novita AI ou integrá-la em seus projetos existentes.
Novita AI é a plataforma cloud all-in-one que impulsiona suas ambições de IA. APIs integradas, serverless, instância GPU — as ferramentas econômicas que você precisa. Elimine infraestrutura, comece gratuitamente e torne sua visão de IA realidade.
