Opinion: AI Observability Tools Like OpenTelemetry 1.20 and LangChain 0.3 Are More Important Than AI Model Accuracy in 2026

import os
import logging
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv

# LangChain 0.3 core imports
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

# OpenTelemetry 1.20 GenAI instrumentation
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.langchain import LangChainInstrumentor
from opentelemetry.semconv.ai import SpanAttributes as AISemanticConventions

# Configure logging for error handling
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables (OpenAI key, OTel collector endpoint)
load_dotenv()

def init_otel_tracer() -> trace.Tracer:
    \"\"\"Initialize OpenTelemetry 1.20 tracer with GenAI semantic conventions.\"\"\"
    try:
        # Set up tracer provider with GenAI attributes
        provider = TracerProvider()
        # Configure OTLP exporter to send spans to collector (e.g., Jaeger, Prometheus)
        otlp_exporter = OTLPSpanExporter(
            endpoint=os.getenv('OTEL_EXPORTER_OTLP_ENDPOINT', 'http://localhost:4317')
        )
        span_processor = BatchSpanProcessor(otlp_exporter)
        provider.add_span_processor(span_processor)
        trace.set_tracer_provider(provider)

        # Instrument LangChain 0.3 with OpenTelemetry
        LangChainInstrumentor().instrument()
        logger.info('OpenTelemetry 1.20 tracer initialized with LangChain 0.3 instrumentation')
        return trace.get_tracer(__name__)
    except Exception as e:
        logger.error(f'Failed to initialize OTel tracer: {str(e)}')
        raise

def setup_rag_pipeline(tracer: trace.Tracer) -> Any:
    \"\"\"Set up LangChain 0.3 RAG pipeline with OTel instrumentation.\"\"\"
    with tracer.start_as_current_span('setup_rag_pipeline') as span:
        try:
            # Load and split documents
            loader = TextLoader('company_docs.txt')
            documents = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
            chunks = text_splitter.split_documents(documents)
            span.set_attribute(AISemanticConventions.AI_DOCUMENT_COUNT, len(chunks))

            # Create vector store with embeddings
            embeddings = OpenAIEmbeddings()
            vectorstore = Chroma.from_documents(chunks, embeddings)
            retriever = vectorstore.as_retriever(search_kwargs={'k': 3})
            span.set_attribute(AISemanticConventions.AI_RETRIEVER_TOP_K, 3)

            # Set up LLM and chain
            llm = ChatOpenAI(model='gpt-4o-mini', temperature=0.1)
            prompt = ChatPromptTemplate.from_messages([
                ('system', 'You are a company support agent. Use the provided context to answer questions.'),
                ('human', 'Context: {context}\n\nQuestion: {question}')
            ])
            chain = prompt | llm | StrOutputParser()
            span.set_attribute(AISemanticConventions.AI_LLM_MODEL, 'gpt-4o-mini')

            logger.info('RAG pipeline setup complete with OTel instrumentation')
            return retriever, chain
        except Exception as e:
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            logger.error(f'RAG pipeline setup failed: {str(e)}')
            raise

if __name__ == '__main__':
    try:
        tracer = init_otel_tracer()
        retriever, chain = setup_rag_pipeline(tracer)
        # Test query with tracing
        with tracer.start_as_current_span('test_rag_query') as query_span:
            question = 'What is our refund policy?'
            context = retriever.invoke(question)
            response = chain.invoke({'context': context, 'question': question})
            query_span.set_attribute(AISemanticConventions.AI_QUERY_TEXT, question)
            query_span.set_attribute(AISemanticConventions.AI_RESPONSE_TEXT, response)
            print(f'Response: {response}')
    except Exception as e:
        logger.error(f'Application failed: {str(e)}')
        exit(1)

import { ChatOpenAI } from '@langchain/openai';
import { ChatPromptTemplate } from '@langchain/core/prompts';
import { StringOutputParser } from '@langchain/core/output_parsers';
import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter';
import { TextLoader } from 'langchain/document_loaders/fs/text';
import { Chroma } from '@langchain/community/vectorstores/chroma';
import { OpenAIEmbeddings } from '@langchain/openai';
import { trace, Tracer, Span, SpanStatusCode } from '@opentelemetry/api';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { LangChainInstrumentation } from '@opentelemetry/instrumentation-langchain';
import { registerInstrumentations } from '@opentelemetry/instrumentation';
import * as dotenv from 'dotenv';
import { logger } from './logger'; // Assume winston-based logger

dotenv.config();

