async.md

Manejo Asíncrono

Visión General

El sistema utiliza programación asíncrona para optimizar el rendimiento y manejar múltiples solicitudes simultáneas de manera eficiente.

Implementación

Controlador Asíncrono

@descriptions_bp.route("/generate", methods=["POST"])
async def generate_descriptions():
    try:
        data = request.get_json()
        product_request = ProductRequest(**data)
        response_data = await generate_descriptions_for_product(product_request)
        return jsonify(response_data), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

Servicio OpenAI Asíncrono

class OpenAIService:
    def __init__(self):
        self.client = AsyncOpenAI(api_key=Config.OPENAI_API_KEY)

    async def generate_text(self, prompt: str, **kwargs):
        response = await self.client.chat.completions.create(
            model=self.default_model,
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )
        return response

Manejo de Concurrencia

Procesamiento en Paralelo

async def process_multiple_fields(product_request):
    tasks = []
    for field, prompt in product_request.prompts.items():
        task = generate_field(field, prompt)
        tasks.append(task)
    results = await asyncio.gather(*tasks)
    return results

Rate Limiting

from asyncio import Semaphore

class RateLimiter:
    def __init__(self, limit):
        self.semaphore = Semaphore(limit)

    async def __aenter__(self):
        await self.semaphore.acquire()

    async def __aexit__(self, exc_type, exc, tb):
        self.semaphore.release()

Patrones Asíncronos

Circuit Breaker

class AsyncCircuitBreaker:
    def __init__(self, failure_threshold):
        self.failures = 0
        self.threshold = failure_threshold
        self.state = "closed"

    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            raise Exception("Circuit breaker is open")
        
        try:
            result = await func(*args, **kwargs)
            self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.threshold:
                self.state = "open"
            raise

Retry Pattern

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def retry_operation():
    # Operación que puede fallar
    pass

Buenas Prácticas

1. Manejo de Errores

• Usar try/except en operaciones asíncronas
• Implementar timeouts apropiados
• Logging asíncrono

2. Optimización

• Evitar bloqueos largos
• Usar asyncio.gather para operaciones paralelas
• Implementar backoff exponencial

3. Monitorización

• Tracking de tiempos de respuesta
• Métricas de concurrencia
• Logs de operaciones asíncronas

Ejemplos de Uso

Procesamiento por Lotes

async def batch_process_products(products: List[ProductRequest]):
    batch_size = 5
    results = []
    
    for i in range(0, len(products), batch_size):
        batch = products[i:i + batch_size]
        batch_results = await asyncio.gather(
            *[process_product(p) for p in batch]
        )
        results.extend(batch_results)
    
    return results

Cancelación de Tareas

async def process_with_timeout(request):
    try:
        async with asyncio.timeout(30):
            return await process_request(request)
    except asyncio.TimeoutError:
        # Manejar timeout
        pass

Debugging

Logs Asíncronos

async def async_logger(message, level="INFO"):
    await asyncio.to_thread(
        logging.log,
        getattr(logging, level),
        message
    )

Traceo de Tareas

async def traced_operation():
    task = asyncio.current_task()
    task.set_name("operation_name")
    # Resto de la operación

Consideraciones de Rendimiento

1. Memory Management

   • Liberar recursos apropiadamente
   • Evitar memory leaks
   • Monitorear uso de memoria

2. CPU Bound vs IO Bound

   • Usar ProcessPoolExecutor para CPU
   • ThreadPoolExecutor para I/O
   • Balancear cargas

3. Escalabilidad

   • Limitar concurrencia máxima
   • Implementar backpressure
   • Monitorear recursos