En este artículo, aprenderá cinco conceptos esenciales de Python que todo ingeniero de IA debe dominar para crear sistemas de IA escalables y de nivel de producción.
Los temas que cubriremos incluyen:
Cómo los generadores y la evaluación diferida le permiten transmitir grandes conjuntos de datos con una sobrecarga de memoria constante. Cómo los administradores de contexto, la programación asincrónica y los modelos de Pydantic lo ayudan a administrar recursos de hardware, escalar llamadas API y validar configuraciones de forma segura. Cómo los métodos mágicos de Python le permiten crear abstracciones personalizadas que se integran limpiamente con marcos de aprendizaje profundo como PyTorch.
Conceptos de Python que todo ingeniero de IA debe dominar
Lo que los ingenieros de IA necesitan saber
La transición de escribir scripts experimentales locales a construir sistemas de inteligencia artificial escalables y de grado de producción requiere un cambio en la forma en que escribimos Python. Si bien la escritura dinámica, los bucles básicos y la comprensión de listas son razonables para crear prototipos de modelos o explorar datos, no cumplen con las limitaciones de rendimiento, memoria y latencia de las aplicaciones de IA del mundo real.
La ingeniería de IA no se trata solo de entrenar algoritmos o cargar pesos previamente entrenados: se trata de manejar enormes conjuntos de datos, administrar costosos recursos de hardware como GPU, conectarse a API externas simultáneamente y crear interfaces de software limpias y con seguridad de tipos. Para operar a este nivel, debe dominar las construcciones del lenguaje nativo en las que se basan los desarrolladores profesionales y los marcos de aprendizaje profundo.
En este artículo, exploraremos cinco conceptos críticos de Python que usted, el ingeniero de IA, debe dominar:
Generadores y evaluación diferida: para transmitir conjuntos de datos enormes con una sobrecarga de memoria constante. Administradores de contexto: para gestionar estados valiosos del hardware y limpieza de recursos. Programación asincrónica: para escalar consultas API de LLM y ejecución simultánea de herramientas de agentes. Clases de datos y Pydantic: para validar configuraciones y crear esquemas estructurados para la llamada de herramientas. Métodos mágicos: para diseñar abstracciones de aprendizaje automático compatibles con el marco desde cero.
1. Generadores y evaluación diferida (transmisión de datos con uso eficiente de la memoria)
Al entrenar modelos o ejecutar inferencia por lotes en conjuntos de datos a gran escala, cargar todos los datos en la memoria a la vez es una receta para errores de falta de memoria. Si su conjunto de datos contiene millones de documentos de texto, imágenes de alta resolución o vectores de características, una lista estándar obliga a Python a asignar memoria para todos los elementos a la vez.
Los generadores resuelven esto con una evaluación diferida. Al utilizar la palabra clave rendimiento, un generador devuelve un iterador que calcula y produce elementos según demanda, uno a la vez. Esto mantiene el uso de RAM estable, ya sea que esté transmitiendo 100 muestras o 100 millones.
En este enfoque ingenuo, leemos y preprocesamos un conjunto de datos de cargas de texto, cargando todos los diccionarios procesados en una única lista masiva en la memoria antes de que podamos iterar sobre ellos:
import json
import io
# A mock JSONL file stream of raw text payloads
def get_dataset_stream():
data = “n”.join([json.dumps({“id”: i, “text”: f”User query raw text payload {i}”}) for i in range(50000)])
return io.StringIO(data)
# Naive list function processing all records at once
def load_all_records_naive(stream):
records =[]para línea en flujo: payload = json.loads(line) # Procesar datos inmediatamente y agregarlos a una lista procesada = { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) } records.append(processed) return records # Ejecutar esto requiere cargar todos los 50,000 diccionarios procesados en RAM stream = get_dataset_stream() data = load_all_records_naive(stream) print(f"Registros cargados {len(data)} estilo ingenuo.")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
importar json
importar io
# Un flujo de archivo JSONL simulado de cargas útiles de texto sin formato
def get_dataset_stream ( ) :
datos = "n" . unirse ( [ json . dumps ( { "id" : i , "texto" : f "Carga útil de texto sin formato de consulta de usuario {i}" } ) para i en rango ( 50000 ) ] )
devolver yo . CadenaIO ( datos )
# Función de lista ingenua que procesa todos los registros a la vez
def load_all_records_naive ( corriente ) :
archivos = [ ]
para línea en arroyo :
carga útil = json . cargas ( línea )
# Procesar datos inmediatamente y agregarlos a una lista
procesado = {
"identificación" : carga útil [ "id" ] ,
"texto" : carga útil [ "texto" ] . más bajo ( ) ,
"longitud" : len ( carga útil [ "texto" ] )
}
registros . anexar ( procesado )
devolver archivos
# Ejecutar esto requiere cargar los 50.000 diccionarios procesados en la RAM
arroyo = get_dataset_stream ( )
datos = load_all_records_naive ( transmisión )
print ( f "Registros {len(data)} cargados en estilo ingenuo." )
Al convertir nuestro lector en un generador, transmitimos las cargas útiles preprocesadas lote por lote según demanda. Veamos un script que utiliza la biblioteca tracemalloc de Python para medir la diferencia en el uso máximo de memoria:
import json
import io
import tracemalloc
# A mock JSONL file stream of raw text payloads
def get_dataset_stream():
data = “n”.join([json.dumps({“id”: i, “text”: f”User query raw text payload {i}”}) for i in range(50000)])
return io.StringIO(data)
# Naive list function processing all records at once
def load_all_records_naive(stream):
records =[]para línea en flujo: payload = json.loads(line) # Procesar datos inmediatamente y agregarlos a una lista procesada = { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) } records.append(processed) return records # Función generadora que genera registros preprocesados uno por uno def stream_records_generator(stream): para line in stream: payload = json.loads(line) yield { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) } # Medir la implementación ingenua tracemalloc.start() stream_naive = get_dataset_stream() records_list = load_all_records_naive(stream_naive) para r en records_list: pass # Simular un paso de bucle de entrenamiento _, pico_naive = tracemalloc.get_traced_memory() tracemalloc.stop() # Medir la implementación del generador tracemalloc.start() stream_gen = get_dataset_stream() records_generator = stream_records_generator(stream_gen) for r en records_generator: pass # Simular un paso de bucle de entrenamiento _, pico_gen = tracemalloc.get_traced_memory() tracemalloc.stop() # Resultados de salida print(f"RAM pico ingenuo: {peak_naive / 1024 / 1024:.4f} MB") print(f"RAM máxima del generador: {peak_gen / 1024 / 1024:.4f} MB")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
importar json
importar io
importar tracemalloc
# Un flujo de archivo JSONL simulado de cargas útiles de texto sin formato
def get_dataset_stream ( ) :
datos = "n" . unirse ( [ json . dumps ( { "id" : i , "texto" : f "Carga útil de texto sin formato de consulta de usuario {i}" } ) para i en rango ( 50000 ) ] )
devolver yo . CadenaIO ( datos )
# Función de lista ingenua que procesa todos los registros a la vez
def load_all_records_naive ( corriente ) :
archivos = [ ]
para línea en arroyo :
carga útil = json . cargas ( línea )
# Procesar datos inmediatamente y agregarlos a una lista
procesado = {
"identificación" : carga útil [ "id" ] ,
"texto" : carga útil [ "texto" ] . más bajo ( ) ,
"longitud" : len ( carga útil [ "texto" ] )
}
registros . anexar ( procesado )
devolver archivos
# Función generadora que genera registros preprocesados uno por uno
def stream_records_generator ( corriente ) :
para línea en arroyo :
carga útil = json . cargas ( línea )
producir {
"identificación" : carga útil [ "id" ] ,
"texto" : carga útil [ "texto" ] . más bajo ( ) ,
"longitud" : len ( carga útil [ "texto" ] )
}
# Medir la implementación ingenua
trazamalloc . comenzar ( )
corriente_naive = get_dataset_stream ( )
lista_registros = load_all_records_naive ( corriente_naive )
para r en lista_registros :
aprobar # Simular un paso de bucle de entrenamiento
_ , pico_ingenuo = trazamalloc . get_traced_memory ( )
trazamalloc . detener ( )
# Medir la implementación del generador.
trazamalloc . comenzar ( )
flujo_gen = get_dataset_stream ( )
generador_de_registros = stream_records_generator ( stream_gen )
para r en generador_de_registros :
aprobar # Simular un paso de bucle de entrenamiento
_ , pico_gen = trazamalloc . get_traced_memory ( )
trazamalloc . detener ( )
# Resultados de salida
print ( f "RAM ingenua máxima: {peak_naive / 1024/1024:.4f} MB" )
print ( f "RAM máxima del generador: {peak_gen / 1024 / 1024:.4f} MB" )
Producción:
Naive peak RAM: 25.2114 MB
Generator peak RAM: 13.9610 MB
Naive peak RAM : 25.2114 MB
Generator peak RAM : 13.9610 MB
Al utilizar generadores, el consumo máximo de RAM se redujo a casi la mitad. Cuando se trabaja con conjuntos de datos de texto de varios gigabytes para modelos de lenguaje grandes o imágenes por lotes para modelos de visión, la transmisión de datos garantiza que el consumo de memoria se mantenga estable y predecible, evitando la preocupación de quedarse sin RAM en producción.
2. Administradores de contexto (estado de hardware y gestión de recursos)
¡No, ese contexto no!
Las aplicaciones de IA son grandes consumidoras de recursos físicos y estatales. Debe abrir y cerrar conexiones a bases de datos vectoriales, administrar cálculos de gradiente de PyTorch o perfilar dinámicamente bloques de latencia.
Si no limpia los recursos o si se produce una excepción antes de restaurar una configuración, corre el riesgo de perder memoria o mantener las variables de estado atascadas en la configuración incorrecta. Los administradores de contexto utilizan la instrucción with para encapsular los bloques de ejecución, asegurando que la lógica de configuración y desmontaje se ejecute limpiamente, incluso si se produce un error.
Aquí, intentamos establecer temporalmente un modelo simulado en modo de evaluación, rastrear su latencia de inferencia y borrar el caché de la GPU manualmente mediante un bloque try-finally. Este enfoque es muy repetitivo y se utiliza como ejemplo:
import time
class MockPyTorchModel:
def __init__(self):
self.training = True
def __call__(self, x):
return [val * 1.5 for val in x]
# Create model
model = MockPyTorchModel()
# Start manual setup and execution
start_time = time.perf_counter()
original_mode = model.training
# Manually set model to evaluation mode
model.training = False
try:
# Perform inference
outputs = model([1.0, 2.0, 3.0])
print(f”Inference outputs: {outputs}”)
finally:
# We must explicitly clean up and restore state
model.training = original_mode
elapsed = time.perf_counter() – start_time
print(f”[Manual Profile] Inference took {elapsed:.6f}s”)
print(“[Manual GPU] Simulating: torch.cuda.empty_cache()”)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import time
class MockPyTorchModel :
def __init__ ( self ) :
self . training = True
def __call__ ( self , x ) :
devolver [ valor * 1.5 para valor en x ]
# Crear modelo
modelo = Modelo simulado de PyTorch ( )
# Iniciar la configuración y ejecución manual
hora_inicio = tiempo . contador_perf ( )
modo_original = modelo . capacitación
# Establecer manualmente el modelo en modo de evaluación
modelo . capacitación = FALSO
intentar :
# Realizar inferencia
salidas = modelo ( [ 1.0 , 2.0 , 3.0 ] )
print ( f "Salidas de inferencia: {salidas}" )
finalmente :
# Debemos limpiar y restaurar explícitamente el estado
modelo . capacitación = modo_original
transcurrido = tiempo . contador_perf ( ) – hora_inicio
print ( f "[Perfil manual] La inferencia tomó {elapsed:.6f}s" )
print ( "[GPU manual] Simulando: torch.cuda.empty_cache()" )
Podemos encapsular este comportamiento en un administrador de contexto limpio y reutilizable utilizando los métodos __enter__ y __exit__ basados en clases estándar de Python:
import time
class MockPyTorchModel:
def __init__(self):
self.training = True
def __call__(self, x):
return [val * 1.5 for val in x]
class InferenceProfiler:
def __init__(self, model):
self.model = model
def __enter__(self):
self.start_time = time.perf_counter()
self.original_mode = self.model.training
# Set model to evaluation mode
self.model.training = False
print(“[Enter] Switched model to eval mode, started timer.”)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
# Restore the original training state
self.model.training = self.original_mode
elapsed = time.perf_counter() – self.start_time
print(f”[Exit] Block latency: {elapsed:.6f} seconds”)
print(“[Exit] Restored training state. Simulating CUDA cache clean.”)
# Returning False ensures any exception that occurred is not suppressed
return False
# Execution becomes incredibly clean and robust
model = MockPyTorchModel()
with InferenceProfiler(model):
res = model([1.0, 2.0, 3.0])
print(f”Prediction inside context: {res}”)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import time
class MockPyTorchModel :
def __init__ ( self ) :
self . training = True
def __call__ ( self , x ) :
return [ val * 1.5 for val in x ]
class InferenceProfiler :
def __init__ ( self , model ) :
self . model = model
def __enter__ ( self ) :
self . start_time = time . perf_counter ( )
self . original_mode = self . model . training
# Set model to evaluation mode
self . model . training = False
print ( “[Enter] Switched model to eval mode, started timer.” )
devolver ser
def __salir__ ( yo , tipo_exc , exc_val , exc_tb ) :
# Restaurar el estado de entrenamiento original
ser . modelo . capacitación = ser . modo_original
transcurrido = tiempo . contador_perf ( ) – ser . hora_inicio
print ( f "[Salir] Latencia de bloque: {transcurrido:.6f} segundos" )
print ( "[Salir] Estado de entrenamiento restaurado. Simulación de limpieza de caché CUDA." )
# Devolver False garantiza que cualquier excepción que haya ocurrido no se suprima
devolver FALSO
# La ejecución se vuelve increíblemente limpia y robusta
modelo = Modelo simulado de PyTorch ( )
con InferenceProfiler ( modelo ) :
res = modelo ( [ 1.0 , 2.0 , 3.0 ] )
print ( f "Predicción dentro del contexto: {res}" )
Producción:
[Enter] Switched model to eval mode, started timer.
Prediction inside context: [1.5, 3.0, 4.5]
[Exit] Block latency: 0.000045 seconds
[Exit] Restored training state. Simulating CUDA cache clean.
[ Enter ] Switched model to eval mode , started timer .
Prediction inside context : [ 1.5 , 3.0 , 4.5 ]
[ Exit ] Block latency : 0.000045 seconds
[ Exit ] Restored training state . Simulating CUDA cache clean .
Al definir InferenceProfiler, se abstrae la lógica de limpieza y manejo de errores. Ya sea que la inferencia tenga éxito o falle en pleno vuelo, el administrador de contexto garantiza que se restablezca el estado de entrenamiento original del modelo y que la telemetría de ejecución se capture de forma segura.
3. Programación asincrónica (ampliación de API de LLM y llamadas de herramientas de agentes)
Gracias a las aplicaciones basadas en LLM y a los flujos de trabajo agentes, la entrada/salida (E/S) de la red suele ser el principal cuello de botella de latencia. Si su agente necesita evaluar 50 mensajes de usuario utilizando una API en la nube o consultar un almacén de vectores remoto, el envío de estas solicitudes bloquea secuencialmente su programa en cada llamada de red.
La programación asincrónica con asyncio permite a Python manejar múltiples tareas al mismo tiempo. En lugar de esperar ociosamente una respuesta HTTP, Python pausa la tarea actual y ejecuta otras operaciones, acelerando los bucles multiagente y las ejecuciones de herramientas.
Aquí, iteramos a través de las indicaciones y realizamos una llamada de red síncrona estándar para cada una. El programa permanece completamente inactivo durante el tiempo de espera HTTP simulado:
import time
# Mocking a synchronous external API call to an LLM
def query_llm_sync(prompt: str) -> str:
time.sleep(0.1) # Simulate 100ms network latency
return f”Response to ‘{prompt}'”
def run_sequential(prompts):
start = time.perf_counter()
results =[]para p en indicaciones: results.append(query_llm_sync(p)) transcurrido = time.perf_counter() – iniciar impresión(f"El procesamiento secuencial tardó {elapsed:.4f} segundos.") devolver indicaciones de resultados = [f"Explicar el tema {i}" para i en el rango(20)] _ = run_sequential(indicaciones)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
tiempo de importación
# Burlarse de una llamada API externa sincrónica a un LLM
def query_llm_sync ( mensaje : cadena ) -> cadena :
tiempo . dormir ( 0.1 ) # Simular una latencia de red de 100 ms
devolver f "Respuesta a '{prompt}'"
def run_sequential ( indicaciones ) :
comenzar = tiempo . contador_perf ( )
resultados = [ ]
para pag en indicaciones :
resultados . agregar ( query_llm_sync ( p ) )
transcurrido = tiempo . contador_perf ( ) – comenzar
print ( f "El procesamiento secuencial tardó {elapsed:.4f} segundos." )
devolver resultados
indicaciones = [ f "Explicar el tema {i}" para i en rango ( 20 ) ]
_ = run_sequential ( indicaciones )
Producción:
Sequential processing took 2.0864 seconds.
Sequential processing took 2.0864 seconds .
Usando asyncio y await, podemos enviar las 20 tareas de red simultáneamente. Esto se asigna perfectamente a bibliotecas de producción como httpx y SDK asíncronos como AsyncOpenAI:
import asyncio
import time
# Mocking an asynchronous external API call to an LLM
async def query_llm_async(prompt: str) -> str:
await asyncio.sleep(0.1) # Non-blocking sleep simulates async network I/O
return f”Response to ‘{prompt}'”
async def run_concurrent(prompts):
start = time.perf_counter()
# Schedule all LLM calls to execute concurrently
tasks = [query_llm_async(p) for p in prompts]
results = await asyncio.gather(*tasks)
elapsed = time.perf_counter() – start
print(f”Concurrent processing took {elapsed:.4f} seconds.”)
return results
# Executing the async runner
prompts = [f”Explain topic {i}” for i in range(20)]
_ = asyncio.run(run_concurrent(prompts))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import asyncio
import time
# Mocking an asynchronous external API call to an LLM
async def query_llm_async ( prompt : str ) -> str :
await asyncio . sleep ( 0.1 ) # Non-blocking sleep simulates async network I/O
return f “Response to ‘{prompt}'”
async def run_concurrent ( prompts ) :
start = time . perf_counter ( )
# Schedule all LLM calls to execute concurrently
tasks = [ query_llm_async ( p ) for p in prompts ]
results = await asyncio . gather ( * tasks )
elapsed = time . perf_counter ( ) – start
print ( f “Concurrent processing took {elapsed:.4f} seconds.” )
return results
# Executing the async runner
prompts = [ f “Explain topic {i}” for i in range ( 20 ) ]
_ = asincio . ejecutar ( run_concurrent ( solicitudes ) )
Producción:
Concurrent processing took 0.1013 seconds.
Concurrent processing took 0.1013 seconds .
Al cambiar a asyncio, logramos una aceleración de ~20 veces para 20 llamadas API. Dado que las llamadas se ejecutan simultáneamente, el tiempo de ejecución total está limitado por la solicitud más lenta, en lugar de por la suma de todas las solicitudes.
4. Clases de datos y Pydantic (configuraciones estructuradas y validación de herramientas)
Los modelos de aprendizaje automático son muy sensibles a la configuración. Un solo error tipográfico en una clave de hiperparámetro (como tasa de aprendizaje en lugar de tasa_de_aprendizaje) puede volver silenciosamente a los valores predeterminados, haciendo que las ejecuciones de entrenamiento sean inútiles. Además, las API LLM modernas utilizan esquemas JSON estructurados para admitir llamadas de herramientas y resultados estructurados.
Las clases de datos estándar de Python proporcionan una forma limpia de definir plantillas de configuración estructuradas. Para la validación en tiempo de ejecución, Pydantic amplía este concepto, analizando tipos automáticamente, imponiendo restricciones (por ejemplo, límites de rango coincidentes) y exportando esquemas JSON listos para usar.
Depender de diccionarios sin formato para la configuración de hiperparámetros permite que los errores tipográficos y las discrepancias de tipos pasen silenciosamente, provocando errores matemáticos o comportamientos de entrenamiento inesperados:
def train_model(config: dict):
# Untyped extraction with default fallbacks
learning_rate = config.get(“learning_rate”, 0.001)
batch_size = config.get(“batch_size”, 32)
optimizer = config.get(“optimizer”, “adam”)
# Typing bug: if batch_size is passed as a string “64”, this math fails
num_steps = 1000 // batch_size
print(f”Training with LR={learning_rate}, Batch Size={batch_size}, Steps={num_steps}”)
# Typos or incorrect types pass without immediate warnings
train_model({“learning_rate”: -0.05, “batch_size”: “64”})
def train_model ( config : dict ) :
# Untyped extraction with default fallbacks
learning_rate = config . get ( “learning_rate” , 0.001 )
batch_size = config . get ( “batch_size” , 32 )
optimizer = config . get ( “optimizer” , “adam” )
# Typing bug: if batch_size is passed as a string “64”, this math fails
num_steps = 1000 // batch_size
print ( f “Training with LR={learning_rate}, Batch Size={batch_size}, Steps={num_steps}” )
# Typos or incorrect types pass without immediate warnings
train_model ( { “learning_rate” : – 0.05 , “batch_size” : “64” } )
Al definir configuraciones con Pydantic, los parámetros se analizan y verifican estrictamente al crear instancias. Esto garantiza que las configuraciones se validen antes de que se ejecute el código de capacitación y genera esquemas JSON limpios para los LLM:
from pydantic import BaseModel, Field, ValidationError
class ModelConfig(BaseModel):
learning_rate: float = Field(gt=0.0, lt=1.0, description=”Learning rate must be between 0 and 1″)
batch_size: int = Field(gt=0, description=”Batch size must be a positive integer”)
optimizer: str = Field(default=”adam”)
# Pydantic performs runtime type coercion (coercing string “64” to int 64)
try:
valid_config = ModelConfig(learning_rate=0.001, batch_size=”64″)
print(f”Valid configuration initialized: {valid_config}”)
except ValidationError as e:
print(f”Unexpected error: {e}”)
# Catching invalid parameters instantly
try:
invalid_config = ModelConfig(learning_rate=-0.05, batch_size=0)
except ValidationError as e:
print(“nValidation Errors Caught:”)
print(e)
# Export schema directly for LLM Tool / Function Calling schemas
print(“nJSON Schema for LLM Tool Definition:”)
print(ModelConfig.model_json_schema())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from pydantic import BaseModel , Field , ValidationError
class ModelConfig ( BaseModel ) :
learning_rate : float = Field ( gt = 0.0 , lt = 1.0 , description = “Learning rate must be between 0 and 1” )
tamaño_lote : entero = Campo ( gt = 0 , descripción = "El tamaño del lote debe ser un número entero positivo" )
optimizador : cadena = Campo ( predeterminado = "adam" )
# Pydantic realiza coerción de tipo en tiempo de ejecución (coaccionando la cadena "64" a int 64)
intentar :
configuración_válida = ModelConfig ( tasa_aprendizaje = 0,001 , tamaño_lote = "64" )
print ( f "Configuración válida inicializada: {valid_config}" )
excepto ValidationError como mi :
imprimir ( f "Error inesperado: {e}" )
# Detectar parámetros no válidos al instante
intentar :
configuración_inválida = ModelConfig ( tasa_aprendizaje = – 0,05 , tamaño_lote = 0 )
excepto ValidationError como mi :
print ( "nErrores de validación detectados: " )
imprimir ( mi )
# Exportar esquema directamente para esquemas de llamada de función/herramienta LLM
print ( "nEsquema JSON para la definición de la herramienta LLM: " )
imprimir ( ModelConfig . model_json_schema ( ) )
Producción:
Valid configuration initialized: learning_rate=0.001 batch_size=64 optimizer=”adam”
Validation Errors Caught:
2 validation errors for ModelConfig
learning_rate
Input should be greater than 0 [type=greater_than, input_value=-0.05, input_type=float]
For further information visit https://errors.pydantic.dev/2.12/v/greater_than
batch_size
Input should be greater than 0 [type=greater_than, input_value=0, input_type=int]
For further information visit https://errors.pydantic.dev/2.12/v/greater_than
JSON Schema for LLM Tool Definition:
{‘properties’: {‘learning_rate’: {‘description’: ‘Learning rate must be between 0 and 1’, ‘exclusiveMaximum’: 1.0, ‘exclusiveMinimum’: 0.0, ‘title’: ‘Learning Rate’, ‘type’: ‘number’}, ‘batch_size’: {‘description’: ‘Batch size must be a positive integer’, ‘exclusiveMinimum’: 0, ‘title’: ‘Batch Size’, ‘type’: ‘integer’}, ‘optimizer’: {‘default’: ‘adam’, ‘title’: ‘Optimizer’, ‘type’: ‘string’}}, ‘required’: [‘learning_rate’, ‘batch_size’], ‘title’: ‘ModelConfig’, ‘type’: ‘object’}
Valid configuration initialized : learning_rate = 0.001 batch_size = 64 optimizer = ‘adam’
Validation Errors Caught :
2 validation errors for ModelConfig
learning_rate
Input should be greater than 0 [ type = greater_than , input_value = – 0.05 , input_type = float ]
For further information visit https : //errors.pydantic.dev/2.12/v/greater_than
batch_size
Input should be greater than 0 [ type = greater_than , input_value = 0 , input_type = int ]
For further information visit https : //errors.pydantic.dev/2.12/v/greater_than
JSON Schema for LLM Tool Definition :
{ 'propiedades' : { 'tasa_de_aprendizaje' : { 'descripción' : 'La tasa de aprendizaje debe estar entre 0 y 1' , 'exclusivoMáximo' : 1.0 , 'exclusivoMínimo' : 0.0 , 'título' : 'Tasa de aprendizaje' , 'tipo' : 'número' } , 'tamaño_lote' : { 'descripción' : 'El tamaño del lote debe ser un número entero positivo' , 'exclusivoMínimo' : 0 , 'título' : 'Tamaño de lote' , 'tipo' : 'entero' } , 'optimizador' : { 'por defecto' : 'Adán' , 'título' : 'Optimizador' , 'tipo' : 'cadena' } } , 'requerido' : [ 'tasa_de_aprendizaje' , 'tamaño_lote' ] , 'título' : 'ModelConfig' , 'tipo' : 'objeto' }
El uso de Pydantic protege sus entornos de ejecución de errores de configuración, analiza entradas sin procesar de forma segura y automatiza las definiciones de esquemas para las funciones del agente.
5. Métodos mágicos (construcción de abstracciones personalizadas)
Los canales de capacitación personalizados y los motores de inferencia deben interactuar sin problemas con los ecosistemas de bibliotecas externos. Por ejemplo, si crea un cargador de texto personalizado, el DataLoader de PyTorch debería poder indexarlo y tomar muestras de él de forma natural.
Python utiliza métodos mágicos de doble guión ("dunder") para implementar interfaces de objetos. Al escribir lógica personalizada para métodos como __len__, __getitem__ y __call__, hace que sus clases personalizadas de Python actúen como listas integradas o funciones ejecutables.
Escribamos una clase personalizada con nombres de métodos arbitrarios. Este conjunto de datos no se puede pasar directamente a bibliotecas externas que esperan protocolos estándar de Python:
class CustomDataset:
def __init__(self, data_list):
self.data_list = data_list
def fetch_index(self, i):
return self.data_list[i]
def count_items(self):
return len(self.data_list)
dataset = CustomDataset([“Sample A”, “Sample B”, “Sample C”])
# Client code is forced to learn custom APIs
print(f”Items: {dataset.count_items()}, First item: {dataset.fetch_index(0)}”)
# Trying len(dataset) or dataset[0]desencadena una impresión TypeError(f"Longitud del conjunto de datos: {len(conjunto de datos)}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
clase Conjunto de datos personalizado :
def __init__ ( yo , lista_datos ) :
ser . lista_datos = lista_datos
def fetch_index ( auto , i ) :
devolver ser . lista_datos [ i ]
def contar_elementos ( yo ) :
devolver len ( self . lista_datos )
conjunto de datos = Conjunto de datos personalizado ( [ "Muestra A" , "Muestra B" , "Muestra C" ] )
# El código del cliente se ve obligado a aprender API personalizadas
print ( f "Elementos: {dataset.count_items()}, Primer elemento: {dataset.fetch_index(0)}" )
# Probando len(conjunto de datos) o conjunto de datos[0]desencadena un error de tipo
print ( f "Longitud del conjunto de datos: {len(conjunto de datos)}" )
Producción:
Items: 3, First item: Sample A
Traceback (most recent call last):
File “./testing.py”, line 15, in
print(f”Dataset length: {len(dataset)}”)
^^^^^^^^^^^^
TypeError: object of type ‘CustomDataset’ has no len()
Elementos : 3 , Primer artículo : Muestra A
Rastreo ( última llamada más reciente ) :
Archivo "./prueba.py" , línea 15 , en < módulo >
print ( f "Longitud del conjunto de datos: {len(conjunto de datos)}" )
^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^
Error de tipo : objeto de tipo 'Conjunto de datos personalizado' no tiene longitud ( )
Al implementar __len__ y __getitem__, hacemos que nuestra clase actúe como una secuencia nativa. Al implementar __call__, hacemos que nuestra instancia de canalización de inferencia personalizada se comporte como una función:
class CustomDatasetPythonic:
def __init__(self, data_list):
self.data = data_list
def __len__(self) -> int:
return len(self.data)
def __getitem__(self, idx: int):
return self.data[idx]
class PredictionPipeline:
def __init__(self, step_value: float):
self.step_value = step_value
def __call__(self, x: float) -> float:
# Implementing __call__ makes instances callable like functions
return x * self.step_value
# Instantiating the protocol-compatible dataset
dataset = CustomDatasetPythonic([“Sample A”, “Sample B”, “Sample C”])
print(f”Dataset length: {len(dataset)}”)
print(f”Index access[1]: {conjunto de datos[1]}") # Crear una instancia de la canalización invocable pipeline = PredictionPipeline(step_value=2.5) # Llamar al objeto directamente resultado = pipeline(10.0) print(f"Resultado de la ejecución de la llamada de canalización: {resultado}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
clase Conjunto de datos personalizadoPythonic :
def __init__ ( yo , lista_datos ) :
ser . datos = lista_datos
def __len__ ( yo ) -> entero :
devolver len ( auto . datos )
def __getitem__ ( yo , identificación : int ) :
devolver ser . datos [ idx ]
clase Tubería de predicción :
def __init__ ( yo , valor_paso : flotar ) :
ser . valor_paso = valor_paso
def __llamar__ ( yo , x : flotar ) -> flotar :
# La implementación de __call__ hace que las instancias sean invocables como funciones
devolver x* ser . paso _ valor
# Creación de instancias del conjunto de datos compatible con el protocolo
conjunto de datos = CustomDatasetPythonic ( [ "Muestra A" , "Muestra B" , "Muestra C" ] )
print ( f "Longitud del conjunto de datos: {len(conjunto de datos)}" )
print ( f "Acceso al índice[1]: {conjunto de datos[1]}" )
# Creación de instancias de la canalización invocable
tubería = PredictionPipeline ( valor_paso = 2,5 )
# Llamar al objeto directamente
resultado = tubería ( 10.0 )
print ( f "Resultado de la ejecución de la llamada de canalización: {resultado}" )
Producción:
Dataset length: 3
Index access[1]: Resultado de ejecución de llamada de canalización de muestra B: 25,0
Longitud del conjunto de datos : 3
Acceso al índice [ 1 ] : Muestra B
Resultado de la ejecución de la llamada de canalización : 25.0
En las bibliotecas de aprendizaje profundo, adquiera el hábito de ejecutar capas o modelos utilizando la sintaxis de llamada (modelo(x)) en lugar de llamar explícitamente al método de avance (model.forward(x)). El nn.Module base de PyTorch anula __call__ para registrar y ejecutar enlaces hacia atrás/adelante antes de llamar hacia adelante(). La ejecución directa de .forward() omite estos ganchos, lo que provoca gradientes rotos o errores de seguimiento.
Concluyendo
La transición de portátiles simples a aplicaciones robustas de IA requiere el uso de los mecanismos de ingeniería nativos de Python para escribir código eficiente, legible y limpio.
Estas son las conclusiones clave:
Transmita datos con generadores para mantener el uso de la memoria estable al procesar grandes conjuntos de datos. Administre los estados del sistema y del hardware de manera limpia con administradores de contexto para proteger los límites de su GPU. Resuelva los cuellos de botella de la red al consultar API externas mediante el uso de canalizaciones asincio concurrentes. Proteja configuraciones y genere automáticamente esquemas para herramientas LLM usando modelos de validación de Pydantic. Integre abstracciones personalizadas de manera limpia en paquetes de marco mediante la implementación de métodos mágicos.
Al tratar sus canales de código con rigor en ingeniería de software, garantiza que sus sistemas de IA funcionen rápido, fallen de manera segura y se integren limpiamente con la infraestructura de producción.