Debugging LLM Hallucinations in Prod: How We Used LangSmith 0.8 and Datadog 7.0 to Cut Errors by 60%

import os
import sys
from typing import Dict, Any, Optional
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
from langsmith import Client, traceable
from langsmith.run_trees import RunTree

# Initialize LangSmith client with error handling
def init_langsmith_client() -> Optional[Client]:
    """Initialize LangSmith client, return None if misconfigured."""
    api_key = os.getenv("LANGSMITH_API_KEY")
    project_name = os.getenv("LANGSMITH_PROJECT", "prod-llm-support")
    endpoint = os.getenv("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")

    if not api_key:
        print("ERROR: LANGSMITH_API_KEY not set", file=sys.stderr)
        return None
    try:
        client = Client(api_key=api_key, api_url=endpoint)
        # Verify project exists, create if not
        try:
            client.read_project(project_name=project_name)
        except Exception:
            client.create_project(project_name=project_name)
        print(f"LangSmith client initialized for project: {project_name}")
        return client
    except Exception as e:
        print(f"ERROR: Failed to initialize LangSmith client: {str(e)}", file=sys.stderr)
        return None

# Initialize LangSmith tracing for the chain
@traceable(
    name="support-chatbot-chain",
    tags=["env:prod", "llm:gpt-4-turbo", "langsmith-version:0.8"],
    metadata={"team": "customer-support-eng", "pipeline": "chatbot-v2"}
)
def run_support_chain(user_query: str, session_id: str) -> Dict[str, Any]:
    """Run the support chatbot chain with full LangSmith tracing."""
    # Initialize LLM with error handling
    openai_api_key = os.getenv("OPENAI_API_KEY")
    if not openai_api_key:
        raise ValueError("OPENAI_API_KEY environment variable not set")

    llm = ChatOpenAI(
        model="gpt-4-turbo-preview",
        temperature=0.1,  # Low temp to reduce hallucinations
        openai_api_key=openai_api_key,
        max_retries=3,
        request_timeout=30
    )

    # Prompt template with guardrails to reduce hallucinations
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a customer support agent for CloudOps Inc. 
        Only answer questions about our SaaS product features, billing, and support policies.
        If you don't know the answer, say "I don't have that information, please contact support@cloudops.com".
        Never make up product features, pricing, or policies.
        Current policy version: 2025-02-01"""),
        ("human", "{user_query}"),
        ("ai", "Session ID: {session_id}")
    ])

    # Create chain
    chain = LLMChain(llm=llm, prompt=prompt)

    try:
        # Run chain with tracing metadata
        response = chain.invoke({
            "user_query": user_query,
            "session_id": session_id
        }, config={
            "run_name": f"support-session-{session_id}",
            "tags": [f"session:{session_id}"],
            "metadata": {"user_query_length": len(user_query)}
        })
        return {
            "response": response["text"],
            "session_id": session_id,
            "status": "success"
        }
    except Exception as e:
        # Log error to LangSmith as a failed run
        raise RuntimeError(f"Chain execution failed: {str(e)}") from e

if __name__ == "__main__":
    # Example usage
    langsmith_client = init_langsmith_client()
    if not langsmith_client:
        sys.exit(1)

    test_query = "What's the price of your enterprise plan?"
    test_session = "sess_123456"
    try:
        result = run_support_chain(test_query, test_session)
        print(f"Response: {result['response']}")
    except Exception as e:
        print(f"ERROR: {str(e)}", file=sys.stderr)
        sys.exit(1)

import os
import sys
import time
from typing import Dict, List, Optional
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v2.api.metrics_api import MetricsApi
from datadog_api_client.v2.model.metric_payload import MetricPayload
from datadog_api_client.v2.model.metric_point import MetricPoint
from datadog_api_client.v2.model.metric_series import MetricSeries
from datadog_api_client.v2.model.metric_resource import MetricResource

# Initialize Datadog client for version 7.0 API
def init_datadog_client() -> Optional[Configuration]:
    """Initialize Datadog API client with v7 config, return None on error."""
    api_key = os.getenv("DATADOG_API_KEY")
    app_key = os.getenv("DATADOG_APP_KEY")
    site = os.getenv("DATADOG_SITE", "datadoghq.com")  # Default to US site

    if not api_key or not app_key:
        print("ERROR: DATADOG_API_KEY and DATADOG_APP_KEY must be set", file=sys.stderr)
        return None

    try:
        config = Configuration()
        config.api_key["apiKeyAuth"] = api_key
        config.api_key["appKeyAuth"] = app_key
        config.server_variables["site"] = site
        config.debug = False  # Set to True for local debugging
        print(f"Datadog client initialized for site: {site}")
        return config
    except Exception as e:
        print(f"ERROR: Failed to initialize Datadog client: {str(e)}", file=sys.stderr)
        return None

def submit_llm_metrics(
    config: Configuration,
    session_id: str,
    prompt_tokens: int,
    completion_tokens: int,
    latency_ms: int,
    is_hallucination: bool,
    model_name: str = "gpt-4-turbo-preview"
) -> bool:
    """Submit custom LLM metrics to Datadog 7.0, return success status."""
    # Define metric series with LLM-specific tags
    metrics: List[MetricSeries] = []

    # 1. Token usage metrics
    metrics.append(MetricSeries(
        metric="llm.token_usage.prompt",
        type="count",
        points=[MetricPoint(timestamp=int(time.time()), value=prompt_tokens)],
        tags=[
            f"env:prod",
            f"model:{model_name}",
            f"session:{session_id}",
            f"pipeline:support-chatbot"
        ],
        resources=[MetricResource(name="llm-pipeline")]
    ))
    metrics.append(MetricSeries(
        metric="llm.token_usage.completion",
        type="count",
        points=[MetricPoint(timestamp=int(time.time()), value=completion_tokens)],
        tags=[
            f"env:prod",
            f"model:{model_name}",
            f"session:{session_id}",
            f"pipeline:support-chatbot"
        ],
        resources=[MetricResource(name="llm-pipeline")]
    ))

    # 2. Latency metric
    metrics.append(MetricSeries(
        metric="llm.latency.ms",
        type="gauge",
        points=[MetricPoint(timestamp=int(time.time()), value=latency_ms)],
        tags=[
            f"env:prod",
            f"model:{model_name}",
            f"session:{session_id}"
        ]
    ))

    # 3. Hallucination flag metric (1 for hallucination, 0 for valid)
    metrics.append(MetricSeries(
        metric="llm.hallucination.flag",
        type="count",
        points=[MetricPoint(timestamp=int(time.time()), value=1 if is_hallucination else 0)],
        tags=[
            f"env:prod",
            f"model:{model_name}",
            f"session:{session_id}",
            f"hallucination_type:factually_incorrect"
        ]
    ))

    # Submit metrics
    try:
        with ApiClient(config) as api_client:
            api_instance = MetricsApi(api_client)
            payload = MetricPayload(series=metrics)
            response = api_instance.submit_metrics(payload)
            print(f"Submitted {len(metrics)} metrics to Datadog: {response.get('status', 'unknown')}")
            return True
    except Exception as e:
        print(f"ERROR: Failed to submit metrics: {str(e)}", file=sys.stderr)
        return False

if __name__ == "__main__":
    # Example usage
    dd_config = init_datadog_client()
    if not dd_config:
        sys.exit(1)

    # Simulated chain run output
    test_session = "sess_123456"
    submit_llm_metrics(
        config=dd_config,
        session_id=test_session,
        prompt_tokens=128,
        completion_tokens=256,
        latency_ms=1200,
        is_hallucination=False,
        model_name="gpt-4-turbo-preview"
    )

    # Simulated hallucination case
    submit_llm_metrics(
        config=dd_config,
        session_id="sess_789012",
        prompt_tokens=96,
        completion_tokens=320,
        latency_ms=1500,
        is_hallucination=True,
        model_name="gpt-4-turbo-preview"
    )

import os
import sys
import time
from typing import Dict, Any, List, Optional
from langsmith import Client as LangSmithClient
from datadog_api_client import ApiClient, Configuration as DatadogConfig
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v2.api.monitors_api import MonitorsApi
from datadog_api_client.v2.model.monitor import Monitor
from datadog_api_client.v2.model.monitor_type import MonitorType
from datadog_api_client.v2.model.monitor_options import MonitorOptions
from datadog_api_client.v2.model.monitor_thresholds import MonitorThresholds

# Initialize both clients
def init_clients() -> Dict[str, Any]:
    """Initialize LangSmith and Datadog clients, return dict or empty on error."""
    clients = {}

    # LangSmith client
    ls_api_key = os.getenv("LANGSMITH_API_KEY")
    ls_project = os.getenv("LANGSMITH_PROJECT", "prod-llm-support")
    if not ls_api_key:
        print("ERROR: LANGSMITH_API_KEY not set", file=sys.stderr)
        return {}
    try:
        clients["langsmith"] = LangSmithClient(
            api_key=ls_api_key,
            api_url=os.getenv("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")
        )
        clients["langsmith_project"] = ls_project
    except Exception as e:
        print(f"ERROR: LangSmith init failed: {str(e)}", file=sys.stderr)
        return {}

    # Datadog client
    dd_api_key = os.getenv("DATADOG_API_KEY")
    dd_app_key = os.getenv("DATADOG_APP_KEY")
    if not dd_api_key or not dd_app_key:
        print("ERROR: Datadog keys not set", file=sys.stderr)
        return {}
    try:
        dd_config = DatadogConfig()
        dd_config.api_key["apiKeyAuth"] = dd_api_key
        dd_config.api_key["appKeyAuth"] = dd_app_key
        dd_config.server_variables["site"] = os.getenv("DATADOG_SITE", "datadoghq.com")
        clients["datadog_config"] = dd_config
    except Exception as e:
        print(f"ERROR: Datadog init failed: {str(e)}", file=sys.stderr)
        return {}

    return clients

def fetch_recent_hallucinations(ls_client: LangSmithClient, project: str, hours: int = 1) -> List[Dict[str, Any]]:
    """Fetch runs tagged as hallucinations from LangSmith in last N hours."""
    try:
        # Query runs with hallucination tag, failed status
        runs = ls_client.list_runs(
            project_name=project,
            filter="tag:hallucination_detected",
            start_time=time.time() - (hours * 3600),
            limit=100
        )
        return [run for run in runs if run.error is None]  # Exclude errored runs
    except Exception as e:
        print(f"ERROR: Failed to fetch LangSmith runs: {str(e)}", file=sys.stderr)
        return []

def create_datadog_hallucination_alert(dd_config: DatadogConfig, threshold_pct: float = 10.0) -> Optional[str]:
    """Create a Datadog monitor for hallucination rate exceeding threshold."""
    monitor_name = f"Prod LLM Hallucination Rate > {threshold_pct}%"
    query = f"avg(last_5m):sum:llm.hallucination.flag{{env:prod,pipeline:support-chatbot}}.as_count() / sum:llm.chain.run.count{{env:prod,pipeline:support-chatbot}}.as_count() * 100 > {threshold_pct}"

    monitor = Monitor(
        name=monitor_name,
        type=MonitorType.METRIC,
        query=query,
        message="""Hallucination rate for support chatbot exceeded threshold.
        Check LangSmith dashboard: https://smith.langchain.com/projects/prod-llm-support
        Datadog dashboard: https://app.datadoghq.com/dashboard/llm-support
        @support-eng-team""",
        tags=["env:prod", "pipeline:support-chatbot", "alert_type:llm_hallucination"],
        options=MonitorOptions(
            thresholds=MonitorThresholds(critical=threshold_pct),
            notify_no_data=True,
            no_data_timeframe=10
        )
    )

    try:
        with ApiClient(dd_config) as api_client:
            api_instance = MonitorsApi(api_client)
            response = api_instance.create_monitor(monitor)
            monitor_id = response.get("id")
            print(f"Created Datadog monitor: {monitor_name} (ID: {monitor_id})")
            return monitor_id
    except Exception as e:
        print(f"ERROR: Failed to create Datadog monitor: {str(e)}", file=sys.stderr)
        return None

def sync_hallucination_feedback(clients: Dict[str, Any]) -> None:
    """Sync hallucination flags from LangSmith to Datadog as logs."""
    ls_client = clients.get("langsmith")
    dd_config = clients.get("datadog_config")
    project = clients.get("langsmith_project")

    if not all([ls_client, dd_config, project]):
        print("ERROR: Missing clients for sync", file=sys.stderr)
        return

    # Fetch recent hallucinations
    hallucinations = fetch_recent_hallucinations(ls_client, project, hours=1)
    print(f"Found {len(hallucinations)} recent hallucinations to sync")

    # Submit each as a Datadog log
    for run in hallucinations:
        log_payload = {
            "ddsource": "langsmith",
            "ddtags": f"env:prod,pipeline:support-chatbot,run_id:{run.id},session:{run.metadata.get('session_id', 'unknown')}",
            "message": f"Hallucination detected in run {run.id}",
            "run_id": str(run.id),
            "prompt": run.inputs.get("user_query", ""),
            "response": run.outputs.get("response", "") if run.outputs else "",
            "timestamp": int(run.start_time.timestamp())
        }

        try:
            with ApiClient(dd_config) as api_client:
                api_instance = LogsApi(api_client)
                api_instance.submit_log(log_payload)
        except Exception as e:
            print(f"ERROR: Failed to submit log for run {run.id}: {str(e)}", file=sys.stderr)

if __name__ == "__main__":
    clients = init_clients()
    if not clients:
        sys.exit(1)

    # Sync recent hallucinations
    sync_hallucination_feedback(clients)

    # Create alert if not exists (idempotent check omitted for brevity)
    create_datadog_hallucination_alert(clients["datadog_config"], threshold_pct=10.0)

@traceable(
    name="support-chain",
    tags=["env:prod", "pipeline:support-chatbot", "prompt_version:v2.1"],
    metadata={"model": "gpt-4-turbo", "session_id": session_id}
)
def run_chain(query: str):
    # chain logic here

# Submit prompt version metric to Datadog
MetricSeries(
    metric="llm.prompt.version",
    type="count",
    points=[MetricPoint(timestamp=int(time.time()), value=1)],
    tags=["prompt_version:v2.1", "env:prod"]
)

# Tag a LangSmith run as hallucinated
ls_client.create_feedback(
    run_id=run_id,
    key="hallucination_detected",
    value=True,
    comment="Response conflicts with knowledge base article KB-1234"
)