// Initialize OpenTelemetry 1.20 tracer for Node.js
function initOtelTracer(): Tracer {
    try {
        const provider = new NodeTracerProvider();
        const exporter = new OTLPTraceExporter({
            url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || 'http://localhost:4317',
        });
        const processor = new BatchSpanProcessor(exporter);
        provider.addSpanProcessor(processor);
        provider.register();

        // Register LangChain 0.3 instrumentation
        registerInstrumentations({
            instrumentations: [new LangChainInstrumentation()],
        });

        logger.info('OpenTelemetry 1.20 tracer initialized for LangChain.js 0.3');
        return trace.getTracer('langchain-rag-app');
    } catch (error) {
        logger.error(`OTel initialization failed: ${error.message}`);
        throw error;
    }
}

// Set up RAG pipeline with OTel tracing
async function setupRagPipeline(tracer: Tracer): Promise<{
    retriever: any;
    chain: any;
}> {
    const span: Span = tracer.startSpan('setup_rag_pipeline');
    try {
        span.setAttribute('ai.system', 'rag');
        span.setAttribute('langchain.version', '0.3.0');

        // Load and split documents
        const loader = new TextLoader('./company_docs.txt');
        const docs = await loader.load();
        const splitter = new RecursiveCharacterTextSplitter({
            chunkSize: 1000,
            chunkOverlap: 200,
        });
        const chunks = await splitter.splitDocuments(docs);
        span.setAttribute('ai.document.count', chunks.length);

        // Create vector store
        const embeddings = new OpenAIEmbeddings();
        const vectorstore = await Chroma.fromDocuments(chunks, embeddings);
        const retriever = vectorstore.asRetriever({ k: 3 });
        span.setAttribute('ai.retriever.top_k', 3);

        // Initialize LLM and chain
        const llm = new ChatOpenAI({
            modelName: 'gpt-4o-mini',
            temperature: 0.1,
        });
        const prompt = ChatPromptTemplate.fromMessages([
            ['system', 'You are a company support agent. Use context to answer questions.'],
            ['human', 'Context: {context}\n\nQuestion: {question}'],
        ]);
        const chain = prompt.pipe(llm).pipe(new StringOutputParser());
        span.setAttribute('ai.llm.model', 'gpt-4o-mini');

        logger.info('LangChain.js 0.3 RAG pipeline setup complete');
        span.setStatus({ code: SpanStatusCode.OK });
        return { retriever, chain };
    } catch (error) {
        span.recordException(error);
        span.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
        logger.error(`RAG setup failed: ${error.message}`);
        throw error;
    } finally {
        span.end();
    }
}

// Execute sample query with tracing
async function runSampleQuery(
    tracer: Tracer,
    retriever: any,
    chain: any
): Promise {
    const span: Span = tracer.startSpan('sample_rag_query');
    try {
        const question = 'What is our refund policy?';
        span.setAttribute('ai.query.text', question);

        const context = await retriever.invoke(question);
        span.setAttribute('ai.context.length', JSON.stringify(context).length);

        const response = await chain.invoke({ context, question });
        span.setAttribute('ai.response.text', response);

        console.log(`Response: ${response}`);
        span.setStatus({ code: SpanStatusCode.OK });
    } catch (error) {
        span.recordException(error);
        span.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
        logger.error(`Query failed: ${error.message}`);
        throw error;
    } finally {
        span.end();
    }
}

// Main execution
(async () => {
    try {
        const tracer = initOtelTracer();
        const { retriever, chain } = await setupRagPipeline(tracer);
        await runSampleQuery(tracer, retriever, chain);
    } catch (error) {
        logger.error(`Application crashed: ${error.message}`);
        process.exit(1);
    }
})();

import os
import time
import logging
from typing import List, Dict, Tuple
from datetime import datetime, timedelta

# OpenTelemetry 1.20 metrics imports
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.semconv.ai import MetricAttributes as AIMetrics

# Mock model accuracy and drift data (simulate production data)
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ModelPrediction:
    timestamp: datetime
    query: str
    response: str
    confidence: float
    actual_label: Optional[str] = None  # Ground truth for accuracy calc
    is_accurate: Optional[bool] = None

def init_otel_metrics() -> metrics.Meter:
    \"\"\"Initialize OpenTelemetry 1.20 metrics for AI model monitoring.\"\"\"
    try:
        # Set up metric exporter to send to collector
        metric_exporter = OTLPMetricExporter(
            endpoint=os.getenv('OTEL_EXPORTER_OTLP_ENDPOINT', 'http://localhost:4317')
        )
        metric_reader = PeriodicExportingMetricReader(
            exporter=metric_exporter,
            export_interval_millis=10000  # Export every 10s
        )
        provider = MeterProvider(metric_readers=[metric_reader])
        metrics.set_meter_provider(provider)

        meter = provider.get_meter('ai-model-monitor')
        logger.info('OpenTelemetry 1.20 metrics initialized for model monitoring')
        return meter
    except Exception as e:
        logger.error(f'Failed to initialize OTel metrics: {str(e)}')
        raise

