Guide du SDK Python

Le KLA Python Telemetry SDK (kla-otel-python) assure le traçage automatique, des garde-fous budgétaires et des journaux de conformité pour les frameworks Python. Il est conçu pour fonctionner avec FastAPI, LangChain et les schémas d'exécution HTTP standard.

---

Installation

Installez le paquet via pip ou incluez-le dans vos définitions d'environnement :

pip install kla-otel-python

---

Auto-instrumentation (FastAPI)

Pour les API web qui acheminent des entrées de modèle, instanciez le middleware KLA :

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

app = FastAPI()

# Monter le middleware du collecteur de télémétrie
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"}

---

Intégration LangChain

Si vous utilisez LangChain pour orchestrer les prompts et la mémoire, activez le gestionnaire de callbacks KLA afin de tracer automatiquement les agents, les outils et les limites de tokens :

from langchain_openai import ChatOpenAI
from kla_otel.callbacks import KLACallbackHandler

# Charger le traceur de callback KLA
kla_callback = KLACallbackHandler(
    agent_id="agt_9f81a7",
    tenant_id="kpmg-reporting"
)

# Attacher le callback à l'exécution de votre modèle
model = ChatOpenAI(
    model_name="gpt-4o",
    temperature=0,
    callbacks=[kla_callback]
)

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

---

Évaluations d'actions dynamiques

Pour les dérogations de sécurité manuelles (comme la modification de fichiers ou la demande de clés API), évaluez directement les objets d'action :

from kla_otel import KLACheckpoint

checkpoint = KLACheckpoint(agent_id="agt_9f81a7")

def delete_user_account(user_id: str):
    # Effectuer l'évaluation de la politique de sécurité
    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 suspendue, vérification de la file d'attente Decision Desk
        checkpoint.log_escalation(
            user_id=user_id,
            reason="Requires tenant owner approval"
        )
        raise PermissionError("Deletion pending human approval desk review")