Innumerables empresas de inteligencia artificial están presentando ahora sistemas de inteligencia empresarial agéntica que pueden analizar instantáneamente tus datos y ofrecer perspectivas completas sin necesidad de consultas SQL complejas ni análisis manual de hojas de cálculo. Por ejemplo, un sistema de análisis de ventas impulsado por RAG puede responder preguntas como “¿Cuál fue nuestro producto más vendido el trimestre pasado?” en segundos en lugar de horas.
Un sistema de análisis de ventas basado en RAG resuelve estos problemas utilizando agentes de IA que pueden razonar de forma autónoma, tomar decisiones y coordinar herramientas externas para permitir consultas en lenguaje natural en todo tu ecosistema de datos. En lugar de aprender esquemas de bases de datos o fórmulas de hojas de cálculo, el agente descompone tu consulta, determina la mejor fuente de datos, realiza un análisis específico y entrega respuestas completas y basadas en datos con las citas adecuadas.
Por ejemplo, podrías usar un LLM diferente, un formato de documento personalizado o ajustar cómo los agentes analizan y se comportan para producir resultados comerciales específicos. En este tutorial, aprenderás a construir tu propio sistema de análisis de ventas impulsado por RAG utilizando la API LLM unificada de Novita AI, el framework de agentes de LangChain y capacidades avanzadas de procesamiento de documentos. Crearás un sistema que enruta automáticamente las consultas a las fuentes de datos óptimas y ofrece información empresarial procesable.
¿Qué es un sistema de análisis de ventas impulsado por RAG?
Un sistema de análisis de ventas basado en RAG utiliza Generación Aumentada por Recuperación para proporcionar respuestas precisas y fundamentadas a preguntas sobre ventas minoristas, combinando el poder de los modelos de lenguaje grandes (LLM) con una capa de recuperación que puede acceder a tus datos empresariales reales.
Este sistema generalmente implica:
- El agente de IA analiza la consulta y determina si requiere análisis de datos estructurados (SQL), análisis de documentos o ambos.
- El agente enruta inteligentemente las consultas a la fuente de datos más apropiada: bases de datos SQL para análisis cuantitativo o almacenes de documentos para obtener información cualitativa.
- El agente utiliza diversas herramientas para recopilar información, incluidos agentes de Pandas para analizar archivos CSV, agentes SQL para consultar bases de datos y almacenes vectoriales para recuperar documentos. El agente no puede realizar tareas esenciales en el flujo de trabajo sin estas herramientas.
- El agente procesa la información recopilada mediante técnicas especializadas: análisis estadístico para datos numéricos y búsqueda semántica para documentos de texto.
- Después de analizar los datos, el agente crea una respuesta estructurada que incluye resúmenes, cálculos y citas adecuadas.
Herramientas necesarias
Antes de entrar en la parte de construcción de este artículo, configuremos las herramientas necesarias.
Novita AI
Para construir nuestro sistema de análisis basado en RAG, necesitaremos acceso a potentes modelos de lenguaje grandes (LLM) y modelos de incrustación (embeddings). Novita AI ofrece API asequibles y de alto rendimiento que proporcionan acceso a los últimos modelos de lenguaje grandes (LLM), modelos de incrustación y más a través de una única interfaz unificada.
LangChain
LangChain es un framework de código abierto diseñado para construir aplicaciones con LLM. Con LangChain, puedes crear un flujo de trabajo agéntico que razone paso a paso, use herramientas e interactúe con API. Para nuestro sistema de análisis de ventas, utilizaremos LangChain para estructurar el proceso de análisis, emplear herramientas como agentes SQL y procesadores de documentos, y sintetizar todos los datos en información estructurada.
Streamlit
Usaremos Streamlit para construir la interfaz web interactiva. Es perfecto para prototipado rápido y crea interfaces de usuario de aspecto profesional con un mínimo de código, lo que hace que el tutorial sea sencillo de seguir.
FAISS
Para la recuperación de documentos basada en vectores, FAISS proporciona capacidades rápidas de búsqueda de similitud, lo que nos permite encontrar rápidamente fragmentos de documentos relevantes según las consultas de los usuarios.
SQLAlchemy y PyMySQL
Estos manejan nuestras operaciones de base de datos SQL y conectividad MySQL, permitiendo que nuestro sistema consulte datos empresariales estructurados directamente.
pandas
Esencial para la manipulación y análisis de datos CSV. Nuestro sistema utiliza agentes de pandas para realizar cálculos complejos en datos tabulares.
Descripción general de la arquitectura del sistema
Nuestro sistema RAG procesa inteligentemente las consultas de los usuarios mediante un flujo de trabajo de varios pasos que determina automáticamente la mejor fuente de datos y método de procesamiento para cada pregunta. En lugar de obligar a los usuarios a saber dónde residen sus datos, el sistema determina el enfoque óptimo entre bastidores.
Procesamiento de consultas
El paso inicial en nuestro proceso de trabajo con el Agente es el Procesamiento de Consultas. En este caso, el Agente verificará la consulta realizada por el usuario y desarrollará un plan de procesamiento personalizado que se adapte al contexto de la consulta. Luego determinará el tipo de datos con los que se debe analizar la pregunta: datos estructurados, análisis de documentos o ambos.
Selección de fuente de datos
Cuando un usuario envía una pregunta, el sistema emplea un enfoque simple: primero intenta resolver el problema consultando la base de datos SQL. La Selección de fuente de datos tiene sentido porque los datos estructurados a menudo proporcionan las respuestas más precisas a preguntas cuantitativas. El agente SQL puede ejecutar rápidamente consultas y devolver números exactos con cálculos adecuados.
Análisis de documentos
Cuando el agente SQL devuelve resultados vacíos o no puede encontrar información relevante, el sistema cambia automáticamente al análisis de documentos. La ruta de procesamiento de documentos varía según el tipo de archivo. Los archivos CSV y Excel se cargan en DataFrames de Pandas para un análisis de datos complejo, mientras que los PDF, archivos de Word y documentos de texto se dividen en fragmentos y se indexan usando FAISS para búsqueda semántica.
Gestión de fuentes
Es esencial citar adecuadamente todas las fuentes para garantizar que la información proporcionada se pueda rastrear hasta una autoridad creíble. Este paso en el flujo de trabajo ayuda a generar confianza con la audiencia, permitiéndoles verificar cualquier afirmación o dato presentado en el informe final.
Síntesis de respuestas
Para conectar los puntos entre toda la información recopilada, el Agente reúne los conocimientos de todas las fuentes de datos analizadas para formar una respuesta coherente. Sin esta fase del flujo de trabajo, el resultado podría ser un informe inconexo.
Implementación del flujo de trabajo de análisis basado en RAG
Hasta ahora, has comprendido de qué se trata el sistema de análisis de ventas impulsado por RAG. Ahora es el momento de implementarlo. Para comenzar, realizaremos algunos pasos clave:
Instalación y configuración
Antes de continuar, establezcamos la estructura de archivos completa para nuestro sistema de análisis RAG. Deberás crear los siguientes archivos en el directorio de tu proyecto:
| rag-analytics-system/ ├── .env # Variables de entorno ├── requirements.txt # Dependencias de Python ├── query_processor.py # Clase QueryProcessor principal ├── main.py # Interfaz de Streamlit |
Configuración de dependencias
Primero, actualiza el archivo requirements.txt con todas las dependencias necesarias:
| 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 |
Luego, inicia sesión en Novita AI. Una vez que hayas iniciado sesión, navega a la página Manage API Keys, haz clic en el botón Add New Key e ingresa tu Key Name.
Novita AI te proporcionará créditos gratuitos para probar varios modelos después de registrarte, por lo que no necesitas preocuparte por comprar créditos antes de comenzar a construir o experimentar. Agrega tus credenciales al archivo .env, como la clave de API de Novita que creaste anteriormente:
| NOVITA_API_KEY=”tu_clave_api_novita_aquí” |
Ahora, instala todas las dependencias requeridas para el proyecto:
| # Crear un entorno virtual python -m venv rag_analytics_env source rag_analytics_env/bin/activate # En Windows: rag_analytics_env\Scripts\activate # Instalar dependencias pip install -r requirements.txt |
Construcción del procesador de consultas (QueryProcessor)
Todo el flujo de trabajo de análisis RAG está encapsulado en una clase modular llamada QueryProcessor. Esta clase gestionará el proceso de enrutamiento de consultas, realización de análisis SQL, procesamiento de documentos, síntesis de resultados y ejecución del bucle de análisis completo.
El QueryProcessor sirve como el orquestador central de nuestro sistema de análisis. En lugar de construir herramientas separadas para diferentes tipos de datos, este enfoque unificado nos permite enrutar inteligentemente las consultas al método de análisis más apropiado, ya sea bases de datos SQL, archivos CSV o documentos no estructurados.
Primero, importaremos las bibliotecas necesarias para el proyecto en el archivo query_processor.py. Estas importaciones nos proporcionan todo lo necesario para la integración de modelos de lenguaje, conectividad de bases de datos, procesamiento de documentos y almacenamiento vectorial:
connectivity, document processing, and vector storage:
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()
A continuación, inicializamos el agente de análisis con acceso a un LLM a través de la API de Novita y varias herramientas de procesamiento de datos. El método de inicialización configura todos los componentes centrales que necesitaremos: el modelo de lenguaje para entender consultas, las incrustaciones (embeddings) para la búsqueda de similitud de documentos y la división de textos para procesar documentos grandes de manera eficiente.
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()
El constructor inicializa varios componentes clave. Usamos una temperatura de 0 para el LLM para garantizar respuestas consistentes y objetivas en lugar de variaciones creativas. El divisor de texto está configurado con fragmentos superpuestos para mantener la continuidad del contexto al procesar documentos grandes. El método _prepare_data_sources() se llama al final para configurar nuestros agentes especializados.
Creación de integración de datos de múltiples fuentes
El siguiente paso implica crear métodos que inicializan diferentes agentes de procesamiento de datos. Cada agente se especializa en manejar tipos de datos y tareas de análisis específicos, de manera similar a tener diferentes expertos para varios tipos de preguntas.
def _prepare_data_sources(self):
"""Preparar fuentes de datos SQL y de documentos"""
# Preparar agente SQL
self._prepare_sql_agent()
# Preparar agente de documentos
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""Inicializar agente SQL"""
# Convertir SQLAlchemy Engine a 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):
"""Inicializar agente de documentos usando ReadFileTool y almacén vectorial"""
# Obtener todos los archivos compatibles
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"\
Se encontraron {len(supported_files)} archivos compatibles en {self.documents_folder}")
if supported_files:
# Crear ReadFileTool
read_file_tool = ReadFileTool()
# Crear lista de herramientas
tools = [read_file_tool]
# Crear la plantilla de prompt para el agente react
prompt = PromptTemplate.from_template("""
Eres un asistente útil que puede responder preguntas sobre documentos empresariales.
Tienes acceso a las siguientes herramientas:
{tools}
Usa el siguiente formato:
Pregunta: la pregunta de entrada que debes responder
Pensamiento: siempre debes pensar qué hacer
Acción: la acción a realizar, debe ser una de [{tool_names}]
Entrada de acción: la entrada para la acción
Observación: el resultado de la acción
... (este Pensamiento/Acción/Entrada de acción/Observación puede repetirse N veces)
Pensamiento: Ahora sé la respuesta final
Respuesta final: la respuesta final a la pregunta de entrada original
¡Comienza!
Pregunta: {input}
Pensamiento: Debo leer los documentos relevantes para encontrar la respuesta
{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("\
Agente de documentos inicializado correctamente")
else:
print("\
No se encontraron documentos en la carpeta especificada.")
El método _prepare_sql_agent() crea un agente especializado para consultas de bases de datos. Usamos el tipo de agente ZERO_SHOT_REACT_DESCRIPTION, lo que significa que el agente puede razonar sobre consultas SQL sin necesidad de ejemplos específicos, lo que lo hace adaptable a diferentes esquemas de bases de datos. El método _prepare_document_agent() escanea los tipos de archivos compatibles y crea un agente flexible que puede leer varios formatos de documentos. La plantilla de prompt ReAct (Razonamiento y Actuación) guía al agente a través de un proceso de pensamiento estructurado, similar a cómo un analista humano abordaría el análisis de documentos.
Procesamiento de documentos y almacenamiento vectorial
A continuación, necesitamos manejar cómo el flujo de trabajo procesa diferentes tipos de documentos. Esta sección crea métodos que manejan archivos CSV/Excel utilizando agentes de pandas y documentos no estructurados usando almacenamiento vectorial. La idea clave aquí es que diferentes tipos de datos requieren diferentes estrategias de procesamiento.
CSV_PROMPT_PREFIX = """
IMPORTANTE: Estás trabajando con un DataFrame de pandas llamado 'df' que se ha cargado con los datos reales.
NO crees datos de muestra ni inventes datos. Usa SOLO el DataFrame real 'df' que está disponible para ti.
Primero, explora el DataFrame:
1. Configura las opciones de visualización de pandas para mostrar todas las columnas: pd.set_option('display.max_columns', None)
2. Verifica la forma del DataFrame: print(df.shape)
3. Obtén los nombres de las columnas: print(df.columns.tolist())
4. Verifica los tipos de datos: print(df.dtypes)
5. Mira las primeras filas: print(df.head())
6. Luego responde la pregunta usando los datos reales en el DataFrame.
"""
CSV_PROMPT_SUFFIX = """
- **CRÍTICO**: Usa SOLO los datos reales en el DataFrame. NO crees datos de muestra ni uses datos ficticios.
- **SIEMPRE** antes de dar la Respuesta Final, intenta otro método para verificar tus resultados.
- Luego reflexiona sobre las respuestas de los dos métodos que realizaste y pregúntate si responden correctamente la pregunta original.
- Si no estás seguro, intenta otro método.
- FORMATEA 4 CIFRAS O MÁS CON COMAS.
- Si los métodos probados no dan el mismo resultado, reflexiona e inténtalo de nuevo hasta que tengas dos métodos que tengan el mismo resultado.
- Si aún no puedes llegar a un resultado consistente, di que no estás seguro de la respuesta.
- Si estás seguro de la respuesta correcta, crea una respuesta hermosa y completa usando Markdown.
- **NO INVENTES UNA RESPUESTA NI USES CONOCIMIENTOS PREVIOS, USA SOLO LOS RESULTADOS DE LOS CÁLCULOS QUE HAS REALIZADO**.
- **SIEMPRE**, como parte de tu "Respuesta Final", explica cómo llegaste a la respuesta en una sección que comience con: "\\
\\
Explicación:\\
".
- En la explicación, menciona los nombres de las columnas que usaste para llegar a la respuesta final.
- Muestra tu trabajo mostrando las operaciones relevantes del DataFrame y sus resultados.
"""
def _process_document_query(self, query: str) -> str:
"""Procesar consulta de documentos usando agentes CSV/Excel para datos tabulares, y LLM para datos no estructurados."""
try:
print(f"\\
Procesando consulta: {query}")
# Obtener todos los archivos compatibles
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"Se encontraron {len(supported_files)} archivos compatibles en {self.documents_folder}")
if not supported_files:
return "No se encontraron documentos para buscar."
# Verificar archivos CSV/Excel primero
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"\\
Procesando archivo CSV: {csv_file}")
try:
# Primero intentar con el agente DataFrame de pandas
df = pd.read_csv(csv_file)
print(f"CSV cargado correctamente con {len(df)} filas y columnas: {df.columns.tolist()}")
# Crear el agente con configuración mejorada
print("Creando agente DataFrame de pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Evitar límites de token con DataFrames grandes
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Procesar la consulta con nuestro prompt personalizado
print(f"Procesando consulta con el agente: {query}")
# Prompt mejorado que asegura que el agente use el DataFrame real
prompt = f"""
Tienes acceso a un DataFrame de pandas llamado 'df' con {len(df)} filas y las siguientes columnas: {df.columns.tolist()}.
Aquí están las primeras filas de los datos:
{df.head().to_string()}
Tipos de datos:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Pregunta: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Respuesta del agente: {response}")
return response['output']
except Exception as e:
print(f"Error con el agente DataFrame de pandas: {str(e)}")
print("Intentando con el agente CSV alternativo...")
# Recurrir al agente CSV
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Pregunta: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"Error procesando archivo CSV: {str(e2)}")
import traceback
print(f"Traceback completo: {traceback.format_exc()}")
return f"Error procesando el archivo CSV: {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Procesando archivo Excel: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel cargado correctamente con {len(df)} filas y columnas: {df.columns.tolist()}")
# Crear el agente con configuración mejorada
print("Creando agente DataFrame de pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Evitar límites de token con DataFrames grandes
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Procesar la consulta con nuestro prompt personalizado
print(f"Procesando consulta con el agente: {query}")
# Prompt mejorado que asegura que el agente use el DataFrame real
prompt = f"""
Tienes acceso a un DataFrame de pandas llamado 'df' con {len(df)} filas y las siguientes columnas: {df.columns.tolist()}.
Aquí están las primeras filas de los datos:
{df.head().to_string()}
Tipos de datos:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Pregunta: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Respuesta del agente: {response}")
return response['output']
except Exception as e:
print(f"Error procesando archivo Excel: {str(e)}")
import traceback
print(f"Traceback completo: {traceback.format_exc()}")
return f"Error procesando el archivo Excel: {str(e)}"
# Para archivos de texto no estructurados
print("\\
Procesando archivos de texto no estructurados...")
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 leyendo archivo {file_path}: {str(e)}")
continue
if not all_content:
return "No se pudo leer ningún documento."
# Procesar texto no estructurado usando almacén vectorial
print("Procesando texto con almacén vectorial...")
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"Longitud del contexto generado: {len(context)}")
response = self.llm.invoke(
f"""Basado en el siguiente contexto, responde la pregunta: {query}
\\
Contexto:\\
{context}\\
\\
Respuesta:"""
)
return response.content
except Exception as e:
print(f"Error procesando consulta de documentos: {str(e)}")
import traceback
print(f"Traceback completo: {traceback.format_exc()}")
return f"Error procesando la consulta de documentos: {str(e)}"
La sección de procesamiento CSV incluye una ingeniería de prompts detallada para garantizar un análisis preciso. Los prompts de prefijo y sufijo son cruciales porque evitan que el agente alucine datos o proporcione resultados incorrectos. El paso de verificación, donde el agente prueba múltiples métodos, ayuda a garantizar la precisión, similar a como un analista cuidadoso volvería a verificar sus cálculos. Priorizamos los agentes DataFrame de pandas sobre los agentes CSV porque proporcionan un manejo de datos más robusto y un mejor rendimiento con conjuntos de datos grandes. El mecanismo de respaldo asegura que si un enfoque falla, tenemos métodos alternativos para procesar los datos. Ahora completemos el método _process_document_query para manejar archivos Excel y documentos no estructurados:
elif excel_files:
excel_file = excel_files[0]
print(f"\\
Procesando archivo Excel: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel cargado correctamente con {len(df)} filas y columnas: {df.columns.tolist()}")
# Crear el agente con configuración mejorada
print("Creando agente DataFrame de pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Evitar límites de token con DataFrames grandes
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Procesar la consulta con nuestro prompt personalizado
print(f"Procesando consulta con el agente: {query}")
# Prompt mejorado que asegura que el agente use el DataFrame real
prompt = f"""
Tienes acceso a un DataFrame de pandas llamado 'df' con {len(df)} filas y las siguientes columnas: {df.columns.tolist()}.
Aquí están las primeras filas de los datos:
{df.head().to_string()}
Tipos de datos:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Pregunta: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Respuesta del agente: {response}")
return response['output']
except Exception as e:
print(f"Error procesando archivo Excel: {str(e)}")
import traceback
print(f"Traceback completo: {traceback.format_exc()}")
return f"Error procesando el archivo Excel: {str(e)}"
# Para archivos de texto no estructurados
print("\\
Procesando archivos de texto no estructurados...")
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 leyendo archivo {file_path}: {str(e)}")
continue
if not all_content:
return "No se pudo leer ningún documento."
# Procesar texto no estructurado usando almacén vectorial
print("Procesando texto con almacén vectorial...")
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"Longitud del contexto generado: {len(context)}")
response = self.llm.invoke(
f"""Basado en el siguiente contexto, responde la pregunta: {query}
\\
Contexto:\\
{context}\\
\\
Respuesta:"""
)
return response.content
except Exception as e:
print(f"Error procesando consulta de documentos: {str(e)}")
import traceback
print(f"Traceback completo: {traceback.format_exc()}")
return f"Error procesando la consulta de documentos: {str(e)}"
Integración de base de datos SQL
Ahora que hemos manejado el procesamiento de documentos, necesitamos crear el sistema de enrutamiento inteligente de consultas que decide entre SQL y análisis de documentos. Esta es la inteligencia central de nuestro sistema: determinar qué fuente de datos tiene más probabilidades de contener la respuesta a una pregunta determinada.
def process_query(self, query: str) -> str:
"""
Procesar consulta usando agentes para decidir inteligentemente entre SQL y documentos
"""
# Primero intentar con el agente SQL
try:
print("\\
Intentando con el agente SQL...")
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"De la base de datos SQL: {sql_result}"
else:
print("El agente SQL no pudo responder, intentando con documentos...")
except Exception as e:
print(f"Error del agente SQL: {str(e)}")
print("Recurriendo a documentos...")
# Si el agente SQL falla o no devuelve resultados, intentar procesamiento de documentos
try:
print("\\
Procesando documentos...")
doc_result = self._process_document_query(query)
if doc_result:
return f"De los documentos: {doc_result}"
else:
print("El procesamiento de documentos no devolvió resultados")
except Exception as e:
print(f"Error en el procesamiento de documentos: {str(e)}")
return "No se pudo encontrar información relevante ni en la base de datos SQL ni en los documentos."
La lógica de enrutamiento de consultas sigue un sistema de prioridades: las bases de datos SQL se prueban primero porque normalmente contienen datos estructurados y cuantitativos que pueden responder preguntas comerciales de manera rápida y precisa. Si el agente SQL devuelve respuestas poco claras o negativas (detectadas a través de nuestra lista no_answer_phrases), el sistema recurre automáticamente al procesamiento de documentos. Este enfoque imita cómo trabajaría un analista humano: primero verificar fuentes de datos estructuradas, luego recurrir a documentos e informes cuando la base de datos no tiene la información necesaria.
Creación de la interfaz de Streamlit
Ahora construyamos la interfaz de usuario que hace que nuestro sistema de análisis sea accesible para los usuarios empresariales. La interfaz de Streamlit ofrece una experiencia intuitiva basada en chat que oculta la complejidad técnica mientras ofrece potentes capacidades de análisis. Agrega los siguientes fragmentos de código al archivo main.py:
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# Cargar variables de entorno
load_dotenv()
#st.text_input
# Inicializar estado de sesión
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# Configurar página
st.set_page_config(
page_title="Chatbot de Análisis de Documentos",
page_icon="🤖",
layout="wide"
)
# CSS personalizado para botón y título
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)
# Título y descripción
st.title("Chatbot de Análisis de Documentos usando Novita")
Esta configuración inicial crea una interfaz de aspecto profesional con estilo personalizado. La gestión del estado de sesión asegura que los usuarios no pierdan su historial de conversaciones ni necesiten reinicializar sus fuentes de datos al interactuar con la interfaz. El CSS personalizado proporciona retroalimentación visual y mantiene una experiencia de marca consistente.
Integración final del sistema
Completemos la interfaz de Streamlit con la inicialización de la fuente de datos y la funcionalidad de chat. Esta sección final consolida todos nuestros componentes en una aplicación fácil de usar que los analistas de negocios pueden utilizar sin necesidad de experiencia técnica.
# Verificar si la fuente de datos está inicializada
if st.session_state.processor is None:
# Centrar la configuración de la fuente de datos
st.markdown("<br><br>", unsafe_allow_html=True)
# Crear columnas centradas
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 Comenzar")
st.write("Inicializa tu fuente de datos para empezar a chatear con tus documentos")
# Configuración de la base de datos SQL (oculta)
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# Configuración de la carpeta de documentos
st.write("**Ruta de la carpeta de documentos:**")
documents_folder = st.text_input(
"Ruta de la carpeta de documentos",
placeholder="Introduce la ruta a tu carpeta de documentos (p.ej., docs)",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# Centrar el botón
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("Inicializar fuente de datos", use_container_width=True):
try:
# Validar y resolver la ruta de la carpeta de documentos
if not documents_folder:
st.error("Por favor, proporciona una ruta de carpeta de documentos")
else:
# Convertir a ruta absoluta
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"Carpeta de documentos no encontrada: {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"La ruta no es un directorio: {abs_documents_folder}")
else:
# Inicializar el procesador de consultas con un motor SQL ficticio
# Crear motor SQL
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("¡Fuente de datos inicializada correctamente!")
st.rerun()
except Exception as e:
st.error(f"Error al inicializar la fuente de datos: {str(e)}")
else:
# Mostrar estado de la fuente de datos en la barra lateral
with st.sidebar:
st.header("📊 Fuente de datos")
st.success("✅ Fuente de datos inicializada")
if st.button("Restablecer fuente de datos"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# Interfaz principal de chat
st.header("Interfaz de chat")
# Mostrar mensajes del chat
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# Entrada de chat
if prompt := st.chat_input("Haz una pregunta sobre tus documentos"):
# Agregar mensaje del usuario al historial de chat
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# Procesar consulta
try:
response = st.session_state.processor.process_query(prompt)
# Agregar respuesta del asistente al historial de chat
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 al procesar la consulta: {str(e)}")
# Agregar un botón para borrar el chat solo si hay mensajes
if st.session_state.messages:
if st.button("Borrar chat"):
st.session_state.messages = []
st.rerun()
El proceso de inicialización incluye un manejo robusto de errores y validación de rutas para prevenir errores comunes de los usuarios. La barra lateral proporciona información clara sobre el estado, y la interfaz de chat sigue patrones familiares que los usuarios esperan de los asistentes de IA modernos. El sistema mantiene el historial de la conversación, lo que permite a los usuarios construir sobre preguntas anteriores y crear un flujo de trabajo analítico natural. La funcionalidad de borrar chat permite a los usuarios empezar de nuevo cuando sea necesario.
Ejecución del sistema de análisis
Ahora que hemos construido el sistema de análisis de ventas basado en RAG, pongámoslo a prueba y veamos qué tan bien se desempeña en consultas empresariales del mundo real.
Primero, crea un archivo data_generator.py y copia el código del script de Python aquí para generar datos de muestra que proporcionaremos al sistema, y ejecuta el script con el comando:
| python data_generator.py |
Esto creará una carpeta sample_documents con:
- large_sales_dataset.csv - 10,000 registros de ventas
- business_strategy_2024.txt - Documento estratégico de negocio
- sales_meeting_notes.txt - Notas de reuniones y elementos de acción
Luego, inicia la aplicación Streamlit:
| streamlit run main.py |
La aplicación se abrirá en tu navegador en http://localhost:8501.
Una vez inicializada, puedes probar el sistema ingresando la ruta de tus documentos como sample_documents y hacer varias preguntas comerciales:
| # Consultas de muestra para probar el sistema topic_1 = “¿Cuál fue el monto total de ventas de electrónicos en 2024?” topic_2 = “¿Cuáles son nuestras iniciativas estratégicas clave para la experiencia del cliente?” topic_3 = “¿Qué representante de ventas tuvo las ventas totales más altas y cuál fue el monto?” |
Conclusión
A lo largo de este tutorial, hemos construido un sistema de análisis basado en RAG que demuestra cómo la IA moderna puede transformar los flujos de trabajo de inteligencia empresarial. Al combinar múltiples enfoques de procesamiento de datos bajo una única interfaz inteligente, hemos creado una herramienta que puede manejar todo el espectro de preguntas comerciales que los analistas enfrentan a diario. Los puntos fuertes clave del sistema incluyen:
- Enrutamiento inteligente de consultas que determina automáticamente la mejor fuente de datos para cada pregunta.
- Soporte multi-formato para manejar bases de datos SQL, archivos CSV, hojas de cálculo de Excel y documentos de texto.
- Manejo robusto de errores que proporciona mecanismos de respaldo y mensajes de error claros.
- Interacción basada en chat que no requiere experiencia técnica.
- Verificación y validación integradas para garantizar resultados fiables.
Puedes extender esta aplicación para incorporar fuentes adicionales, técnicas de análisis más avanzadas y lógica empresarial personalizada. Ahora que tienes este conocimiento, adelante, intenta construir tus propios agentes con Novita AI o intégralo en tus proyectos existentes.
Novita AI es la plataforma integral en la nube que impulsa tus ambiciones de IA. API integradas, sin servidor, instancias de GPU: las herramientas rentables que necesitas. Elimina la infraestructura, comienza gratis y haz realidad tu visión de IA.
