Inteligencia Artificial

Seed Data Relacional Con LLMs

Echo Code

Construye scripts para generar miles de registros de prueba interrelacionales en SQL o JSON usando LLMs, garantizando integridad referencial.

Simplemente semillas.

AD PLACEHOLDER

Generación de Datos en la Actualidad#

Poblar bases de datos de desarrollo con información realista es un problema histórico en la ingeniería de software. Durante años hemos dependido de herramientas como Faker o FactoryBot. Estas librerías cumplen su función para pruebas unitarias aisladas, pero fracasan rotundamente cuando intentamos simular entornos complejos: el texto generado carece de cohesión semántica (el clásico “Lorem Ipsum”), las relaciones entre tablas no tienen sentido de negocio, y replicar sesgos estadísticos o distribuciones del mundo real requiere configuraciones manuales inmanejables.

La alternativa tradicional —anonimizar dumps de bases de datos de producción— es un campo minado de riesgos de seguridad y cumplimiento normativo (GDPR, HIPAA).

Los Modelos de Lenguaje Grande (LLMs) han cambiado las reglas del juego. Sin embargo, no puedes simplemente abrir una interfaz web y pedirle a un LLM “genérame 10,000 usuarios con sus posts y comentarios en SQL”. Se quedará sin contexto, alucinará claves foráneas que no existen, o violará la estructura de tu esquema.

Como ingenieros, necesitamos sistemas deterministas. En este artículo, desarrollaremos una arquitectura basada en scripts para orquestar a un LLM. El objetivo: generar miles de registros de prueba interrelacionales, asegurando una estricta integridad referencial y serializando el resultado en SQL o JSON listo para consumir.

La Arquitectura Top-Down para Generación de Datos#

La regla de oro para generar datos relacionales con LLMs es el enfoque Top-Down en lotes. El LLM no debe ser el responsable de gestionar el estado global ni de recordar qué IDs ha generado. Ese es el trabajo de tu script. El LLM debe actuar únicamente como un motor de traducción estructurada: tú le pasas un contexto (ej. un arreglo de IDs válidos) y él te devuelve objetos que cumplen con un esquema.

El flujo es el siguiente:

  1. Entidades Independientes (Nivel 1): Generas los registros base (ej. Users). El script asigna los Primary Keys (UUIDs o enteros autoincrementales) y guarda las referencias en memoria.
  2. Entidades Dependientes (Nivel 2): Divides los IDs de los usuarios generados en lotes. Por cada lote, inyectas esos IDs en el prompt del LLM y le instruyes generar registros hijos (ej. Posts) vinculados estrictamente a la lista proporcionada.
  3. Entidades Altamente Dependientes (Nivel 3): Repites el proceso. Inyectas combinaciones de user_id y post_id para que el LLM genere Comments, garantizando que el usuario y el post existen, y que el contexto del comentario tenga sentido respecto al título del post.

Definiendo el Esquema Objetivo#

Para nuestro caso de estudio, poblaremos el siguiente esquema relacional estándar de una plataforma de blogs. Utilizaremos PostgreSQL, aunque la lógica es agnóstica a la base de datos.

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    bio TEXT,
    created_at TIMESTAMP NOT NULL
);


CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    user_id INT NOT NULL REFERENCES users(id),
    title VARCHAR(200) NOT NULL,
    content TEXT NOT NULL,
    status VARCHAR(20) CHECK (status IN ('draft', 'published', 'archived')),
    created_at TIMESTAMP NOT NULL
);


CREATE TABLE comments (
    id SERIAL PRIMARY KEY,
    post_id INT NOT NULL REFERENCES posts(id),
    user_id INT NOT NULL REFERENCES users(id),
    body TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL
);

Estructurando las Salidas con Pydantic#

El mayor riesgo al usar LLMs es obtener texto libre en lugar de datos estructurados. Para mitigar esto, utilizamos el modo de “Structured Outputs” (Salidas Estructuradas) soportado por modelos modernos (como GPT-4o, Claude 3.5 Sonnet o Gemini 1.5 Pro). En Python, la mejor forma de definir este contrato es usando Pydantic.

from pydantic import BaseModel, Field
from typing import List
from datetime import datetime


class User(BaseModel):
    username: str = Field(description="Nombre de usuario único, estilo gamer o profesional.")
    bio: str = Field(description="Biografía corta, realista, acorde al username.")
    created_at: datetime


class Post(BaseModel):
    user_id: int = Field(description="Debe ser uno de los IDs proporcionados en el prompt.")
    title: str = Field(description="Título clickbait o técnico para un artículo de blog.")
    content: str = Field(description="Cuerpo del post, al menos 3 párrafos coherentes con el título.")
    status: str = Field(description="Solo puede ser: draft, published, o archived.")
    created_at: datetime


