# llm_with_tracing.py
# Function: A production-ready LLM name wrapper with full observability.
# Each name is traced in Langfuse: enter, output, tokens, value, latency.
#
# Stipulations:
# pip set up langfuse anthropic python-dotenv
#
# Setup:
# 1. Create a free account at https://cloud.langfuse.com
# 2. Get your keys from Settings > API Keys
# 3. Create a .env file with the variables beneath
#
# Run:
# python llm_with_tracing.py
import os
import time
from dotenv import load_dotenv
import anthropic
from langfuse import Langfuse
# Load surroundings variables from .env file
load_dotenv()
# Required surroundings variables in your .env:
# LANGFUSE_PUBLIC_KEY=pk-lf-…
# LANGFUSE_SECRET_KEY=sk-lf-…
# LANGFUSE_HOST=https://cloud.langfuse.com (or your self-hosted URL)
# ANTHROPIC_API_KEY=sk-ant-…
# Initialize purchasers
langfuse_client = Langfuse() # Reads keys mechanically from surroundings
anthropic_client = anthropic.Anthropic() # Reads ANTHROPIC_API_KEY from surroundings
# ── Configuration ─────────────────────────────────────────────────────────────
# Retailer your immediate right here, not inline within the API name.
# This makes it versionable and testable independently.
SYSTEM_PROMPT = “”“You’re a useful buyer help assistant.
Reply questions clearly and concisely.
Should you have no idea one thing, say so immediately — don’t guess.”“”
MODEL = “claude-sonnet-4-20250514”
# Anthropic’s pricing as of mid-2026 (replace when pricing modifications)
# Used to calculate value per name for value monitoring
COST_PER_INPUT_TOKEN = 3.00 / 1_000_000 # $3.00 per million enter tokens
COST_PER_OUTPUT_TOKEN = 15.00 / 1_000_000 # $15.00 per million output tokens
def call_llm_with_tracing(
user_message: str,
session_id: str = “default-session”,
user_id: str = “nameless”
) -> str:
“”“
Make a traced LLM name. Each name creates a Langfuse hint with:
– Full enter and output
– Token utilization (enter, output, complete)
– Calculated value in USD
– Latency in milliseconds
– Mannequin used and session context
Parameters:
user_message : The message from the person
session_id : Teams associated calls into one dialog in Langfuse
user_id : Associates the decision with a selected person for analytics
Returns:
The LLM response as a string
““”
# Create a top-level hint for this person interplay
# The hint seems within the Langfuse dashboard as one unit of labor
hint = langfuse_client.hint(
title=“customer-support-call”,
session_id=session_id,
user_id=user_id,
enter={“user_message”: user_message, “system_prompt”: SYSTEM_PROMPT}
)
# Create a era span contained in the hint
# This captures model-specific particulars: mannequin title, tokens, value
era = hint.era(
title=“claude-completion”,
mannequin=MODEL,
enter={
“system”: SYSTEM_PROMPT,
“messages”: [{“role”: “user”, “content”: user_message}]
}
)
start_time = time.time()
attempt:
# Make the API name
response = anthropic_client.messages.create(
mannequin=MODEL,
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=[{“role”: “user”, “content”: user_message}]
)
latency_ms = int((time.time() – start_time) * 1000)
# Extract the response textual content
response_text = response.content material[0].textual content
# Extract token utilization from the response
input_tokens = response.utilization.input_tokens
output_tokens = response.utilization.output_tokens
total_tokens = input_tokens + output_tokens
# Calculate value for this name
cost_usd = (
input_tokens * COST_PER_INPUT_TOKEN +
output_tokens * COST_PER_OUTPUT_TOKEN
)
# Replace the era span with outcomes
# This information populates the Langfuse value and token dashboards
era.finish(
output=response_text,
utilization={
“enter”: input_tokens,
“output”: output_tokens,
“complete”: total_tokens,
“unit”: “TOKENS”
},
metadata={
“latency_ms”: latency_ms,
“cost_usd”: spherical(cost_usd, 6),
“mannequin”: MODEL
}
)
# Replace the hint with the ultimate output
hint.replace(
output={“response”: response_text},
metadata={“total_cost_usd”: spherical(cost_usd, 6)}
)
# Print a abstract to stdout for native visibility
print(f“n{‘─’ * 60}”)
print(f“Consumer: {user_message}”)
print(f“Claude: {response_text}”)
print(f“Tokens: {input_tokens} in / {output_tokens} out / {total_tokens} complete”)
print(f“Price: ${cost_usd:.6f}”)
print(f“Latency: {latency_ms}ms”)
print(f“Hint: {langfuse_client.base_url}/hint/{hint.id}”)
print(f“{‘─’ * 60}n”)
return response_text
besides Exception as e:
# File the error within the hint so it exhibits up in Langfuse
era.finish(
output=None,
metadata={“error”: str(e), “latency_ms”: int((time.time() – start_time) * 1000)}
)
hint.replace(output={“error”: str(e)})
# At all times flush earlier than elevating — ensures the error hint is distributed
langfuse_client.flush()
elevate
lastly:
# Flush sends all buffered occasions to Langfuse
# In a long-running service, Langfuse flushes mechanically.
# In a script, it’s essential to flush manually earlier than the method exits.
langfuse_client.flush()
# ── Run an indication ────────────────────────────────────────────────────────
if __name__ == “__main__”:
# Simulate two turns of a buyer help dialog
test_messages = [
“What is your return policy for electronics?”,
“Can I return an item I bought 45 days ago?”
]
session = “demo-session-001”
for i, message in enumerate(test_messages):
print(f“nCall {i + 1}/{len(test_messages)}”)
attempt:
call_llm_with_tracing(
user_message=message,
session_id=session,
user_id=“test-user-42”
)
besides Exception as e:
print(f“Error on name {i + 1}: {e}”)
![How To Grow to be An AI Search Authority In search engine optimization [Webinar]](https://blog.aimactgrow.com/wp-content/uploads/2026/04/1-812-120x86.png)
