Guía del SDK de Python

El KLA Python Telemetry SDK (kla-otel-python) ofrece trazas automáticas, controles de presupuesto y registros de cumplimiento para frameworks de Python. Está diseñado para funcionar con FastAPI, LangChain y patrones estándar de ejecución HTTP.

---

Instalación

Instala el paquete con pip o inclúyelo en las definiciones de tu entorno:

pip install kla-otel-python

---

Instrumentación automática (FastAPI)

Para las API web que enrutan entradas de modelos, instancia el middleware de KLA:

# main.py
import os
from fastapi import FastAPI
from kla_otel import KLATelemetryMiddleware

app = FastAPI()

# Mount telemetry collector middleware
app.add_middleware(
    KLATelemetryMiddleware,
    tenant_id=os.getenv("KLA_TENANT_ID"),
    api_key=os.getenv("KLA_API_KEY"),
    endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://api.kla.digital/collector"),
    mask_pii=True
)

@app.get("/")
def read_root():
    return {"status": "governed"}

---

Integración con LangChain

Si usas LangChain para orquestar prompts y memoria, activa el callback handler de KLA para trazar automáticamente los agentes, las herramientas y los límites de tokens:

from langchain_openai import ChatOpenAI
from kla_otel.callbacks import KLACallbackHandler

# Load the KLA callback tracer
kla_callback = KLACallbackHandler(
    agent_id="agt_9f81a7",
    tenant_id="kpmg-reporting"
)

# Attach callback to your model execution
model = ChatOpenAI(
    model_name="gpt-4o",
    temperature=0,
    callbacks=[kla_callback]
)

response = model.invoke("Analyze customer invoice records for Q1.")

---

Evaluaciones dinámicas de acciones

Para anulaciones manuales de seguridad (como modificar archivos o solicitar claves de API), evalúa los objetos de acción directamente:

from kla_otel import KLACheckpoint

checkpoint = KLACheckpoint(agent_id="agt_9f81a7")

def delete_user_account(user_id: str):
    # Perform safety policy evaluation
    decision = checkpoint.evaluate_action(
        action="delete_account",
        context={"user_id": user_id}
    )

    if decision.is_approved():
        db.delete(user_id)
        checkpoint.log_success(user_id=user_id)
    else:
        # Action is suspended, checking Decision Desk queue
        checkpoint.log_escalation(
            user_id=user_id,
            reason="Requires tenant owner approval"
        )
        raise PermissionError("Deletion pending human approval desk review")