Lab Guide

Format: Python script (lab.py)  ·  Time estimate: 45–60 minutes  ·  Prerequisites: Python 3.10+, anthropic, langfuse, matplotlib, python-dotenv

Setup

Install the required packages:

pip install anthropic langfuse matplotlib python-dotenv numpy

Create or update your .env file with the following keys:

ANTHROPIC_API_KEY=your_anthropic_key
LANGFUSE_PUBLIC_KEY=your_langfuse_public_key
LANGFUSE_SECRET_KEY=your_langfuse_secret_key
LANGFUSE_HOST=https://cloud.langfuse.com  # or your self-hosted URL

Exercise 1: Structured Logging Wrapper

Goal

Build a reusable wrapper function that logs structured JSON for every Claude API call, capturing request ID, model, token counts, latency, cost, and status.

Steps

1. Create a LLMLogger class that wraps the Anthropic client:

   import time, uuid, json, logging
   import anthropic
   from dotenv import load_dotenv
   
   load_dotenv()
   logger = logging.getLogger("llm.calls")
   logging.basicConfig(level=logging.INFO)
   
   # Model pricing per million tokens
   PRICING = {
       "claude-sonnet-4-20250514": {"input": 3.00, "output": 15.00},
       "claude-haiku-4-20250414":  {"input": 0.80, "output": 4.00},
   }
   
   class LLMLogger:
       def __init__(self):
           self.client = anthropic.Anthropic()
           self.logs = []  # in-memory store for this lab
   
       def call(self, messages, model="claude-sonnet-4-20250514", **kwargs):
           # TODO: implement structured logging wrapper
           pass

2. Inside the call method:
   • Generate a request_id using uuid.uuid4()
   • Record start_time using time.perf_counter()
   • Make the API call inside a try/except block
   • On success: extract input_tokens, output_tokens from response.usage, calculate latency_ms and cost_usd
   • On error: log the error type and message
   • Append the structured dict to self.logs and emit via logger.info()
3. Test by making 5 API calls with different prompts. Print the logs list formatted as JSON.
4. Add a summary() method that prints total calls, total tokens, total cost, and average latency.

Expected Output

{"request_id": "a1b2c3d4-...", "model": "claude-sonnet-4-20250514", "input_tokens": 42, "output_tokens": 156, "latency_ms": 1823, "cost_usd": 0.002466, "status": "success"}
{"request_id": "e5f6g7h8-...", "model": "claude-sonnet-4-20250514", "input_tokens": 38, "output_tokens": 89, "latency_ms": 1204, "cost_usd": 0.001449, "status": "success"}
...

=== Summary ===
Total calls:    5
Total tokens:   1,247 (input: 203, output: 1,044)
Total cost:     $0.0168
Avg latency:    1,542 ms

Exercise 2: Langfuse Tracing Integration

Goal

Instrument a multi-step pipeline with Langfuse tracing so that retrieval, LLM calls, and post-processing appear as linked spans in a single trace.

Steps

1. Initialise the Langfuse client:

   from langfuse import Langfuse
   
   langfuse = Langfuse()

2. Simulate a RAG pipeline with three steps:
   • retrieve: simulate document retrieval (use time.sleep(0.2) to mimic latency, return hardcoded context)
   • generate: call Claude with the retrieved context + user question
   • validate: check the response is non-empty and under a token limit
3. Wrap each step in a Langfuse span:

   def rag_pipeline(question: str):
       trace = langfuse.trace(name="rag-pipeline", input=question)
   
       # Step 1: Retrieve
       retrieval_span = trace.span(name="retrieve-docs")
       context = retrieve_documents(question)
       retrieval_span.end(output={"doc_count": len(context)})
   
       # Step 2: Generate
       generation = trace.generation(
           name="llm-generate",
           model="claude-sonnet-4-20250514",
           input=[
               {"role": "system", "content": f"Context: {context}"},
               {"role": "user", "content": question},
           ],
       )
       response = client.messages.create(...)
       generation.end(
           output=response.content[0].text,
           usage={"input": response.usage.input_tokens,
                  "output": response.usage.output_tokens},
       )
   
       # Step 3: Validate
       validation_span = trace.span(name="validate-response")
       is_valid = validate(response)
       validation_span.end(output={"valid": is_valid})
   
       trace.update(output=response.content[0].text)
       return response.content[0].text

