Guida all'SDK Python

Il KLA Python Telemetry SDK (kla-otel-python) offre tracing automatico, guardrail di budget e log di conformità per i framework Python. È progettato per funzionare con FastAPI, LangChain e i pattern standard di esecuzione HTTP.

---

Installazione

Installa il pacchetto tramite pip oppure includilo nelle tue definizioni di ambiente:

pip install kla-otel-python

---

Strumentazione automatica (FastAPI)

Per le API web che instradano gli input dei modelli, istanzia il middleware 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"}

---

Integrazione con LangChain

Se utilizzi LangChain per orchestrare prompt e memoria, abilita il callback handler di KLA per tracciare automaticamente agenti, tool e i confini dei token:

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.")

---

Valutazioni dinamiche delle azioni

Per gli override manuali di sicurezza (come la modifica di file o la richiesta di chiavi API), valuta direttamente gli oggetti azione:

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")