class Comment(BaseModel):
    post_id: int = Field(description="ID del post en el que se comenta.")
    user_id: int = Field(description="ID del usuario que realiza el comentario.")
    body: str = Field(description="Contenido del comentario. Puede ser positivo, crítico o troll, pero relevante al post.")
    created_at: datetime


class UserList(BaseModel):
    users: List[User]


class PostList(BaseModel):
    posts: List[Post]


class CommentList(BaseModel):
    comments: List[Comment]

Al obligar al modelo a adherirse a estas estructuras, delegamos la validación a nivel de aplicación. Si el LLM intenta introducir una fecha con un formato erróneo o un status que no coincide con nuestro CHECK de SQL, Pydantic lanzará una excepción que nuestro script puede atrapar para reintentar la llamada.

Ingeniería de Prompts para Integridad Referencial y Temporal#

Lograr coherencia requiere inyectar lógica de negocio en el System Prompt. Dos de los problemas más complejos en el seeding relacional son:

  1. Coherencia Temporal: Un usuario no puede publicar un post antes de registrarse. Un comentario no puede existir antes de que el post sea publicado.
  2. Coherencia Semántica: Si un post trata sobre “Kubernetes en Producción”, el comentario debe debatir sobre orquestación de contenedores, no sobre recetas de cocina.

Para resolver esto, el script de Python debe calcular los límites y pasarlos dinámicamente.

import instructor
from openai import AsyncOpenAI
from datetime import datetime, timedelta
import random


# Inicializamos el cliente. 'instructor' parchea el cliente para forzar Pydantic
client = instructor.patch(AsyncOpenAI())


async def generate_posts_batch(user_ids: List[int], start_date: str, end_date: str, count: int) -> PostList:
    system_prompt = f"""
    Eres un motor avanzado de generación de datos semilla (Seed Data) para una base de datos de pruebas.
    Tu tarea es generar exactamente {count} registros para la tabla 'posts'.


    REGLAS ESTRICTAS DE INTEGRIDAD:
    1. 'user_id' DEBE ser seleccionado aleatoriamente ÚNICAMENTE de esta lista: {user_ids}.
       Bajo ninguna circunstancia inventes un 'user_id' que no esté en ese arreglo.
    2. 'created_at' DEBE ser una fecha y hora aleatoria estrictamente entre {start_date} y {end_date}.
    3. Asegura una distribución realista de estados ('status'): 80% published, 15% draft, 5% archived.
    4. El 'title' y 'content' deben ser altamente realistas, orientados a la ingeniería de software y tecnología.
    """


    response = await client.chat.completions.create(
        model="gpt-4o",
        response_model=PostList,
        messages=[{"role": "system", "content": system_prompt}],
        temperature=0.7,
    )
    return response

Orquestación y Concurrencia#

Generar miles de registros de forma secuencial tomaría horas. Necesitamos concurrencia y un manejo estricto de los límites de tasa (Rate Limits) impuestos por la API del LLM. Utilizaremos asyncio con semáforos para evitar abrumar al proveedor.

El siguiente es el orquestador principal que maneja el flujo descendente (Top-Down):

import asyncio
import json


# Variables de estado global para controlar IDs generados
current_user_id = 1
current_post_id = 1
current_comment_id = 1


global_users = []
global_posts = []


async def worker_users(batch_size: int, semaphore: asyncio.Semaphore):
    global current_user_id
    async with semaphore:
        # Petición al LLM para generar batch_size usuarios
        # (Asume la existencia de la función generate_users_batch similar a posts)
        response = await generate_users_batch(batch_size)


        batch_records = []
        for u in response.users:
            record = u.model_dump()
            record['id'] = current_user_id
            current_user_id += 1
            batch_records.append(record)


        global_users.extend(batch_records)
        return batch_records


