El Agente de Marketing que Aprendió de 10.000 Campañas: Arquitectura Multi-Agente sobre BigQuery y Vertex AI

Matriz de Decisión: Enfoques para Analítica Conversacional sobre Datos

Elegir el patrón adecuado para conectar Modelos de Lenguaje Masivo (LLM) con almacenes de datos empresariales determina el coste por consulta, la latencia y la tasa de error sintáctico o analítico:

Implementación Técnica: Orquestador Multi-Agente en Python

A continuación se muestra el código modular que define los sub-agentes (Data Agent conectado a BigQuery mediante Conversational Analytics API y Visualization Agent con extensión Code Interpreter), orquestados por un Root Agent en ADK.

import os
from typing import Dict, Any
from google.cloud import bigquery
from google.cloud import conversational_analytics_v1alpha as ca_api
from vertexai.preview.extensions import Extension

# Configuración de variables de entorno y metadatos de GCP
PROJECT_ID = os.getenv("GCP_PROJECT_ID", "data-agent-prod")
LOCATION = os.getenv("GCP_LOCATION", "europe-southwest1")
DATASET_ID = os.getenv("BQ_DATASET", "marketing_finance_bi")
CODE_INTERPRETER_EXT_ID = os.getenv("VERTEX_A2_CODE_INTERPRETER_EXTENSION", "6419358487739469812")

class MarketingAnalyticsAgentSystem:
    """
    Sistema multi-agente desacoplado para auditoría de campañas publicitarias
    y proyecciones financieras en BigQuery.
    """
    def __init__(self, project_id: str, location: str, dataset_id: str):
        self.project_id = project_id
        self.location = location
        self.dataset_id = dataset_id
        self.bq_client = bigquery.Client(project=project_id)
        
        # Inicialización de la extensión de Code Interpreter en Vertex AI
        self.code_interpreter = Extension(
            extension_name=f"projects/{self.project_id}/locations/{self.location}/extensions/{CODE_INTERPRETER_EXT_ID}"
        )

    def get_bigquery_data_tool(self, user_prompt: str) -> Dict[str, Any]:
        """
        Tool consumida por DataAgent: Llama a Conversational Analytics API
        sobre el grafo semántico de tablas en BigQuery.
        """
        tables = [
            "actividad_clientes",
            "atribucion_conversiones",
            "balance_inicial",
            "campanas_marketing_detalle",
            "cuentas_financieras",
            "presupuestos_mensuales",
            "mv_cashflow_proyectado_30_dias"
        ]
        
        # Referencias fuertemente tipadas a BigQuery Data Analytics Client
        data_references = [
            ca_api.BigQueryTableReference(
                project_id=self.project_id,
                dataset_id=self.dataset_id,
                table_id=table_name
            ) for table_name in tables
        ]
        
        # Petición estructurada al motor semántico
        request = ca_api.AnalyzeBusinessDataRequest(
            project=f"projects/{self.project_id}/locations/{self.location}",
            user_question=user_prompt,
            context_tables=data_references,
            llm_model="gemini-1.5-flash-002"
        )
        
        # Simulación de extracción determinista (retorna insights y datos tipados)
        return {
            "status": "success",
            "insights": "Google Search lidera ROI (2674.62) sobre YouTube Ads (1155.50) y Meta Ads (275.27).",
            "raw_dataset": [
                {"canal": "Google Search", "roi": 2674.62, "inversion": 5000.0, "ingresos": 138731.0},
                {"canal": "YouTube Ads", "roi": 1155.50, "inversion": 3200.0, "ingresos": 40176.0},
                {"canal": "Meta Ads", "roi": 275.27, "inversion": 8400.0, "ingresos": 31522.68}
            ]
        }

    def generate_visualization_tool(self, raw_data: list, chart_type: str = "bar") -> str:
        """
        Tool consumida por VisualizationAgent: Ejecuta código en sandbox Vertex AI Code Interpreter.
        """
        python_script = f"""
import pandas as pd
import matplotlib.pyplot as plt

data = {raw_data}
df = pd.DataFrame(data)
plt.figure(figsize=(10, 6))
plt.bar(df['canal'], df['roi'], color=['#36f097', '#2de389', '#1b8a54'])
plt.title('Comparativa de ROI Semanal por Canal')
plt.xlabel('Canal Publicitario')
plt.ylabel('Retorno de Inversión (ROI %)')
plt.savefig('/tmp/roi_chart.png')
"""
        execution_result = self.code_interpreter.execute(
            code=python_script,
            language="python"
        )
        return execution_result.get("output_file_uri", "gs://analytics_bucket/roi_chart.png")

# Definición del flujo multi-agente en ADK (Root Agent Router)
def root_agent_router(user_message: str, agent_system: MarketingAnalyticsAgentSystem):
    """
    Router orquestador: delega a DataAgent para cómputo BigQuery y a VisualizationAgent si se solicita gráfico.
    """
    # 1. Recuperar contexto semántico de datos
    analytics_payload = agent_system.get_bigquery_data_tool(user_message)
    
    # 2. Verificar intención de visualización
    if "visualización" in user_message.lower() or "gráfico" in user_message.lower():
        chart_uri = agent_system.generate_visualization_tool(analytics_payload["raw_dataset"])
        return {
            "text_response": analytics_payload["insights"],
            "artifact_uri": chart_uri
        }
        
    return {"text_response": analytics_payload["insights"]}

if __name__ == "__main__":
    system = MarketingAnalyticsAgentSystem(PROJECT_ID, LOCATION, DATASET_ID)
    response = root_agent_router("¿Qué canal generó más ROI esta semana?", system)
    print(f"[Agent System Response]: {response['text_response']}")

Patrones de Diseño y Mejores Prácticas de Arquitectura

Al operar un agente de analítica sobre miles de campañas de marketing y flujos de tesorería en tiempo real, deben aplicarse tres principios arquitectónicos fundamentales:

1. Desacoplamiento de Agentes (Separation of Concerns)

El Root Agent actúa únicamente como punto de contacto con el usuario para determinar la intención. El Data Agent cuenta con acceso de solo lectura sobre BigQuery y nunca compila visualizaciones en el mismo thread. El Visualization Agent consume la salida estructurada del Data Agent y la ejecuta en un entorno aislado (Vertex AI Code Interpreter sandbox), garantizando seguridad y trazabilidad.

2. Esquemas Normalizados con Descripciones Semánticas

Las tablas de BigQuery (campanas_marketing_detalle, actividad_clientes, mv_cashflow_proyectado_30_dias) deben contar con metadatos descriptivos en cada columna (ej. "Identificador único para el registro diario de una campaña publicitaria"). Gemini utiliza este diccionario semántico para resolver cálculos complejos sin requerir reentrenamiento del modelo.

3. Vistas Materializadas para Proyecciones de Tesorería

Cálculos complejos como el Cashflow proyectado a 30 días o la atribución multi-touch deben precalcularse en Materialized Views o tablas de snapshot diario dentro de BigQuery. De esta forma, el agente consulta datos pre-agregados en milisegundos sin penalizar el rendimiento ni saturar los slots de cómputo.

Preguntas Frecuentes (FAQ)