def setup_model_metrics(meter: metrics.Meter) -> Tuple[metrics.Counter, metrics.Histogram, metrics.Gauge]:
    \"\"\"Define custom AI metrics for accuracy and drift tracking.\"\"\"
    try:
        # Counter: total predictions made
        prediction_counter = meter.create_counter(
            name='ai.model.predictions.total',
            description='Total number of model predictions',
            unit='1'
        )

        # Histogram: prediction confidence distribution
        confidence_histogram = meter.create_histogram(
            name='ai.model.confidence',
            description='Distribution of model prediction confidence scores',
            unit='1',
            boundaries=[0.1, 0.3, 0.5, 0.7, 0.9, 1.0]
        )

        # Gauge: current model accuracy (calculated from ground truth)
        accuracy_gauge = meter.create_gauge(
            name='ai.model.accuracy',
            description='Current model accuracy (0-1)',
            unit='1'
        )

        # Gauge: drift score (KL divergence between training and production data)
        drift_gauge = meter.create_gauge(
            name='ai.model.drift_score',
            description='KL divergence drift score (0 = no drift, >0.5 = critical drift)',
            unit='1'
        )

        logger.info('Model metrics defined: predictions, confidence, accuracy, drift')
        return prediction_counter, confidence_histogram, accuracy_gauge, drift_gauge
    except Exception as e:
        logger.error(f'Failed to setup model metrics: {str(e)}')
        raise

def simulate_production_predictions(
    meter: metrics.Meter,
    prediction_counter: metrics.Counter,
    confidence_histogram: metrics.Histogram,
    accuracy_gauge: metrics.Gauge,
    drift_gauge: metrics.Gauge,
    num_predictions: int = 1000
) -> List[ModelPrediction]:
    \"\"\"Simulate production predictions and record metrics.\"\"\"
    predictions = []
    accurate_count = 0

    for i in range(num_predictions):
        # Simulate prediction with random confidence (drift increases over time)
        drift_factor = min(1.0, i / num_predictions * 2)  # Simulate increasing drift
        confidence = max(0.1, 0.9 - drift_factor * 0.4)  # Confidence drops as drift increases
        is_accurate = confidence > 0.7  # Simulate accuracy tied to confidence
        if is_accurate:
            accurate_count += 1

        # Create prediction record
        pred = ModelPrediction(
            timestamp=datetime.now() - timedelta(minutes=num_predictions - i),
            query=f'Sample query {i}',
            response=f'Sample response {i}',
            confidence=confidence,
            is_accurate=is_accurate
        )
        predictions.append(pred)

        # Record metrics
        prediction_counter.add(1, {AIMetrics.AI_MODEL_TYPE: 'llm', AIMetrics.AI_LLM_MODEL: 'gpt-4o-mini'})
        confidence_histogram.record(confidence, {AIMetrics.AI_MODEL_TYPE: 'llm'})
        current_accuracy = accurate_count / (i + 1)
        accuracy_gauge.set(current_accuracy, {AIMetrics.AI_MODEL_TYPE: 'llm'})
        drift_gauge.set(drift_factor, {AIMetrics.AI_MODEL_TYPE: 'llm'})  # Simulate drift score

        if i % 100 == 0:
            logger.info(f'Processed {i} predictions. Current accuracy: {current_accuracy:.2f}, Drift: {drift_factor:.2f}')

        time.sleep(0.01)  # Simulate production pace

    final_accuracy = accurate_count / num_predictions
    logger.info(f'Simulation complete. Final accuracy: {final_accuracy:.2f}, Max drift: {drift_factor:.2f}')
    return predictions

if __name__ == '__main__':
    try:
        meter = init_otel_metrics()
        prediction_counter, confidence_histogram, accuracy_gauge, drift_gauge = setup_model_metrics(meter)
        simulate_production_predictions(
            meter, prediction_counter, confidence_histogram, accuracy_gauge, drift_gauge
        )
        # Keep process alive to allow metric export
        logger.info('Metrics exported. Keeping process alive for 30s...')
        time.sleep(30)
    except Exception as e:
        logger.error(f'Metrics simulation failed: {str(e)}')
        exit(1)

from opentelemetry.instrumentation.langchain import LangChainInstrumentor
LangChainInstrumentor().instrument()

from opentelemetry import trace
tracer = trace.get_tracer(__name__)

def retrieve_documents(query: str) -> List[Document]:
    with tracer.start_as_current_span('retriever_invoke') as span:
        span.set_attribute('ai.retriever.query', query)
        docs = retriever.invoke(query)
        span.set_attribute('ai.document.count', len(docs))
        return docs