“”“ Continuous batching = iteration-level scheduling + ragged (packed) batching.   Two approaches are compared (both run BATCH_SIZE sequences concurrently, so the comparison is slot-for-slot fair):     1. Static batching (baseline):        Prompts are processed BATCH_SIZE at a time.  Each wave is padded to a        common length and run together until the LONGEST request in that wave        finishes; a hard “batch barrier” then has to clear before the next wave        starts.  Short requests sit idle behind the barrier.     2. Continuous batching (production-aligned):        Two ideas combine to keep the GPU busy.          (a) Iteration-level scheduling: the moment a sequence finishes it frees            its slot, and the next queued prompt is admitted on the SAME step –            no waiting for the rest of the batch.          (b) Ragged / packed batching – the part that makes it truly “continuous“:            instead of padding every sequence into a rectangular [B, max_len]            tensor, ALL in-flight tokens are concatenated into a single unpadded            [1, total_tokens] row and run in ONE forward pass.  A block-diagonal            causal attention mask stops tokens from attending across sequence            boundaries, so packing is mathematically identical to running each            sequence on its own (verified: greedy output matches per-prompt            generation token-for-token).          Because attention is governed entirely by the mask, a newly admitted        prompt’s multi-token PREFILL rides along in the same forward pass as        every other sequence’s single-token DECODE step.  Prefill and decode are        fused: no padding, no separate prefill pass.          KV cache: each sequence keeps its own DynamicCache; every step the caches        are concatenated along the time axis into one packed cache, and the newly        computed KV is scattered back per sequence.  (Real engines store the        cache in fixed-size pages – “paged attention” – to avoid this per-step        reassembly, but the attention/masking logic is exactly what you see here.) ““”   import time   import torch from dataclasses import dataclass, field from typing import Optional from transformers import AutoTokenizer, AutoModelForCausalLM, DynamicCache from transformers.cache_utils import DynamicLayer   MODEL_ID = “openai-community/gpt2”   # swap for any causal LM BATCH_SIZE = 3                       # max concurrent sequences (slots)     def _device_sync(model) -> None:     “”“Block until queued GPU work finishes, so timings are accurate.”“”     if model.device.type == “cuda”:         torch.cuda.synchronize()     elif model.device.type == “mps”:         torch.mps.synchronize()     def static_batching(requests: list[tuple[str, int]], tokenizer, model) -> list[str]:     “”“Baseline. Process requests BATCH_SIZE at a time; each wave runs together     until its LONGEST request finishes, then a batch barrier clears before the     next wave starts.       Downside: short requests in a wave idle until the wave’s longest is done –     and no slot can be refilled until the whole wave clears the barrier.     ““”     if not requests:         return []       tokenizer.padding_side = “left”     results: dict[int, str] = {}     indexed = list(enumerate(requests))            # (req_id, (prompt, cap))       for wave_start in range(0, len(indexed), BATCH_SIZE):         wave = indexed[wave_start: wave_start + BATCH_SIZE]         wave_max = max(cap for _, (_, cap) in wave)           # Show which request occupies each slot in this wave.         for slot, (req_id, (prompt, cap)) in enumerate(wave):             print(f”  ++ slot {slot} <- req {req_id} ({cap} tok cap): {prompt!r}”, flush=True)           prompts = [p for _, (p, _) in wave]         inputs = tokenizer(             prompts, return_tensors=“pt”, padding=True, truncation=True         ).to(model.device)         with torch.no_grad():             output_ids = model.generate(                 **inputs,                 max_new_tokens=wave_max,           # whole wave decodes to the longest                 pad_token_id=tokenizer.eos_token_id,                 do_sample=False,             )           width = inputs.input_ids.shape[1]         print(             f”  *** batch barrier: all {len(wave)} slots wait for the longest “             f“({wave_max} tokens) ***”,             flush=True,         )         for slot, ((req_id, (prompt, cap)), row) in enumerate(zip(wave, output_ids)):             text = prompt + tokenizer.decode(row[width:width + cap], skip_special_tokens=True)             results[req_id] = text             print(                 f”  — slot {slot} done  req {req_id} ({cap}/{wave_max} tokens): {text[:90]}”,                 flush=True,             )       return [results[k] for k in sorted(results)]     @dataclass class Sequence:     “”“State for a single in-flight sequence.”“”     req_id: int                 # original request index (for ordering results)     prompt: str     max_new_tokens: int         # per-request cap so short requests finish early     # Tokens to feed on the NEXT step: the whole prompt right after admission     # (prefill), then a single token per step (decode).     pending_ids: list[int]     # Per-sequence KV-cache; None until this sequence has run once.     kv_cache: Optional[DynamicCache] = None     kv_len: int = 0             # number of cached tokens (prompt + generated)     tokens_generated: int = 0     output_ids: list[int] = field(default_factory=list)     def _make_cache(layers_kv: list[tuple[torch.Tensor, torch.Tensor]]) -> DynamicCache:     “”“Build a DynamicCache from explicit per-layer (keys, values) tensors.       We SET the tensors directly instead of calling DynamicLayer.update() (which     would append), because we are assembling caches from scratch each step.     ““”     cache = DynamicCache()     for k, v in layers_kv:         layer = DynamicLayer()         layer.lazy_initialization(k, v)         layer.keys = k         layer.values = v         cache.layers.append(layer)     return cache     def _ragged_step(seqs: list[Sequence], model, device, dtype) -> list[int]:     “”“Run ONE packed forward pass over every active sequence.       All sequences are flattened into a single row (batch dim = 1):           input_ids      [1, total_q]            – every sequence’s pending tokens         position_ids   [1, total_q]            – each token’s position in ITS sequence         attention_mask [1, 1, total_q, total_kv + total_q]  – block-diagonal causal         past_key_values  packed cache [1, H, total_kv, D]       total_q  = sum of pending tokens (1 per decoding seq, prompt_len per new seq)     total_kv = sum of already-cached tokens across sequences       Returns the next greedy token for each sequence (same order as “seqs“).     ““”     q_lens = [len(s.pending_ids) for s in seqs]     total_q = sum(q_lens)     total_kv = sum(s.kv_len for s in seqs)       # Packed inputs: concatenate every sequence’s pending tokens into one row.     flat_ids = [t for s in seqs for t in s.pending_ids]     input_ids = torch.tensor([flat_ids], dtype=torch.long, device=device)       # Tag every KEY and every QUERY token with (sequence index, position-in-sequence).     # Key space is laid out as [ cached tokens | this step’s new tokens ], matching     # how the model appends new KV to the end of the packed cache.     key_seq, key_pos = [], []     for si, s in enumerate(seqs):                 # cached block         for p in range(s.kv_len):             key_seq.append(si)             key_pos.append(p)     q_seq, q_pos = [], []     for si, s in enumerate(seqs):                 # new block (also queries)         for j in range(len(s.pending_ids)):             pos = s.kv_len

