langchain-observability

name	langchain-observability
description	Set up comprehensive observability for LangChain integrations. Use when implementing monitoring, setting up dashboards, or configuring alerting for LangChain application health. Trigger with phrases like "langchain monitoring", "langchain metrics", "langchain observability", "langchain tracing", "langchain alerts".
allowed-tools	Read, Write, Edit
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>

LangChain Observability

Overview

Set up comprehensive observability for LangChain applications with LangSmith, OpenTelemetry, and Prometheus.

Prerequisites

• LangChain application in staging/production
• LangSmith account (optional but recommended)
• Prometheus/Grafana infrastructure
• OpenTelemetry collector (optional)

Instructions

Step 1: Enable LangSmith Tracing

import os

# Configure LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-production-app"

# Optional: Set endpoint for self-hosted
# os.environ["LANGCHAIN_ENDPOINT"] = "https://langsmith.example.com"

from langchain_openai import ChatOpenAI

# All chains are automatically traced
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke("Hello!")  # Traced in LangSmith

Step 2: Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge, start_http_server
from langchain_core.callbacks import BaseCallbackHandler
import time

# Define metrics
LLM_REQUESTS = Counter(
    "langchain_llm_requests_total",
    "Total LLM requests",
    ["model", "status"]
)

LLM_LATENCY = Histogram(
    "langchain_llm_latency_seconds",
    "LLM request latency",
    ["model"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

LLM_TOKENS = Counter(
    "langchain_llm_tokens_total",
    "Total tokens processed",
    ["model", "type"]  # type: input or output
)

ACTIVE_REQUESTS = Gauge(
    "langchain_active_requests",
    "Currently active LLM requests"
)

class PrometheusCallback(BaseCallbackHandler):
    """Export metrics to Prometheus."""

    def __init__(self):
        self.start_times = {}

    def on_llm_start(self, serialized, prompts, run_id, **kwargs) -> None:
        ACTIVE_REQUESTS.inc()
        self.start_times[str(run_id)] = time.time()

    def on_llm_end() -> :
        ACTIVE_REQUESTS.dec()
        model = response.llm_output.get(, )  response.llm_output  

        
         (run_id)  .start_times:
            latency = time.time() - .start_times.pop((run_id))
            LLM_LATENCY.labels(model=model).observe(latency)

        
        LLM_REQUESTS.labels(model=model, status=).inc()

        
         response.llm_output    response.llm_output:
            usage = response.llm_output[]
            LLM_TOKENS.labels(model=model, =).inc(usage.get(, ))
            LLM_TOKENS.labels(model=model, =).inc(usage.get(, ))

     () -> :
        ACTIVE_REQUESTS.dec()
        LLM_REQUESTS.labels(model=, status=).inc()


start_http_server()  

Step 3: OpenTelemetry Integration

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

# Configure OpenTelemetry
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Instrument HTTP client (used by LangChain)
HTTPXClientInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

class OpenTelemetryCallback(BaseCallbackHandler):
    """Add OpenTelemetry spans for LangChain operations."""

    def __init__(self):
        self.spans = {}

    def on_chain_start(self, serialized, inputs, run_id, **kwargs) -> None:
        span = tracer.start_span(
            name=f"chain.{serialized.get('name', 'unknown')}",
            attributes={
                "langchain.chain_type": serialized.get("id", ["unknown"])[-1],
                "langchain.run_id": str(run_id),
            }
        )
        self.spans[str(run_id)] = span

    def on_chain_end(self, outputs, run_id, **kwargs) -> :
         (run_id)  .spans:
            span = .spans.pop((run_id))
            span.set_attribute(, (outputs.keys()))
            span.end()

     () -> :
        parent_span = .spans.get((parent_run_id))
        context = trace.set_span_in_context(parent_span)  parent_span  

        span = tracer.start_span(
            name=,
            context=context,
            attributes={
                : serialized.get(, [])[-],
                : (prompts),
            }
        )
        .spans[(run_id)] = span

     () -> :
         (run_id)  .spans:
            span = .spans.pop((run_id))
             response.llm_output    response.llm_output:
                usage = response.llm_output[]
                span.set_attribute(, usage.get(, ))
                span.set_attribute(, usage.get(, ))
            span.end()

Step 4: Structured Logging

import structlog
from datetime import datetime

# Configure structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
)

logger = structlog.get_logger()

class StructuredLoggingCallback(BaseCallbackHandler):
    """Emit structured logs for LangChain operations."""

    def on_llm_start(self, serialized, prompts, run_id, **kwargs) -> None:
        logger.info(
            "llm_start",
            run_id=str(run_id),
            model=serialized.get("name"),
            prompt_count=len(prompts)
        )

    def on_llm_end(self, response, run_id, **kwargs) -> None:
        token_usage = {}
        if response.llm_output and "token_usage" in response.llm_output:
            token_usage = response.llm_output["token_usage"]

        logger.info(
            "llm_end",
            run_id=str(run_id),
            generations=len(response.generations),
            **token_usage
        )

    def on_llm_error(self, error, run_id, **kwargs) -> None:
        logger.error(
            "llm_error",
            run_id=str(run_id),
            error_type=type(error).__name__,
            error_message=str(error)
        )

Step 5: Grafana Dashboard

{
  "title": "LangChain Observability",
  "panels": [
    {
      "title": "Request Rate",
      "type": "graph",
      "targets": [
        {
          "expr": "rate(langchain_llm_requests_total[5m])",
          "legendFormat": "{{model}} - {{status}}"
        }
      ]
    },
    {
      "title": "Latency P95",
      "type": "graph",
      "targets": [
        {
          "expr": "histogram_quantile(0.95, rate(langchain_llm_latency_seconds_bucket[5m]))",
          "legendFormat": "{{model}}"
        }
      ]
    
    
       
       
       
        
           
           
        
      
    
    
       
       
       
        
           
        
      
    
  

Step 6: Alerting Rules

# prometheus/alerts.yml
groups:
  - name: langchain
    rules:
      - alert: HighErrorRate
        expr: |
          sum(rate(langchain_llm_requests_total{status="error"}[5m]))
          / sum(rate(langchain_llm_requests_total[5m])) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High LLM error rate"
          description: "Error rate is {{ $value | humanizePercentage }}"

      - alert: HighLatency
        expr: |
          histogram_quantile(0.95, rate(langchain_llm_latency_seconds_bucket[5m])) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High LLM latency"
          description: "P95 latency is {{ $value }}s"

      - alert: TokenBudgetExceeded
        expr: |
          sum(increase(langchain_llm_tokens_total[1h])) > 1000000
        labels:
          severity: warning
        annotations:
          summary: 
           

Output

• LangSmith tracing enabled
• Prometheus metrics exported
• OpenTelemetry spans
• Structured logging
• Grafana dashboard and alerts

Resources

• LangSmith Documentation
• OpenTelemetry Python
• Prometheus Python Client

Next Steps

Use langchain-incident-runbook for incident response procedures.