# llm_with_tracing.py
# Zweck: Ein produktionsbereiter LLM-Aufruf-Wrapper mit vollständiger Beobachtbarkeit.
# Jeder Anruf wird in Langfuse verfolgt: Eingabe, Ausgabe, Token, Kosten, Latenz.
#
# Voraussetzungen:
# pip set up langfuse anthropic python-dotenv
#
# Aufstellen:
# 1. Erstellen Sie ein kostenloses Konto unter https://cloud.langfuse.com
# 2. Holen Sie sich Ihre Schlüssel über Einstellungen > API-Schlüssel
# 3. Erstellen Sie eine .env-Datei mit den folgenden Variablen
#
# Laufen:
# python llm_with_tracing.py
Import os
Import Zeit
aus dotenv Import load_dotenv
Import anthropisch
aus langfuse Import Langfuse
# Umgebungsvariablen aus der .env-Datei laden
load_dotenv()
# Erforderliche Umgebungsvariablen in Ihrer .env:
# LANGFUSE_PUBLIC_KEY=pk-lf-…
# LANGFUSE_SECRET_KEY=sk-lf-…
# LANGFUSE_HOST=https://cloud.langfuse.com (oder Ihre selbst gehostete URL)
# ANTHROPIC_API_KEY=sk-ant-…
# Purchasers initialisieren
langfuse_client = Langfuse() # Liest Schlüssel automatisch aus der Umgebung
anthropic_client = anthropisch.Anthropisch() # Liest ANTHROPIC_API_KEY aus der Umgebung
# ── Konfiguration ────────────────────────────── ───────────────────────────────
# Speichern Sie Ihre Eingabeaufforderung hier, nicht inline im API-Aufruf.
# Dies macht es unabhängig versionierbar und testbar.
SYSTEM_PROMPT = „“„Sie sind ein hilfreicher Kundendienstassistent.
Beantworten Sie Fragen klar und prägnant.
Wenn Sie etwas nicht wissen, sagen Sie es direkt – raten Sie nicht.“„“
MODELL = „claude-sonnet-4-20250514“
# Preise von Anthropic ab Mitte 2026 (Aktualisierung bei Preisänderungen)
# Wird zur Berechnung der Kosten professional Anruf zur Kostenverfolgung verwendet
COST_PER_INPUT_TOKEN = 3,00 / 1_000_000 # 3,00 $ professional Million Enter-Tokens
COST_PER_OUTPUT_TOKEN = 15.00 / 1_000_000 # 15,00 $ professional Million ausgegebener Token
def call_llm_with_tracing(
user_message: str,
Sitzungs-ID: str = „Standardsitzung“,
Benutzer-ID: str = „anonym“
) -> str:
„“„
Führen Sie einen verfolgten LLM-Anruf durch. Jeder Aufruf erstellt einen Langfuse-Hint mit:
– Vollständige Ein- und Ausgabe
– Token-Nutzung (Eingabe, Ausgabe, Gesamt)
– Berechnete Kosten in USD
– Latenz in Millisekunden
– Verwendetes Modell und Sitzungskontext
Parameter:
user_message: Die Nachricht des Benutzers
session_id: Gruppiert verwandte Anrufe in einer Konversation in Langfuse
user_id: Ordnet den Anruf einem bestimmten Benutzer für die Analyse zu
Rückgaben:
Die LLM-Antwort als String
„„“
# Erstellen Sie einen Hint der obersten Ebene für diese Benutzerinteraktion
# Der Hint erscheint im Langfuse-Dashboard als eine Arbeitseinheit
verfolgen = langfuse_client.verfolgen(
Title=„Kundensupport-Anruf“,
Sitzungs-ID=Sitzungs-ID,
Benutzer-ID=Benutzer-ID,
Eingang={„user_message“: user_message, „system_prompt“: SYSTEM_PROMPT}
)
# Erstellen Sie eine Generationsspanne innerhalb des Hint
# Dadurch werden modellspezifische Particulars erfasst: Modellname, Token, Kosten
Era = verfolgen.Era(
Title=„Claude-Vervollständigung“,
Modell=MODELL,
Eingang={
„System“: SYSTEM_PROMPT,
„Nachrichten“: ({„Rolle“: „Benutzer“, „Inhalt“: user_message})
}
)
start_time = Zeit.Zeit()
versuchen:
# Führen Sie den API-Aufruf durch
Antwort = anthropic_client.Nachrichten.erstellen(
Modell=MODELL,
max_tokens=1024,
System=SYSTEM_PROMPT,
Nachrichten=({„Rolle“: „Benutzer“, „Inhalt“: user_message})
)
latenz_ms = int((Zeit.Zeit() – start_time) * 1000)
# Extrahieren Sie den Antworttext
Antworttext = Antwort.Inhalt(0).Textual content
# Token-Nutzung aus der Antwort extrahieren
Eingabetokens = Antwort.Verwendung.Eingabetokens
Ausgabetokens = Antwort.Verwendung.Ausgabetokens
total_tokens = Eingabetokens + Ausgabe_Token
# Berechnen Sie die Kosten für diesen Anruf
cost_usd = (
input_tokens * COST_PER_INPUT_TOKEN +
Ausgabetokens * COST_PER_OUTPUT_TOKEN
)
# Aktualisieren Sie die Generationsspanne mit Ergebnissen
# Diese Daten füllen die Langfuse-Kosten- und Token-Dashboards
Era.Ende(
Ausgabe=Antworttext,
Verwendung={
„Eingang“: Eingabetokens,
„Ausgabe“: Ausgabetokens,
„gesamt“: total_tokens,
„Einheit“: „TOKENS“
},
Metadaten={
„latency_ms“: latenz_ms,
„cost_usd“: runden(cost_usd, 6),
„Modell“: MODELL
}
)
# Aktualisieren Sie den Hint mit der endgültigen Ausgabe
verfolgen.aktualisieren(
Ausgabe={„Antwort“: Antworttext},
Metadaten={„total_cost_usd“: runden(cost_usd, 6)}
)
# Drucken Sie eine Zusammenfassung zur lokalen Sichtbarkeit auf stdout aus
drucken(F„n{‚─‘ * 60}“)
drucken(F„Benutzer: {user_message}“)
drucken(F„Claude: {response_text}“)
drucken(F„Tokens: {input_tokens} rein / {output_tokens} raus / {total_tokens} gesamt“)
drucken(F„Kosten: ${cost_usd:.6f}“)
drucken(F„Latenz: {latency_ms}ms“)
drucken(F„Hint: {langfuse_client.base_url}/hint/{hint.id}“)
drucken(F„{‚─‘ * 60}n“)
zurückkehren Antworttext
außer Ausnahme als e:
# Notieren Sie den Fehler im Hint, damit er in Langfuse angezeigt wird
Era.Ende(
Ausgabe=Keiner,
Metadaten={„Fehler“: str(e), „latency_ms“: int((Zeit.Zeit() – start_time) * 1000)}
)
verfolgen.aktualisieren(Ausgabe={„Fehler“: str(e)})
# Vor dem Erhöhen immer leeren – stellt sicher, dass die Fehlerverfolgung gesendet wird
langfuse_client.spülen()
erheben
Endlich:
# Flush sendet alle gepufferten Ereignisse an Langfuse
# Bei einem lang andauernden Dienst wird Langfuse automatisch geleert.
# In einem Skript müssen Sie manuell leeren, bevor der Prozess beendet wird.
langfuse_client.spülen()
# ── Führen Sie eine Demonstration durch ──────────────────────────── ────────────────────────────
Wenn __Name__ == „__hauptsächlich__“:
# Simulieren Sie zwei Runden eines Kundensupportgesprächs
test_messages = (
„Wie lauten Ihre Rückgabebedingungen für Elektronikgeräte?“,
„Kann ich einen Artikel zurückgeben, den ich vor 45 Tagen gekauft habe?“
)
Sitzung = „demo-session-001“
für ich, Nachricht In aufzählen(test_messages):
drucken(F„nCall {i + 1}/{len(test_messages)}“)
versuchen:
call_llm_with_tracing(
user_message=Nachricht,
Sitzungs-ID=Sitzung,
Benutzer-ID=„test-user-42“
)
außer Ausnahme als e:
drucken(F„Fehler beim Aufruf {i + 1}: {e}“)