4. Run the pipeline 3 times with different questions.
5. Open the Langfuse web UI and explore the traces. Verify you can see the three spans for each trace and their timing.

Expected Langfuse Dashboard View

Trace: rag-pipeline (2.4s total)
  |-- retrieve-docs     [0ms - 210ms]    (span)
  |-- llm-generate      [215ms - 1980ms] (generation, 847 tokens)
  |-- validate-response [1985ms - 1990ms] (span, valid: true)

Exercise 3: Metrics Dashboard

Goal

Compute latency percentiles (p50, p95, p99), error rate, and cumulative cost from your logs, then visualise them with matplotlib.

Steps

1. Generate synthetic log data (or use your real logs from Exercises 1 and 2). Create at least 50 log entries with realistic variation:

   import random, numpy as np
   
   def generate_synthetic_logs(n=100):
       logs = []
       for i in range(n):
           is_error = random.random() < 0.05  # 5% error rate
           latency = np.random.lognormal(mean=7.0, sigma=0.5)  # ~1100ms median
           input_tok = random.randint(30, 500)
           output_tok = 0 if is_error else random.randint(50, 800)
           logs.append({
               "request_id": str(uuid.uuid4()),
               "latency_ms": round(latency) if not is_error else round(latency * 2),
               "input_tokens": input_tok,
               "output_tokens": output_tok,
               "status": "error" if is_error else "success",
               "cost_usd": (input_tok * 3.0 + output_tok * 15.0) / 1_000_000,
           })
       return logs

2. Compute the following metrics:
   • Latency p50, p95, p99 (use numpy.percentile)
   • Error rate as a percentage
   • Total cost and average cost per successful request
   • Total token throughput (input + output)
3. Create a matplotlib figure with 4 subplots:
   • Top-left: Latency histogram with p50/p95/p99 vertical lines
   • Top-right: Cumulative cost over time (request index as x-axis)
   • Bottom-left: Token usage scatter plot (input vs output tokens)
   • Bottom-right: Status bar chart (success vs error counts)
4. Add a detect_anomalies() function that flags any request where latency exceeds 2x the p95 value. Print the flagged request IDs.

Expected Output

=== Metrics Summary ===
Latency p50:    1,087 ms
Latency p95:    2,341 ms
Latency p99:    3,812 ms
Error rate:     5.0%
Total cost:     $0.4721
Avg cost/req:   $0.00497
Total tokens:   47,832

=== Anomalies (latency > 2x p95) ===
  request a1b2c3d4: 5,102 ms (2.18x p95)
  request e5f6g7h8: 4,891 ms (2.09x p95)
Saved dashboard to observability_dashboard.png

Finishing Up

Once you have completed all three exercises:

1. Run the full script end-to-end: python lab.py
2. Verify that structured logs are printed to the console.
3. Check the Langfuse dashboard for your traces (if using Langfuse).
4. Confirm the dashboard PNG was saved with all four subplots.
5. Compare your implementation with lab_answer.py.

Self-Check

• ✓ Exercise 1: Your LLMLogger captures all 6 required fields (request_id, model, input_tokens, output_tokens, latency_ms, status).
• ✓ Exercise 1: Cost is calculated correctly using per-million-token pricing.
• ✓ Exercise 2: Each RAG pipeline run produces a trace with 3 spans in Langfuse (or your mock logger).
• ✓ Exercise 2: Token usage is recorded on the generation span.
• ✓ Exercise 3: Latency percentiles are computed using numpy.percentile, not manual sorting.
• ✓ Exercise 3: The dashboard has 4 subplots with labelled axes and the anomaly detector flags outliers.