Agentes de IA sobre Big Data en Google Cloud: Cómo Dejar Atrás el BI Reactivo

La Analogía del Destornillador Eléctrico: Tres Niveles de Madurez Analítica

Un fontanero que ensambla tuberías con un destornillador manual puede tardar una jornada entera en completar una instalación simple; quien utiliza herramientas eléctricas multiplica por diez su producción en el mismo intervalo temporal. En analítica empresarial sucede con exactitud lo mismo:

Arquitectura de Integración: Function Calling con BigQuery y Vertex AI

A continuación se detalla una implementación en Python para un DataAgent empresarial. El modelo interpreta el lenguaje natural del usuario (e.g., producto con stock crítico o márgenes por canal), selecciona la herramienta de BigQuery adecuada, ejecuta la consulta bajo restricciones de particionamiento y devuelve el estado consolidado.

import os
from google.cloud import bigquery
from vertexai.generative_models import (
    GenerativeModel,
    Tool,
    FunctionDeclaration,
    Part
)

# 1. Definición del cliente y esquema de herramienta de BigQuery
bq_client = bigquery.Client(project=os.getenv("GCP_PROJECT_ID"))

get_stock_deficit_func = FunctionDeclaration(
    name="get_stock_deficit",
    description="Obtiene los productos cuyo inventario está por debajo del umbral de seguridad para una semana específica.",
    parameters={
        "type": "object",
        "properties": {
            "days_threshold": {
                "type": "integer",
                "description": "Número de días de cobertura restante antes de rotura de stock."
            },
            "category": {
                "type": "string",
                "description": "Categoría opcional del producto (e.g., Ropa, Calzado)."
            }
        },
        "required": ["days_threshold"]
    }
)

# 2. Implementación de la función determinista en BigQuery
def execute_stock_query(days_threshold: int, category: str = None) -> list:
    query = """
        SELECT 
            sku,
            product_name,
            color,
            size,
            available_stock,
            predicted_daily_sales,
            ROUND(available_stock / NULLIF(predicted_daily_sales, 0), 1) AS days_to_stockout
        FROM `enterprise_dw.inventory_predictions`
        WHERE (available_stock / NULLIF(predicted_daily_sales, 0)) <= @threshold
    """
    query_params = [bigquery.ScalarQueryParameter("threshold", "INT64", days_threshold)]
    
    if category:
        query += " AND product_category = @category"
        query_params.append(bigquery.ScalarQueryParameter("category", "STRING", category))
        
    query += " ORDER BY days_to_stockout ASC LIMIT 50;"
    
    job_config = bigquery.QueryJobConfig(query_parameters=query_params)
    query_job = bq_client.query(query, job_config=job_config)
    return [dict(row) for row in query_job.result()]

# 3. Inicialización del modelo con Tool Calling habilitado
stock_tool = Tool(function_declarations=[get_stock_deficit_func])
agent_model = GenerativeModel(
    model_name="gemini-1.5-pro",
    tools=[stock_tool],
    system_instruction="""Eres un DataAgent experto en la cadena de suministro y finanzas de la empresa.
    Tu misión es responder preguntas sobre stock, márgenes de Q3 y CAC.
    Utiliza siempre las herramientas deterministas conectadas a BigQuery antes de formular conclusiones."""
)

# 4. Flujo de ejecución del Agente
def process_user_query(user_prompt: str):
    chat = agent_model.start_chat()
    response = chat.send_message(user_prompt)
    
    for candidate in response.candidates:
        for part in candidate.content.parts:
            if part.function_call:
                fn_name = part.function_call.name
                fn_args = {k: v for k, v in part.function_call.args.items()}
                
                if fn_name == "get_stock_deficit":
                    tool_result = execute_stock_query(
                        days_threshold=int(fn_args.get("days_threshold", 7)),
                        category=fn_args.get("category")
                    )
                    
                    # Retornar el resultado estructurado de BigQuery al LLM para la síntesis final
                    final_response = chat.send_message(
                        Part.from_function_response(
                            name=fn_name,
                            response={"content": tool_result}
                        )
                    )
                    return final_response.text
    return response.text

Patrones de Diseño y Orquestación Multi-Agente

Para conseguir una plataforma robusta donde cualquier Product Manager o directivo interactúe de forma conversacional con miles de referencias, se requiere desacoplar responsabilidades en diferentes agentes autónomos:

Framework de Adopción: De Datos Crudos a un Agente en Producción

1. Centralización en Google Cloud: Migrar y consolidar fuentes transaccionales (ERP, CRM, plataformas de e-commerce, Google Analytics 4) dentro de BigQuery mediante pipelines de ingesta gestionados.
2. Capa Semántica y Modelado de Negocio: Diseñar vistas analíticas y modelos de predicción (BigQuery ML) que calculen ventas diarias estimadas, stock crítico y márgenes por canal (Online vs Retail).
3. Definición de Herramientas y Tool Spec: Exponer las consultas analíticas mediante funciones tipadas en Vertex AI Agent Builder o extensiones del SDK de Gemini.
4. Evaluación y Guardrails: Configurar filtros de seguridad, tests de regresión sintéticos sobre queries SQL y validación de márgenes mínimos para asegurar que el agente siempre responda con datos verificables.

Preguntas Frecuentes (FAQ)