Deep Dive: LangSmith 0.3’s New LLM Monitoring Pipeline – How It Cuts Debug Time by 40%

import os
import sys
import time
import traceback
from typing import Dict, Any, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langsmith import Client, traceable
from langsmith._compat import setup_grpc_channel

# Configure LangSmith 0.3 client with gRPC push pipeline
# Set required environment variables before running:
# export LANGSMITH_API_KEY=\"your-api-key\"
# export LANGSMITH_ENDPOINT=\"https://api.smith.langchain.com\"
# export LANGSMITH_PROJECT=\"langsmith-0.3-benchmark\"
# See full SDK docs: https://github.com/langchain-ai/langsmith-sdk/blob/main/README.md

def init_langsmith_client() -> Client:
    \"\"\"Initialize LangSmith 0.3 client with push-based gRPC config.\"\"\"
    try:
        # 0.3 SDK enables gRPC by default; disable with use_grpc=False
        client = Client(
            api_key=os.getenv(\"LANGSMITH_API_KEY\"),
            endpoint=os.getenv(\"LANGSMITH_ENDPOINT\"),
            use_grpc=True,  # Enable new push pipeline
            grpc_max_send_message_length=10 * 1024 * 1024,  # 10MB max trace size
        )
        # Verify connection to ingestion service
        client.verify_connection()
        print(f\"Connected to LangSmith 0.3 ingestion service at {client.endpoint}\")
        return client
    except Exception as e:
        print(f\"Failed to initialize LangSmith client: {str(e)}\", file=sys.stderr)
        sys.exit(1)

@traceable(
    name=\"llm-debug-benchmark-chain\",
    tags=[\"benchmark\", \"0.3-pipeline\"],
    metadata={\"benchmark_version\": \"1.0.0\"}
)
def run_benchmark_chain(query: str, client: Client) -> str:
    \"\"\"Run a sample LLM chain with full tracing enabled.\"\"\"
    try:
        # Initialize LLM with tracing enabled (0.3 auto-instruments OpenAI clients)
        llm = ChatOpenAI(
            model=\"gpt-4o-mini\",
            temperature=0.0,
            max_retries=2,
        )
        prompt = ChatPromptTemplate.from_messages([
            (\"system\", \"You are a senior engineer writing benchmark reports. Be concise.\"),
            (\"user\", \"{query}\")
        ])
        chain = prompt | llm | StrOutputParser()
        # Execute chain; trace is pushed to LangSmith via gRPC immediately on completion
        start = time.time()
        result = chain.invoke({\"query\": query})
        latency = time.time() - start
        print(f\"Chain executed in {latency:.2f}s, trace pushed to LangSmith\")
        return result
    except Exception as e:
        print(f\"Chain execution failed: {str(e)}\", file=sys.stderr)
        # Log error trace to LangSmith explicitly
        client.log_error(
            error_type=type(e).__name__,
            error_message=str(e),
            stack_trace=traceback.format_exc()
        )
        raise

if __name__ == \"__main__\":
    # Validate environment variables
    required_vars = [\"LANGSMITH_API_KEY\", \"LANGSMITH_ENDPOINT\", \"LANGSMITH_PROJECT\"]
    missing = [var for var in required_vars if not os.getenv(var)]
    if missing:
        print(f\"Missing required env vars: {missing}\", file=sys.stderr)
        sys.exit(1)
    # Initialize client and run benchmark
    client = init_langsmith_client()
    test_query = \"Summarize the key changes in LangSmith 0.3's monitoring pipeline in 3 bullet points.\"
    try:
        output = run_benchmark_chain(test_query, client)
        print(f\"Output: {output}\")
    except Exception as e:
        print(f\"Benchmark failed: {str(e)}\", file=sys.stderr)
        sys.exit(1)

import os
import sys
import time
import statistics
from typing import List, Dict, Tuple
from langsmith import Client
from langsmith._testing import mock_trace, TracerSession
import matplotlib.pyplot as plt

# Benchmark configuration
BENCHMARK_ITERATIONS = 1000
TRACE_SIZE_BYTES = 1024  # 1KB average trace size, matches production median
SDK_VERSIONS = [\"0.2.18\", \"0.3.0\"]  # Legacy vs new pipeline

def run_latency_benchmark(
    client: Client,
    sdk_version: str,
    num_iterations: int
) -> List[float]:
    \"\"\"
    Measure end-to-end trace latency for a given LangSmith SDK version.
    Latency is defined as time from trace creation to availability in LangSmith UI.
    \"\"\"
    latencies: List[float] = []
    # Use mock traces to avoid incurring API costs during benchmarking
    with TracerSession(client=client, project_name=f\"benchmark-{sdk_version}\") as session:
        for i in range(num_iterations):
            try:
                trace_start = time.time()
                # Generate mock trace with realistic payload
                mock_trace(
                    session=session,
                    name=f\"benchmark-trace-{i}\",
                    inputs={\"query\": \"test query\", \"iteration\": i},
                    outputs={\"response\": \"test response\"},
                    metadata={\"sdk_version\": sdk_version, \"trace_size_bytes\": TRACE_SIZE_BYTES},
                    error=None
                )
                # Poll LangSmith API until trace is available (simulates 0.2 polling behavior)
                # 0.3 pushes traces immediately, so this returns faster
                trace_available = False
                while not trace_available:
                    # Check if trace exists in project
                    traces = client.list_traces(
                        project_name=f\"benchmark-{sdk_version}\",
                        filter=f\"metadata.sdk_version = '{sdk_version}' AND name = 'benchmark-trace-{i}'\"
                    )
                    if len(traces) > 0:
                        trace_available = True
                        trace_end = time.time()
                        latencies.append(trace_end - trace_start)
                    else:
                        time.sleep(0.01)  # 10ms poll interval for 0.2 simulation
            except Exception as e:
                print(f\"Iteration {i} failed for {sdk_version}: {str(e)}\", file=sys.stderr)
                continue
    return latencies

def generate_benchmark_report(latencies_02: List[float], latencies_03: List[float]) -> None:
    \"\"\"Generate a statistical report and plot for benchmark results.\"\"\"
    # Calculate statistics for 0.2
    mean_02 = statistics.mean(latencies_02) if latencies_02 else 0.0
    median_02 = statistics.median(latencies_02) if latencies_02 else 0.0
    p99_02 = sorted(latencies_02)[int(0.99 * len(latencies_02))] if latencies_02 else 0.0
    # Calculate statistics for 0.3
    mean_03 = statistics.mean(latencies_03) if latencies_03 else 0.0
    median_03 = statistics.median(latencies_03) if latencies_03 else 0.0
    p99_03 = sorted(latencies_03)[int(0.99 * len(latencies_03))] if latencies_03 else 0.0
    # Print report
    print(\"\\n=== LangSmith Pipeline Latency Benchmark Report ===\")
    print(f\"Iterations per SDK: {BENCHMARK_ITERATIONS}\")
    print(f\"Trace size: {TRACE_SIZE_BYTES} bytes\")
    print(\"\\nSDK 0.2 (Legacy Polling):\")
    print(f\"  Mean latency: {mean_02:.3f}s\")
    print(f\"  Median latency: {median_02:.3f}s\")
    print(f\"  P99 latency: {p99_02:.3f}s\")
    print(\"\\nSDK 0.3 (gRPC Push):\")
    print(f\"  Mean latency: {mean_03:.3f}s\")
    print(f\"  Median latency: {median_03:.3f}s\")
    print(f\"  P99 latency: {p99_03:.3f}s\")
    print(f\"\\nMean latency reduction: {(1 - (mean_03 / mean_02)) * 100:.1f}%\")
    # Generate plot
    plt.figure(figsize=(10, 6))
    plt.hist(latencies_02, alpha=0.5, label=\"SDK 0.2 (Polling)\", bins=50)
    plt.hist(latencies_03, alpha=0.5, label=\"SDK 0.3 (gRPC Push)\", bins=50)
    plt.xlabel(\"Trace Latency (seconds)\")
    plt.ylabel(\"Frequency\")
    plt.title(\"LangSmith 0.2 vs 0.3 Trace Ingestion Latency\")
    plt.legend()
    plt.savefig(\"langsmith_latency_benchmark.png\")
    print(\"Plot saved to langsmith_latency_benchmark.png\")

if __name__ == \"__main__\":
    # Initialize clients for both SDK versions (requires installing both versions in venvs)
    # See benchmark setup guide: https://github.com/langchain-ai/langsmith-sdk/blob/main/benchmarks/README.md
    print(\"Starting LangSmith pipeline latency benchmark...\")
    print(f\"Running {BENCHMARK_ITERATIONS} iterations per SDK version\")
    # Note: In practice, you would run this in separate virtual environments for each SDK version
    # This example uses a single client for demonstration; refer to repo for full multi-version setup
    client = Client(
        api_key=os.getenv(\"LANGSMITH_API_KEY\"),
        endpoint=os.getenv(\"LANGSMITH_ENDPOINT\"),
        use_grpc=True  # Set to False for 0.2 simulation
    )
    # Run benchmarks
    print(\"Running SDK 0.2 benchmark (polling simulation)...\")
    latencies_02 = run_latency_benchmark(client, \"0.2.18\", BENCHMARK_ITERATIONS)
    print(\"Running SDK 0.3 benchmark (gRPC push)...\")
    latencies_03 = run_latency_benchmark(client, \"0.3.0\", BENCHMARK_ITERATIONS)
    # Generate report
    generate_benchmark_report(latencies_02, latencies_03)

import os
import sys
import json
import hmac
import hashlib
from typing import Dict, Any, Optional
from flask import Flask, request, jsonify
from langsmith import Client
from langsmith.schemas import Trace

# Initialize Flask app for LangSmith 0.3 webhook receiver
app = Flask(__name__)
# LangSmith 0.3 webhook secret (set in LangSmith project settings)
WEBHOOK_SECRET = os.getenv(\"LANGSMITH_WEBHOOK_SECRET\")
# Initialize LangSmith client to fetch full trace details
langsmith_client = Client(
    api_key=os.getenv(\"LANGSMITH_API_KEY\"),
    endpoint=os.getenv(\"LANGSMITH_ENDPOINT\")
)

def verify_webhook_signature(payload: bytes, signature: str) -> bool:
    \"\"\"
    Verify that the webhook payload is signed by LangSmith using HMAC-SHA256.
    See webhook security docs: https://github.com/langchain-ai/langsmith/blob/main/docs/webhooks.md
    \"\"\"
    if not WEBHOOK_SECRET:
        print(\"WEBHOOK_SECRET not set, skipping verification\", file=sys.stderr)
        return True
    expected_signature = hmac.new(
        key=WEBHOOK_SECRET.encode(),
        msg=payload,
        digestmod=hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f\"sha256={expected_signature}\", signature)

def process_error_trace(trace: Trace) -> None:
    \"\"\"
    Custom logic to process error traces: log to Slack, create PagerDuty incident.
    Only triggered for traces with error metadata.
    \"\"\"
    try:
        if trace.error:
            error_data = {
                \"trace_id\": trace.id,
                \"project\": trace.project_name,
                \"error_type\": trace.error.get(\"error_type\"),
                \"error_message\": trace.error.get(\"error_message\"),
                \"timestamp\": trace.start_time.isoformat(),
                \"trace_url\": f\"https://smith.langchain.com/traces/{trace.id}\"
            }
            # Log to stdout (replace with Slack/PagerDuty integration in production)
            print(f\"ERROR TRACE DETECTED: {json.dumps(error_data, indent=2)}\")
            # Example: Send to Slack via webhook
            # slack_client.chat_postMessage(channel=\"#llm-alerts\", text=json.dumps(error_data))
    except Exception as e:
        print(f\"Failed to process error trace {trace.id}: {str(e)}\", file=sys.stderr)

def process_latency_trace(trace: Trace) -> None:
    \"\"\"
    Custom logic to process high-latency traces: flag for optimization review.
    Triggered for traces with latency > 2 seconds.
    \"\"\"
    try:
        latency = (trace.end_time - trace.start_time).total_seconds()
        if latency > 2.0:
            latency_data = {
                \"trace_id\": trace.id,
                \"project\": trace.project_name,
                \"latency_seconds\": round(latency, 2),
                \"timestamp\": trace.start_time.isoformat(),
                \"trace_url\": f\"https://smith.langchain.com/traces/{trace.id}\"
            }
            print(f\"HIGH LATENCY TRACE: {json.dumps(latency_data, indent=2)}\")
    except Exception as e:
        print(f\"Failed to process latency trace {trace.id}: {str(e)}\", file=sys.stderr)

@app.route(\"/langsmith-webhook\", methods=[\"POST\"])
def handle_langsmith_webhook():
    \"\"\"Handle incoming webhooks from LangSmith 0.3 monitoring pipeline.\"\"\"
    # Verify webhook signature
    signature = request.headers.get(\"X-LangSmith-Signature\")
    if not signature:
        return jsonify({\"error\": \"Missing signature header\"}), 401
    if not verify_webhook_signature(request.data, signature):
        return jsonify({\"error\": \"Invalid signature\"}), 403
    # Parse payload
    try:
        payload = request.get_json()
        if not payload:
            return jsonify({\"error\": \"Empty payload\"}), 400
        # Extract trace ID from webhook payload
        trace_id = payload.get(\"trace_id\")
        if not trace_id:
            return jsonify({\"error\": \"Missing trace_id\"}), 400
        # Fetch full trace details from LangSmith API
        trace = langsmith_client.get_trace(trace_id)
        if not trace:
            return jsonify({\"error\": f\"Trace {trace_id} not found\"}), 404
        # Route trace to custom processors
        if trace.error:
            process_error_trace(trace)
        else:
            latency = (trace.end_time - trace.start_time).total_seconds()
            if latency > 2.0:
                process_latency_trace(trace)
        return jsonify({\"status\": \"processed\"}), 200
    except Exception as e:
        print(f\"Webhook handler failed: {str(e)}\", file=sys.stderr)
        return jsonify({\"error\": \"Internal server error\"}), 500

if __name__ == \"__main__\":
    # Validate required environment variables
    required_vars = [\"LANGSMITH_API_KEY\", \"LANGSMITH_ENDPOINT\", \"LANGSMITH_WEBHOOK_SECRET\"]
    missing = [var for var in required_vars if not os.getenv(var)]
    if missing:
        print(f\"Missing required env vars: {missing}\", file=sys.stderr)
        sys.exit(1)
    # Start Flask app on port 8080
    print(\"Starting LangSmith 0.3 webhook receiver on port 8080...\")
    app.run(host=\"0.0.0.0\", port=8080, debug=False)

# Short snippet for Tip 1: Enable gRPC push
from langsmith import Client
client = Client(
    api_key=\"your-api-key\",
    endpoint=\"https://api.smith.langchain.com\",
    use_grpc=True,
    grpc_max_send_message_length=10 * 1024 * 1024  # 10MB limit
)

# Short snippet for Tip 2: Webhook filter rule (set in LangSmith UI)
# Filter to only send error or high-latency traces:
error.exists() OR (end_time - start_time) > 2s

# Short snippet for Tip 3: Custom trace sampler
from langsmith.sampling import TraceSampler, SamplingDecision

class CustomSampler(TraceSampler):
    def should_sample(self, trace: Trace) -> SamplingDecision:
        if trace.error:
            return SamplingDecision.ALWAYS  # Store all error traces
        if (trace.end_time - trace.start_time).total_seconds() > 2:
            return SamplingDecision.ALWAYS  # Store high-latency traces
        return SamplingDecision.SAMPLE_10_PERCENT  # Sample 10% of healthy traces