Observabilidad y monitorización de LLMs en producción

				
					import json
import openai
from openai import OpenAI

import opik
from opik import opik_context
from opik.integrations.openai import track_openai

# Instancia del cliente de OpenAI, instrumentado con Opik
cliente = track_openai(OpenAI())

@opik.track(flush=True)
def generar_receta(nombre_plato: str) -> tuple[str, dict]:
    """
    Genera una receta de cocina a partir del nombre de un plato, evalúa
    su calidad y recopila feedback del usuario. Retorna el texto de la
    receta y un diccionario con la evaluación.
    """
    # Prompt principal para generar la receta
    mensajes = [
        {"role": "system", "content": "Eres un chef experto."},
        {
            "role": "user",
            "content": (
                f"Eres un chef de primera clase. Genera una receta para {nombre_plato}.\n"
                "La receta debe incluir:\n"
                "- Lista de ingredientes con cantidades\n"
                "- Pasos detallados de preparación\n"
                "- Tiempo de cocción\n"
                "- Nivel de dificultad\n"
                "- Calorías aproximadas"
            ),
        },
    ]

    respuesta = cliente.chat.completions.create(
        model="gpt-4o-mini",
        messages=mensajes,
        max_tokens=500,
        temperature=0.7
    )

    receta_texto = respuesta.choices[0].message.content

    # Evalúa la receta generada
    evaluacion = evaluar_receta(receta_texto, nombre_plato)

    # Muestra la receta en consola
    print("\n" + "=" * 50)
    print(receta_texto)
    print("=" * 50)

    # Advertencia en caso de "alucinaciones" detectadas
    if evaluacion['tiene_alucinaciones']:
        print("⚠️ Advertencia: Esta receta puede contener información incorrecta.")

    # Recopila feedback del usuario
    feedback, comentario = obtener_feedback_usuario()

    # Envía métricas de feedback a Opik
    opik_context.update_current_trace(
        feedback_scores=[
            {
                "name": "calidad_receta",
                "value": evaluacion["puntuacion"] / 10.0,
                "reason": evaluacion["explicacion"],
            },
            {
                "name": "formato_correcto",
                "value": 1.0 if evaluacion["formato_correcto"] else 0.0,
            },
            {
                "name": "sin_alucinaciones",
                "value": 0.0 if evaluacion["tiene_alucinaciones"] else 1.0,
            },
            {
                "name": "feedback_usuario",
                "value": feedback / 5.0 if feedback is not None else 0.0,
                "reason": comentario if comentario else "Sin comentarios",
            },
        ]
    )

    return receta_texto, evaluacion

def evaluar_receta(receta_texto: str, plato_solicitado: str) -> dict:
    """
    Envía la receta a un evaluador (otro modelo) para comprobar su
    calidad, formato y posibles alucinaciones. Devuelve un diccionario
    con los campos:
        - puntuacion (int, 1 a 10)
        - tiene_alucinaciones (bool)
        - formato_correcto (bool)
        - explicacion (str)
    """
    prompt_evaluador = (
        f"Eres un experto evaluador de recetas. Analiza la siguiente receta para {plato_solicitado}:\n\n"
        f"{receta_texto}\n\n"
        "Evalúa los siguientes aspectos y devuelve un JSON con estos campos:\n"
        "1. puntuacion: del 1 al 10, ¿qué tan buena es esta receta?\n"
        "2. tiene_alucinaciones: true/false, ¿contiene ingredientes imposibles o pasos irreales?\n"
        "3. formato_correcto: true/false, ¿incluye todos los elementos solicitados?\n"
        "4. explicacion: breve explicación de tu evaluación\n"
    )

    try:
        evaluacion_respuesta = cliente.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Eres un evaluador de recetas objetivo."},
                {"role": "user", "content": prompt_evaluador},
            ],
            max_tokens=300,
            temperature=0.2,
            response_format={"type": "json_object"}
        )
        return json.loads(evaluacion_respuesta.choices[0].message.content)

    except Exception as e:
        print(f"Error en evaluación: {e}")
        return {
            "puntuacion": 5,
            "tiene_alucinaciones": False,
            "formato_correcto": False,
            "explicacion": "Error en la evaluación automática"
        }

def obtener_feedback_usuario() -> tuple[int | None, str | None]:
    """
    Solicita al usuario una puntuación de 1 a 5 y un comentario adicional.
    Retorna la puntuación y el comentario (o None si hay error).
    """
    print("\n--- Ayúdanos a mejorar ---")
    try:
        puntuacion = int(input("Del 1 al 5, ¿qué tan útil fue esta receta? "))
        comentario = input("¿Algún comentario adicional? (opcional): ")
        print(f"Feedback registrado: {puntuacion}/5")
        if comentario:
            print(f"Comentario: {comentario}")
        return puntuacion, comentario
    except Exception as e:
        print(f"Error al registrar feedback: {e}")
        return None, None

def main():
    print("🍳 Asistente de Recetas con Observabilidad 🍳")
    plato = input("¿Qué plato quieres cocinar?: ")
    generar_receta(plato)

if __name__ == "__main__":
    main()