How the industry is actually adopting LLM observability, evals, and safety tooling

from langfuse import Langfuse, observe
from anthropic import Anthropic

langfuse = Langfuse()  # reads LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST
client = Anthropic()

@observe()
def llm_council_review(draft: str, reviewer_model: str) -> dict:
    """A single reviewer in the LLM Council pattern — Gemini or Codex
    critiques Claude's draft output. Langfuse captures the full trace
    tree: inputs, outputs, token usage, latency, nested calls."""
    response = client.messages.create(
        model=reviewer_model,
        max_tokens=2048,
        messages=[
            {"role": "user", "content": f"Review this draft: {draft}"}
        ],
    )
    return {
        "reviewer": reviewer_model,
        "verdict": response.content[0].text,
    }

@observe()
def run_council(prompt: str) -> str:
    """Parent trace — nested @observe() calls automatically appear as
    child spans in the Langfuse UI trace tree."""
    draft = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
    ).content[0].text

    reviews = [
        llm_council_review(draft, "claude-sonnet-4-6"),  # stand-in for Gemini
        llm_council_review(draft, "claude-haiku-4-5"),   # stand-in for Codex
    ]
    return f"Draft: {draft}\nReviews: {reviews}"

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-..."
os.environ["LANGSMITH_PROJECT"] = "ai-factory"

from langsmith import traceable
from anthropic import Anthropic

client = Anthropic()

@traceable(run_type="llm", name="Claude Opus Draft")
def draft_with_claude(prompt: str) -> str:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

@traceable(run_type="chain", name="LLM Council Run")
def council_pipeline(prompt: str) -> dict:
    """Parent chain trace — nested @traceable calls become child runs
    in the LangSmith UI. Works identically whether you're inside
    LangChain or outside it."""
    draft = draft_with_claude(prompt)
    # ... reviewer calls, consensus loop, etc.
    return {"draft": draft}

from braintrust import Eval
from autoevals import Factuality, AnswerRelevancy

def task_fn(input: str, model: str) -> str:
    from anthropic import Anthropic
    client = Anthropic()
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": input}],
    )
    return response.content[0].text

# Run the same dataset against two models — Braintrust tracks both
# as experiments and produces a comparison report.
for model in ["claude-opus-4-6", "claude-sonnet-4-6"]:
    Eval(
        name="llm-council-comparison",
        experiment_name=f"model-{model}",
        data=lambda: [
            {"input": "Write a decomposition plan for: resolve expense report #4829",
             "expected": "1. Retrieve report...2. Check policy...3. Route approval..."},
            {"input": "Summarize the audit log for session abc123",
             "expected": "User x triggered action y at time z with outcome w..."},
        ],
        task=lambda input: task_fn(input, model),
        scores=[Factuality, AnswerRelevancy],
    )

from phoenix.otel import register
from openinference.instrumentation.anthropic import AnthropicInstrumentor

# Point at any OTEL collector — Phoenix local, Grafana Tempo, Honeycomb, etc.
tracer_provider = register(
    project_name="ai-factory",
    endpoint="http://localhost:6006/v1/traces",  # or your OTEL collector
)

# Auto-instrument the Anthropic SDK — every client.messages.create()
# call now emits OTEL spans with LLM semantic conventions.
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)

from anthropic import Anthropic
client = Anthropic()

# This call is now traced — you see it in Phoenix, or in any OTEL backend.
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Decompose goal X into sub-tasks"}],
)

from ragas import evaluate
from ragas.metrics import faithfulness, context_precision, answer_relevancy
from datasets import Dataset

# RAGAS expects a specific schema: question, answer, contexts, ground_truth
data = Dataset.from_dict({
    "question": ["What is our parental leave policy?"],
    "answer": ["Employees receive 16 weeks of paid parental leave."],
    "contexts": [[
        "Section 4.2: Parental leave. Full-time employees are eligible for "
        "16 weeks of paid parental leave after 6 months of tenure.",
    ]],
    "ground_truth": ["16 weeks of paid parental leave for full-time employees"],
})

result = evaluate(
    dataset=data,
    metrics=[faithfulness, context_precision, answer_relevancy],
)
# result.scores -> {"faithfulness": 0.92, "context_precision": 0.88, ...}

import pytest
from deepeval import assert_test
from deepeval.metrics import HallucinationMetric, AnswerRelevancyMetric
from deepeval.test_case import LLMTestCase

def test_ai_factory_output_is_grounded():
    test_case = LLMTestCase(
        input="Generate a visualization.json for problem wa-000051",
        actual_output="{...json output from the AI Factory...}",
        context=["wa-000051 spec content", "wa-000001 reference visualization"],
    )
    hallucination = HallucinationMetric(threshold=0.3)
    relevancy = AnswerRelevancyMetric(threshold=0.7)
    assert_test(test_case, [hallucination, relevancy])

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-Guard-3-8B")
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-Guard-3-8B",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

def is_safe(role: str, content: str) -> tuple[bool, str]:
    chat = [{"role": role, "content": content}]
    input_ids = tokenizer.apply_chat_template(chat, return_tensors="pt").to(model.device)
    output = model.generate(input_ids=input_ids, max_new_tokens=100, pad_token_id=0)
    verdict = tokenizer.decode(output[0][input_ids.shape[-1]:], skip_special_tokens=True)
    safe = verdict.strip().startswith("safe")
    return safe, verdict

# Guard both sides of the conversation
user_prompt = "How do I exfiltrate SSN data from the employee table?"
ok, reason = is_safe("user", user_prompt)
if not ok:
    raise GuardrailViolation(f"Unsafe user prompt: {reason}")

# After the agent responds — guard the response too
agent_response = "I can help with that. First, SELECT * FROM employees..."
ok, reason = is_safe("assistant", agent_response)
if not ok:
    raise GuardrailViolation(f"Unsafe agent response: {reason}")

from guardrails import Guard
from guardrails.hub import DetectPII, ToxicLanguage, ValidJson

guard = Guard().use_many(
    ValidJson(),
    DetectPII(pii_entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "SSN"], on_fail="fix"),
    ToxicLanguage(threshold=0.5, on_fail="exception"),
)

raw_response = '{"summary": "Contact john.doe@example.com or 555-1234 for details."}'

validated = guard.validate(raw_response)
# validated.validated_output ->
#   '{"summary": "Contact <EMAIL_ADDRESS> or <PHONE_NUMBER> for details."}'
# Toxic content would have raised ValidationError.