async def orchestrate_data_generation(total_users: int, posts_per_user_avg: int):
    print("Iniciando pipeline de Seed Data...")
    concurrency_limit = 5
    semaphore = asyncio.Semaphore(concurrency_limit)


    # 1. Generar Usuarios en lotes de 20
    user_batch_size = 20
    user_tasks = [
        worker_users(user_batch_size, semaphore)
        for _ in range(total_users // user_batch_size)
    ]
    await asyncio.gather(*user_tasks)
    print(f"✅ Generados {len(global_users)} usuarios.")


    # 2. Generar Posts basados en los Usuarios generados
    post_tasks = []
    # Dividimos los usuarios en sub-grupos para no enviar un array gigante al prompt
    for i in range(0, len(global_users), 50):
        subset = global_users[i:i+50]
        subset_ids = [u['id'] for u in subset]
        # Promedio de posts por grupo
        num_posts = len(subset) * posts_per_user_avg


        post_tasks.append(
            generate_and_map_posts(subset_ids, num_posts, semaphore)
        )


    await asyncio.gather(*post_tasks)
    print(f"✅ Generados {len(global_posts)} posts con integridad relacional.")


    # 3. (Mismo patrón para los comentarios iterando sobre global_posts y global_users)
    # ...

Volcado a SQL para Ambientes de Desarrollo#

Guardar los objetos en memoria está bien, pero el propósito de este pipeline es insertar los datos en una base de datos local o en una canalización CI/CD. Cargar un millón de objetos JSON en memoria usando Python causará un MemoryError. La estrategia correcta es escribir los registros en disco (en modo “append”) tan pronto como cada tarea asíncrona termine.

Podemos crear una función utilitaria que convierta los diccionarios mapeados a sentencias INSERT puras de SQL.

def append_to_sql_file(filename: str, table: str, data: list[dict]):
    if not data: return


    columns = data[0].keys()
    columns_str = ", ".join(columns)


    with open(filename, "a", encoding="utf-8") as f:
        f.write(f"INSERT INTO {table} ({columns_str}) VALUES\n")


        values_lines = []
        for row in data:
            formatted_values = []
            for col in columns:
                val = row[col]
                if isinstance(val, str):
                    # Escapar comillas simples en SQL
                    safe_val = val.replace("'", "''")
                    formatted_values.append(f"'{safe_val}'")
                elif isinstance(val, datetime):
                    formatted_values.append(f"'{val.strftime('%Y-%m-%d %H:%M:%S')}'")
                elif val is None:
                    formatted_values.append("NULL")
                else:
                    formatted_values.append(str(val))


            values_lines.append(f"  ({', '.join(formatted_values)})")


        f.write(",\n".join(values_lines) + ";\n\n")

Al inyectar la llamada a append_to_sql_file dentro del worker asíncrono, tu script generará un archivo seed_data.sql masivo, que puede alcanzar fácilmente gigabytes sin consumir más de unos cuantos megabytes de memoria RAM. Posteriormente, puedes cargar este script en PostgreSQL usando la herramienta nativa: psql -U admin -d my_dev_db -f seed_data.sql.

Gestión de Fallos y Retries#

Cuando se trabaja con LLMs a escala, los errores no son una posibilidad, son una garantía. A veces el modelo truncará el JSON porque excede el max_tokens. Otras veces la API responderá con un HTTP 502 (Bad Gateway). Tu orquestador necesita ser tolerante a fallos.

Para pipelines de datos de este calibre, la librería tenacity de Python es obligatoria. Te permite envolver tus llamadas al LLM con estrategias de backoff exponencial.

from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
from openai import RateLimitError, APIError


@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5),
    retry=retry_if_exception_type((RateLimitError, APIError))
)
async def resilient_llm_call(prompt_data):
    # Lógica de la llamada
    pass

Esto garantiza que tu proceso de generación de 100,000 registros no aborte en el registro 99,000 debido a un fallo temporal de red.

Consideraciones Económicas y de Escalabilidad#

Generar datos con LLMs es sustancialmente más costoso que usar un paquete como Faker. Generar miles de párrafos de contenido de blogs y comentarios con un modelo “flagship” impactará en la factura de la API.

Para pragmatismo empresarial:

  1. Usa modelos rápidos (y baratos) para entidades simples: Tablas que solo contienen estados, nombres o configuraciones pueden ser generadas con modelos tier-2 (como GPT-4o-mini o Gemini Flash) por fracciones de centavo.
  2. Reserva la inteligencia para la semántica: Usa modelos grandes exclusivamente para las tablas de contenido pesado, como Posts o Reviews, donde el razonamiento lógico es indispensable para asegurar que el contenido coincide con los parámetros de la entidad padre.
  3. Híbrida tu sistema: Puedes combinar bibliotecas tradicionales con LLMs. Si necesitas 50,000 usuarios base, genéralos con Faker localmente (nombres, correos). Luego pasa esos perfiles generados por Faker a un LLM y pídele que escriba la actividad relacional compleja.

Conclusión#

La generación de Seed Data ha evolucionado. Dejar atrás las “fábricas” de datos rígidos o el riesgo legal de clonar bases de datos de producción es ahora factible gracias a la orquestación de LLMs. Al combinar el esquema forzado mediante Pydantic, la gestión de relaciones top-down en el script de Python, y la generación asíncrona serializada directamente a .sql, obtenemos lo mejor de ambos mundos. Construimos un gemelo digital realista y dinámico para nuestro entorno de desarrollo, asegurando la robustez de nuestras pruebas y manteniendo la integridad referencial inviolable. La ingeniería de software moderna ya no escribe datos falsos a mano; programa algoritmos para que la IA los sintetice bajo estricto control técnico.