# llm_with_tracing.py # Purpose: A production-ready LLM call wrapper with full observability. # Every call is traced in Langfuse: input, output, tokens, cost, latency. # # Prerequisites: #   pip install 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 below # # Run: #   python llm_with_tracing.py   import os import time from dotenv import load_dotenv import anthropic from langfuse import Langfuse   # Load environment variables from .env file load_dotenv()   # Required environment 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 clients langfuse_client = Langfuse()          # Reads keys automatically from environment anthropic_client = anthropic.Anthropic()  # Reads ANTHROPIC_API_KEY from environment   # ── Configuration ───────────────────────────────────────────────────────────── # Store your prompt here, not inline in the API call. # This makes it versionable and testable independently. SYSTEM_PROMPT = “”“You are a helpful customer support assistant. Answer questions clearly and concisely. If you do not know something, say so directly — do not guess.”“”   MODEL = “claude-sonnet-4-20250514”   # Anthropic’s pricing as of mid-2026 (update when pricing changes) # Used to calculate cost per call for cost tracking COST_PER_INPUT_TOKEN  = 3.00 / 1_000_000   # $3.00 per million input 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 = “anonymous” ) -> str:     “”“     Make a traced LLM call. Every call creates a Langfuse trace with:     – Full input and output     – Token usage (input, output, total)     – Calculated cost in USD     – Latency in milliseconds     – Model used and session context       Parameters:         user_message : The message from the user         session_id   : Groups related calls into one conversation in Langfuse         user_id      : Associates the call with a specific user for analytics       Returns:         The LLM response as a string     ““”       # Create a top-level trace for this user interaction     # The trace appears in the Langfuse dashboard as one unit of work     trace = langfuse_client.trace(         name=“customer-support-call”,         session_id=session_id,         user_id=user_id,         input={“user_message”: user_message, “system_prompt”: SYSTEM_PROMPT}     )       # Create a generation span inside the trace     # This captures model-specific details: model name, tokens, cost     generation = trace.generation(         name=“claude-completion”,         model=MODEL,         input={             “system”: SYSTEM_PROMPT,             “messages”: [{“role”: “user”, “content”: user_message}]         }     )       start_time = time.time()       try:         # Make the API call         response = anthropic_client.messages.create(             model=MODEL,             max_tokens=1024,             system=SYSTEM_PROMPT,             messages=[{“role”: “user”, “content”: user_message}]         )           latency_ms = int((time.time() – start_time) * 1000)           # Extract the response text         response_text = response.content[0].text           # Extract token usage from the response         input_tokens  = response.usage.input_tokens         output_tokens = response.usage.output_tokens         total_tokens  = input_tokens + output_tokens           # Calculate cost for this call         cost_usd = (             input_tokens  * COST_PER_INPUT_TOKEN +             output_tokens * COST_PER_OUTPUT_TOKEN         )           # Update the generation span with results         # This data populates the Langfuse cost and token dashboards         generation.end(             output=response_text,             usage={                 “input”:  input_tokens,                 “output”: output_tokens,                 “total”:  total_tokens,                 “unit”:   “TOKENS”             },             metadata={                 “latency_ms”: latency_ms,                 “cost_usd”:   round(cost_usd, 6),                 “model”:      MODEL             }         )           # Update the trace with the final output         trace.update(             output={“response”: response_text},             metadata={“total_cost_usd”: round(cost_usd, 6)}         )           # Print a summary to stdout for local visibility         print(f“\n{‘─’ * 60}”)         print(f“User:    {user_message}”)         print(f“Claude:  {response_text}”)         print(f“Tokens:  {input_tokens} in / {output_tokens} out / {total_tokens} total”)         print(f“Cost:    ${cost_usd:.6f}”)         print(f“Latency: {latency_ms}ms”)         print(f“Trace:   {langfuse_client.base_url}/trace/{trace.id}”)         print(f“{‘─’ * 60}\n”)           return response_text       except Exception as e:         # Record the error in the trace so it shows up in Langfuse         generation.end(             output=None,             metadata={“error”: str(e), “latency_ms”: int((time.time() – start_time) * 1000)}         )         trace.update(output={“error”: str(e)})           # Always flush before raising — ensures the error trace is sent         langfuse_client.flush()         raise       finally:         # Flush sends all buffered events to Langfuse         # In a long-running service, Langfuse flushes automatically.         # In a script, you must flush manually before the process exits.         langfuse_client.flush()     # ── Run a demonstration ──────────────────────────────────────────────────────── if __name__ == “__main__”:     # Simulate two turns of a customer support conversation     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)}”)         try:             call_llm_with_tracing(                 user_message=message,                 session_id=session,                 user_id=“test-user-42”             )         except Exception as e:             print(f“Error on call {i + 1}: {e}”)

