AM un martes (bueno, técnicamente miércoles, supongo), cuando mi teléfono vibró con esa familiar y temida notificación de PagerDuty.
Ni siquiera necesité abrir mi computadora portátil para saber que el script daily_ingest.py había fallado. De nuevo.
Sigue fallando porque nuestro proveedor de datos siempre cambia el formato de su archivo sin previo aviso. Quiero decir, podrían cambiar aleatoriamente de comas a barras verticales o incluso estropear las fechas de la noche a la mañana.
Por lo general, la solución real me lleva sólo unos treinta segundos: simplemente abro el script, cambio sep=’,’ por sep=’|’ y presiono ejecutar.
Sé que fue rápido, pero honestamente, el costo real no es el tiempo de codificación, sino la interrupción del sueño y lo difícil que es hacer que tu cerebro funcione a las 2 a.m.
Esta rutina me hizo pensar: si la solución es tan obvia que puedo descifrarla con solo echar un vistazo al texto sin formato, ¿por qué un modelo no podría hacerlo?
A menudo escuchamos exageraciones acerca de que la “IA agente” reemplazará a los ingenieros de software, lo cual, sinceramente, me parece un poco exagerado.
Pero entonces, ¿la idea de utilizar un LLM pequeño y rentable para actuar como un desarrollador junior de guardia que se encargue de las aburridas excepciones de los pandas?
Eso parecía un proyecto que valía la pena intentar.
Entonces, construí un canal de “autocuración”. Aunque no es mágico, me ha protegido exitosamente de al menos tres llamadas de atención nocturnas este mes.
Y personalmente, cualquier cosa (por poco que sea) que pueda mejorar mi salud del sueño es definitivamente una gran victoria para mí.
Aquí está el desglose de cómo lo hice para que puedas construirlo tú mismo.
La arquitectura: un ciclo de “intentar-curar-reintentar”
El concepto central de esto es relativamente simple. La mayoría de los canales de datos son frágiles porque suponen que el mundo es perfecto, y cuando los datos de entrada cambian aunque sea ligeramente, fallan.
En lugar de aceptar ese fallo, diseñé mi script para detectar la excepción, capturar la “evidencia de la escena del crimen”, que es básicamente el rastreo y las primeras líneas del archivo, y luego pasarlo a un LLM.
Bastante bonito, ¿verdad?
El LLM ahora actúa como una herramienta de diagnóstico, analizando la evidencia para devolver los parámetros correctos, que luego el script utiliza para reintentar automáticamente la operación.
Para hacer que este sistema sea robusto, me basé en tres herramientas específicas:
Pandas: Para la carga de datos real (obviamente). Pydantic: para garantizar que el LLM devuelva JSON estructurado en lugar de relleno conversacional. Tenacity: una biblioteca de Python que hace que la escritura de una lógica de reintento compleja sea increíblemente limpia.
Paso 1: Definir la “solución”
El principal desafío al utilizar modelos de lenguaje grandes para la generación de código es su tendencia a alucinar. Según mi experiencia, si solicita un parámetro simple, a menudo recibirá a cambio un párrafo de texto conversacional.
Para detener eso, aproveché los resultados estructurados a través de Pydantic y la API de OpenAI.
Esto obliga al modelo a completar una forma estricta, actuando como un filtro entre el razonamiento desordenado de la IA y nuestro código Python limpio.
Este es el esquema que elegí, centrándome estrictamente en los argumentos que más comúnmente causan que read_csv falle:
de pydantic import BaseModel, campo de escribir import Opcional, Literal # Necesitamos un esquema estricto para que el LLM no se limite a ladrarnos. # Solo incluyo los parámetros que realmente causan fallas. clase CsvParams(BaseModel): sep: str = Field(description=”El delimitador, por ejemplo ‘,’ o ‘|’ o ‘;'”) codificación: str = Field(default=”utf-8″, descripción=”Codificación de archivo”) encabezado: Opcional[int | str] = Field(default=”infer”, descripción=”Fila para nombres de columnas”) # A veces el motor C se ahoga con los separadores de expresiones regulares, por lo que dejamos que la IA cambie de motor: Literal[“python”, “c”] = “pitón”
Al definir este modelo base, efectivamente le estamos diciendo al LLM: “No quiero una conversación ni una explicación. Quiero que se completen estas cuatro variables y nada más”.
Paso 2: La función del sanador
Esta función es el corazón del sistema, diseñada para ejecutarse sólo cuando algo ya salió mal.
Obtener el mensaje correcto requirió algo de prueba y error. Y eso se debe a que inicialmente solo proporcioné el mensaje de error, lo que obligó al modelo a adivinar el problema a ciegas.
Rápidamente me di cuenta de que para identificar correctamente problemas como discrepancias en los delimitadores, el modelo necesitaba “ver” realmente una muestra de los datos sin procesar.
Ahora aquí está el gran problema. En realidad, no se puede leer el archivo completo.
Si intenta pasar un CSV de 2 GB al mensaje, explotará su ventana de contexto y aparentemente su billetera.
Afortunadamente, descubrí que simplemente tirar de las primeras líneas le da al modelo la información suficiente para solucionar el problema el 99% de las veces.
import openai import json client = openai.OpenAI() def Ask_the_doctor(fp, error_trace): “”” El ‘Agente de guardia’. Examina el fragmento de archivo y el error, y sugiere nuevos parámetros. “”” print(f”🔥 Fallo detectado en {fp}. Llamando a LLM…”) # Hack: simplemente tome las primeras 4 líneas. No es necesario leer 1 GB. # Usamos errores = ‘reemplazar’ para no fallar al intentar solucionar un problema. intente: con open(fp, “r”, errores=”replace”) como f: head = “”.join([f.readline() for _ in range(4)]) excepto Excepción: head = “<>” # Mantenga el mensaje simple. No hay necesidad de una inyección compleja de “persona”. Prompt = f””” Estoy intentando leer un CSV con pandas y falló. Seguimiento de error: {error_trace} Fragmento de datos (primeras 4 líneas): — {head} — Devuelve los parámetros JSON correctos (sep, codificación, encabezado, motor) para solucionar este problema. “”” # Forzamos al modelo a usar nuestra finalización de esquema Pydantic = client.chat.completions.create( model=”gpt-4o”, # gpt-4o-mini también está bien aquí y mensajes más baratos =[{“role”: “user”, “content”: prompt}]funciones =[{
“name”: “propose_fix”,
“description”: “Extracts valid pandas parameters”,
“parameters”: CsvParams.model_json_schema()
}]function_call={“name”: “propose_fix”} ) # Analizar el resultado en un dict args = json.loads(completion.choices[0].message.function_call.arguments) print(f”💊 Corrección prescrita: {args}”) return args
Estoy pasando por alto la configuración de la API aquí, pero ya entiendes la idea. Toma los “síntomas” y prescribe una “pastilla” (los argumentos).
Paso 3: El ciclo de reintento (donde sucede la magia)
Ahora necesitamos conectar esta herramienta de diagnóstico a nuestro cargador de datos real.
En el pasado, escribí bucles while True feos con bloques try/except anidados que eran una pesadilla de leer.
Luego encontré la tenacidad, que te permite decorar una función con una lógica de reintento limpia.
Y la mejor parte es que la tenacidad también te permite definir una “devolución de llamada” personalizada que se ejecuta entre intentos.
Aquí es exactamente donde inyectamos nuestra función Sanadora.
importar pandas como pd desde el reintento de importación de tenacidad, stop_after_attempt, retry_if_exception_type # Un dictado global sucio para almacenar la “solución” entre reintentos. # En una clase real, esto sería self.state, pero para un script, esto funciona. fix_state = {} def apply_fix(retry_state): # Esto se ejecuta justo después del fallo, antes del siguiente intento e = retry_state.outcome.exception() fp = retry_state.args[0]
# Solicite al LLM una nueva sugerencia de parámetros = Ask_the_doctor(fp, str(e)) # Actualice el estado para que la próxima ejecución utilice la sugerencia fix_state[fp] = sugerencia @retry( stop=stop_after_attempt(3), # Dale 3 strikes retry_if_exception_type(Exception), # Captura todo (arriesgado, pero divertido) before_sleep=apply_fix # <--- Este es el gancho ) def hard_loader(fp): # Comprueba si tenemos una solución sugerida para este archivo; de lo contrario, utiliza coma de forma predeterminada params = fix_state.get(fp, {"sep": ","}) print(f"🔄 Intentando cargar con: {params}") df = pd.read_csv(fp, **params) return df
¿Realmente funciona?
Para probar esto, creé un archivo roto intencionalmente llamado messy_data.csv. Lo hice delimitado por barras verticales (|) pero no se lo dije al script.
Cuando ejecuté hard_loader(‘messy_data.csv’), el script falló, se detuvo por un momento mientras “pensaba” y luego se solucionó automáticamente.
Es sorprendentemente satisfactorio ver cómo el código falla, se diagnostica a sí mismo y se recupera sin ninguna intervención humana.
Los “Gotchas” (porque nada es perfecto)
No quiero exagerar esta solución, ya que definitivamente implica riesgos.
El costo
Primero, recuerde que cada vez que su canalización se interrumpe, está realizando una llamada a la API.
Eso podría estar bien para algunos errores, pero si tiene un procesamiento de trabajo masivo, digamos alrededor de 100,000 archivos, y una mala implementación hace que todos se rompan a la vez, podría despertarse con una sorpresa muy desagradable en su factura de OpenAI.
Si está ejecutando esto a escala, le recomiendo implementar un disyuntor o cambiar a un modelo local como Llama-3 a través de Ollama para mantener bajos los costos.
Seguridad de los datos
Si bien solo envío las primeras cuatro líneas del archivo al LLM, debes tener mucho cuidado con lo que hay en esas líneas. Si sus datos contienen información de identificación personal (PII), efectivamente está enviando esos datos confidenciales a una API externa.
Si trabaja en una industria regulada como la atención médica o las finanzas, utilice un modelo local.
En serio.
No envíe datos del paciente a GPT-4 solo para corregir un error de coma.
El “niño que gritaba lobo”
Finalmente, hay ocasiones en las que los datos deberían fallar.
Si un archivo está vacío o corrupto, no querrás que la IA alucine con una forma de cargarlo de todos modos, lo que podría llenar tu DataFrame con basura.
Pydantic filtra los datos incorrectos, pero no es mágico. Debe tener cuidado de no ocultar errores reales que realmente necesita solucionar usted mismo.
Conclusión y conclusión
Se podría argumentar que usar una IA para corregir archivos CSV es excesivo y, técnicamente, podría tener razón.
Pero en un campo tan cambiante como la ciencia de datos, los mejores ingenieros no son los que se aferran a los métodos que aprendieron hace cinco años; ellos son los que experimentan constantemente con nuevas herramientas para resolver viejos problemas.
Honestamente, este proyecto fue solo un recordatorio para mantenernos flexibles.
No podemos simplemente seguir protegiendo nuestros viejos oleoductos; Tenemos que seguir encontrando formas de mejorarlos. En esta industria, la habilidad más valiosa no es escribir código más rápido; más bien, es tener la curiosidad de probar una forma de trabajar completamente nueva.