#llm_with_tracing.py
# Propósito: un contenedor de llamadas de LLM listo para producción con total observabilidad.
# Cada llamada se rastrea en Langfuse: entrada, salida, tokens, costo, latencia.
#
# Requisitos previos:
# pip install langfuse antrópico python-dotenv
#
# Configuración:
# 1. Cree una cuenta gratuita en https://cloud.langfuse.com
# 2. Obtenga sus claves en Configuración> Claves API
# 3. Cree un archivo .env con las siguientes variables
#
# Correr:
# python llm_with_tracing.py
importar sistema operativo
importar tiempo
de dotenv importar carga_dotenv
importar antrópico
de Langfuse importar Langfuse
# Cargar variables de entorno desde el archivo .env
carga_dotenv()
# Variables de entorno requeridas en su .env:
# LANGFUSE_PUBLIC_KEY=pk-lf-…
# LANGFUSE_SECRET_KEY=sk-lf-…
# LANGFUSE_HOST=https://cloud.langfuse.com (o su URL autohospedada)
# ANTHROPIC_API_KEY=sk-hormiga-…
# Inicializar clientes
langfuse_client = Langfuse() # Lee claves automáticamente desde el entorno.
cliente_antrópico = antrópico.antrópico() # Lee ANTHROPIC_API_KEY del entorno
# ── Configuración ────────────────────────────── ───────────────────────────────
# Guarde su mensaje aquí, no en línea en la llamada API.
# Esto lo hace versionable y comprobable de forma independiente.
SISTEMA_PROMPT = “”“Eres un útil asistente de atención al cliente.
Responda las preguntas de forma clara y concisa.
Si no sabes algo, dilo directamente, no lo adivines.”“”
MODELO = “claude-soneto-4-20250514”
# Precios de Anthropic a mediados de 2026 (actualización cuando cambien los precios)
# Se utiliza para calcular el costo por llamada para el seguimiento de costos
COSTO_PER_INPUT_TOKEN = 3.00 / 1_000_000 # $3.00 por millón de tokens de entrada
COSTE_PER_OUTPUT_TOKEN = 15.00 / 1_000_000 # $15.00 por millón de tokens de salida
definición call_llm_with_tracing(
mensaje_usuario: cadena,
id_sesión: cadena = “sesión predeterminada”,
ID_usuario: cadena = “anónimo”
) -> cadena:
“”“
Realice una llamada de LLM rastreada. Cada llamada crea un seguimiento Langfuse con:
– Entrada y salida completa
– Uso de tokens (entrada, salida, total)
– Costo calculado en USD
– Latencia en milisegundos
– Modelo utilizado y contexto de la sesión.
Parámetros:
user_message: El mensaje del usuario.
session_id: Agrupa llamadas relacionadas en una conversación en Langfuse
user_id: asocia la llamada con un usuario específico para análisis
Devoluciones:
La respuesta LLM como una cadena
““”
# Crear un seguimiento de nivel superior para esta interacción de usuario
# El seguimiento aparece en el panel de Langfuse como una unidad de trabajo.
rastro = langfuse_client.rastro(
nombre=“llamada-de-atención-al-cliente”,
id_sesión=id_sesión,
ID_usuario=ID_usuario,
aporte={“mensaje_usuario”: mensaje_usuario, “sistema_prompt”: SISTEMA_PROMPT}
)
# Crear un lapso de generación dentro del rastro
# Esto captura detalles específicos del modelo: nombre del modelo, tokens, costo
generación = rastro.generación(
nombre=“claude-completado”,
modelo=MODELO,
aporte={
“sistema”: SISTEMA_PROMPT,
“mensajes”: [{“role”: “user”, “content”: user_message}]
}
)
hora_inicio = tiempo.tiempo()
intentar:
# Realizar la llamada API
respuesta = cliente_antrópico.mensajes.crear(
modelo=MODELO,
tokens_max=1024,
sistema=SISTEMA_PROMPT,
mensajes=[{“role”: “user”, “content”: user_message}]
)
latencia_ms = entero((tiempo.tiempo() – hora_inicio) * 1000)
# Extraer el texto de respuesta
texto_respuesta = respuesta.contenido[0].texto
# Extraer el uso del token de la respuesta
tokens_entrada = respuesta.uso.tokens_entrada
tokens_de_salida = respuesta.uso.tokens_de_salida
tokens_totales = tokens_entrada + producción_fichas
# Calcular el costo de esta llamada
costo_usd = (
tokens de entrada * COSTO_PER_INPUT_TOKEN +
tokens_de_salida * COSTO_PER_SALIDA_SIMBÓLICO
)
# Actualizar el lapso de generación con resultados.
# Estos datos completan los paneles de costos y tokens de Langfuse
generación.fin(
producción=texto_respuesta,
uso={
“aporte”: tokens_entrada,
“producción”: tokens_de_salida,
“total”: tokens_totales,
“unidad”: “FICHAS”
},
metadatos={
“latencia_ms”: latencia_ms,
“costo_usd”: redondo(costo_usd, 6),
“modelo”: MODELO
}
)
# Actualizar el seguimiento con el resultado final.
rastro.actualizar(
producción={“respuesta”: texto_respuesta},
metadatos={“costo_total_usd”: redondo(costo_usd, 6)}
)
# Imprima un resumen en stdout para visibilidad local
imprimir(F“\n{‘─’ * 60}”)
imprimir(F“Usuario: {user_message}”)
imprimir(F“Claude: {respuesta_texto}”)
imprimir(F“Tokens: {input_tokens} entrantes / {output_tokens} salientes / {total_tokens} total”)
imprimir(F“Costo: ${cost_usd:.6f}”)
imprimir(F“Latencia: {latency_ms}ms”)
imprimir(F“Seguimiento: {langfuse_client.base_url}/trace/{trace.id}”)
imprimir(F“{‘─’ * 60}\n”)
devolver texto_respuesta
excepto Excepción como mi:
# Registrar el error en el seguimiento para que aparezca en Langfuse
generación.fin(
producción=Ninguno,
metadatos={“error”: cadena(mi), “latencia_ms”: entero((tiempo.tiempo() – hora_inicio) * 1000)}
)
rastro.actualizar(producción={“error”: cadena(mi)})
# Siempre vacíe antes de aumentar: garantiza que se envíe el seguimiento del error
langfuse_client.enjuagar()
aumentar
finalmente:
# Flush envía todos los eventos almacenados en el buffer a Langfuse
# En un servicio de larga duración, Langfuse se descarga automáticamente.
# En un script, debe vaciar manualmente antes de que finalice el proceso.
langfuse_client.enjuagar()
# ── Realizar una demostración ──────────────────────────── ────────────────────────────
si __nombre__ == “__principal__”:
# Simular dos turnos de una conversación de atención al cliente
mensajes_prueba = [
“What is your return policy for electronics?”,
“Can I return an item I bought 45 days ago?”
]
sesión = “sesión-demo-001”
para i, mensaje en enumerar(mensajes_prueba):
imprimir(F“\nLlamar a {i + 1}/{len(test_messages)}”)
intentar:
call_llm_with_tracing(
mensaje_usuario=mensaje,
id_sesión=sesión,
ID_usuario=“usuario-de-prueba-42”
)
excepto Excepción como mi:
imprimir(F“Error en llamada {i + 1}: {e}”)