Python SDK Leitfaden

Das KLA Python Telemetry SDK (kla-otel-python) bietet automatisches Tracing, Budget-Guardrails und Compliance-Logs für Python-Frameworks. Es ist für die Zusammenarbeit mit FastAPI, LangChain und gängigen HTTP-Ausführungsmustern ausgelegt.

---

Installation

Installieren Sie das Paket über pip oder binden Sie es in Ihre Umgebungsdefinitionen ein:

pip install kla-otel-python

---

Auto-Instrumentierung (FastAPI)

Für Web-APIs, die Modelleingaben weiterleiten, instanziieren Sie die KLA Middleware:

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

app = FastAPI()

# Telemetrie-Collector-Middleware einbinden
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"}

---

LangChain-Integration

Wenn Sie LangChain zur Orchestrierung von Prompts und Memory verwenden, aktivieren Sie den KLA Callback-Handler, um Agents, Tools und Token-Grenzen automatisch zu tracen:

from langchain_openai import ChatOpenAI
from kla_otel.callbacks import KLACallbackHandler

# Den KLA Callback-Tracer laden
kla_callback = KLACallbackHandler(
    agent_id="agt_9f81a7",
    tenant_id="kpmg-reporting"
)

# Callback an die Modellausführung anhängen
model = ChatOpenAI(
    model_name="gpt-4o",
    temperature=0,
    callbacks=[kla_callback]
)

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

---

Dynamische Aktionsbewertungen

Für manuelle Sicherheits-Overrides (etwa beim Ändern von Dateien oder Anfordern von API-Schlüsseln) bewerten Sie Aktionsobjekte direkt:

from kla_otel import KLACheckpoint

checkpoint = KLACheckpoint(agent_id="agt_9f81a7")

def delete_user_account(user_id: str):
    # Sicherheitsrichtlinien-Bewertung durchführen
    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:
        # Aktion ist ausgesetzt, Decision Desk-Warteschlange wird geprüft
        checkpoint.log_escalation(
            user_id=user_id,
            reason="Requires tenant owner approval"
        )
        raise PermissionError("Deletion pending human approval desk review")