# research_agent.py # Purpose: A research agent with full AgentOps instrumentation. # Every session is logged, replayed, and cost-tracked in the AgentOps dashboard. # # Prerequisites: #   pip install agentops anthropic python-dotenv # # Environment variables required (in .env): #   AGENTOPS_API_KEY — from https://app.agentops.ai #   ANTHROPIC_API_KEY — from https://console.anthropic.com # # How to run: #   python research_agent.py   import os import json import time from dotenv import load_dotenv import anthropic import agentops from agentops.sdk.decorators import record_function   load_dotenv()   # ── Initialize AgentOps ──────────────────────────────────────────────────────── # This must be called before any agent code runs. # Tags let you filter and group sessions in the dashboard. # The SDK automatically intercepts LLM calls once initialized. agentops.init(     api_key=os.environ[“AGENTOPS_API_KEY”],     tags=[“research-agent”, “production”, “v1.0”],     auto_start_session=True       # Automatically starts a session on init )   # Initialize the Anthropic client after AgentOps — the SDK wraps LLM clients # to automatically capture every call’s input, output, tokens, and cost. client = anthropic.Anthropic(api_key=os.environ[“ANTHROPIC_API_KEY”])   MODEL = “claude-sonnet-4-20250514”   # ── System prompt ───────────────────────────────────────────────────────────── # Stored as a constant, not inline — version-controllable and testable. SYSTEM_PROMPT = “”“You are a research assistant. When given a topic: 1. Use the available tools to gather information systematically 2. Call search_topic to get an overview of the subject 3. Call get_key_facts to extract the most important points 4. Call format_summary to structure the final output   Be thorough but concise. Always call format_summary as your final step.”“”   # ── Tool definitions ────────────────────────────────────────────────────────── # These are the tools the agent can call. In a real system, search_topic # would call a real search API (Tavily, SerpAPI, etc.). Here they are stubs # that return realistic data so you can run the example without external APIs. TOOLS = [     {         “name”: “search_topic”,         “description”: (             “Search for comprehensive information about a topic. “             “Returns an overview with key themes and context. “             “Use this as the first step for any research task.”         ),         “input_schema”: {             “type”: “object”,             “properties”: {                 “topic”: {                     “type”: “string”,                     “description”: “The topic to research. Be specific.”                 },                 “depth”: {                     “type”: “string”,                     “enum”: [“overview”, “detailed”],                     “description”: “How deep to search. Use ‘overview’ first.”                 }             },             “required”: [“topic”]         }     },     {         “name”: “get_key_facts”,         “description”: (             “Extract the most important facts about a topic from search results. “             “Use after search_topic to identify the 5-7 most significant points.”         ),         “input_schema”: {             “type”: “object”,             “properties”: {                 “topic”: {                     “type”: “string”,                     “description”: “The topic to extract facts about”                 },                 “focus”: {                     “type”: “string”,                     “description”: “Optional: specific angle to focus on (e.g., ‘recent developments’, ‘key players’)”                 }             },             “required”: [“topic”]         }     },     {         “name”: “format_summary”,         “description”: (             “Format research findings into a clean structured summary. “             “Always call this as the final step before returning to the user.”         ),         “input_schema”: {             “type”: “object”,             “properties”: {                 “title”: {                     “type”: “string”,                     “description”: “Title for the summary”                 },                 “key_points”: {                     “type”: “array”,                     “items”: {“type”: “string”},                     “description”: “List of key findings (5-7 items)”                 },                 “conclusion”: {                     “type”: “string”,                     “description”: “A 2-3 sentence synthesis of the research”                 }             },             “required”: [“title”, “key_points”, “conclusion”]         }     } ]     # ── Tool implementations ────────────────────────────────────────────────────── # @record_function decorates each tool so AgentOps captures: # – The function name # – Input arguments # – Return value # – Execution time # – Any exceptions # These appear as labeled spans in the session replay timeline.   @record_function(“search_topic”) def search_topic(topic: str, depth: str = “overview”) -> dict:     “”“     Search for information about a topic.     In production: replace this stub with a real search API call.     ““”     # Simulate search latency — remove in production     time.sleep(0.3)       # Stub response — replace with: tavily_client.search(query=topic)     return {         “topic”: topic,         “depth”: depth,         “results”: f“Comprehensive overview of {topic}: This is a rapidly evolving field “                    f“with significant developments in 2025-2026. Key themes include “                    f“technical innovation, adoption patterns, and organizational impact. “                    f“Multiple research groups and companies are actively advancing the field.”,         “source_count”: 12,         “timestamp”: “2026-05-26”     }     @record_function(“get_key_facts”) def get_key_facts(topic: str, focus: str = None) -> dict:     “”“     Extract key facts about a topic.     In production: this would process real search results.     ““”     time.sleep(0.2)       focus_note = f” (focus: {focus})” if focus else “”     return {         “topic”: topic,         “focus”: focus_note,         “facts”: [             f“{topic} has seen 42% year-over-year growth in adoption”,             f“Leading organizations report 3-5x productivity improvements”,             f“Key technical challenges include reliability, cost, and governance”,             f“The market is projected to reach $4.9B by 2028”,             f“Open-source tooling has matured significantly in the past 18 months”,         ],         “confidence”: “high”     }     @record_function(“format_summary”) def format_summary(title: str, key_points: list, conclusion: str) -> dict:     “”“     Format research into a structured summary.     This is always the final step in the research workflow.     ““”     return {         “title”: title,         “key_points”: key_points,         “conclusion”: conclusion,         “format”: “structured_summary”,         “generated_at”: “2026-05-26”     }     def execute_tool(tool_name: str, tool_input: dict) -> str:     “”“     Route tool calls to the correct implementation.     Returns the result as a JSON string for the model to read.     ““”     if tool_name == “search_topic”:         result = search_topic(**tool_input)     elif tool_name == “get_key_facts”:         result = get_key_facts(**tool_input)     elif tool_name == “format_summary”:         result = format_summary(**tool_input)     else:         result = {“error”: f“Unknown tool: {tool_name}”}       return json.dumps(result)     # ── The agent loop ───────────────────────────────────────────────────────────── def run_research_agent(topic: str) -> dict:     “”“     Run the research agent on a given topic.       The loop:     1. Send the goal to Claude with the available tools     2. If Claude wants to call a tool, execute it and return the result     3. Continue until Claude signals it is done (stop_reason == ‘end_turn’)     4. Return the final structured summary       AgentOps captures every iteration automatically because:     – The LLM client is wrapped after agentops.init()     – Each tool is decorated with @record_function     – The session spans the full lifecycle from init to end_session()     ““”     print(f“\nStarting research agent for topic: ‘{topic}’”)     print(“Session will be visible at https://app.agentops.ai\n”)       messages = [         {“role”: “user”, “content”: f“Research this topic and produce a structured summary: {topic}”}     ]       final_summary = None     iteration = 0     max_iterations = 10  # Safety limit — prevents runaway loops       while iteration < max_iterations: