LlamaIndex — Evaluation & Observability¶

from llama_index.core.evaluation import (
    CorrectnessEvaluator, FaithfulnessEvaluator, GuidelineEvaluator, RelevancyEvaluator, SemanticSimilarityEvaluator,
)

judge = OpenAI(model="gpt-4o", temperature=0)
query = "How long does express shipping take?"
response = query_engine.query(query)                                  # Response with source_nodes

faith = FaithfulnessEvaluator(llm=judge).evaluate_response(query=query, response=response)
rel = RelevancyEvaluator(llm=judge).evaluate_response(query=query, response=response)
print(faith.passing, rel.passing, faith.feedback)

corr = CorrectnessEvaluator(llm=judge).evaluate(
    query=query, response=str(response), reference="Express shipping takes 1 to 2 business days."
)
sim = SemanticSimilarityEvaluator(embed_model=embed_model).evaluate(response=str(response), reference="1-2 business days")
rule = GuidelineEvaluator(llm=judge, guidelines="Answer in one sentence and never quote prices.").evaluate(query=query, response=str(response))
assert faith.passing and corr.score >= 4 and rule.passing

from llama_index.core.evaluation import BatchEvalRunner

runner = BatchEvalRunner(
    {"faithfulness": FaithfulnessEvaluator(llm=judge), "relevancy": RelevancyEvaluator(llm=judge)},
    workers=4,                                               # concurrent evaluations
)
results = await runner.aevaluate_queries(query_engine, queries=questions)      # runs the engine, then every evaluator
# results == {"faithfulness": [EvaluationResult, ...], "relevancy": [...]}  in query order
rate = {name: sum(r.passing for r in rs) / len(rs) for name, rs in results.items()}

from llama_index.core.evaluation import RetrieverEvaluator

evaluator = RetrieverEvaluator.from_metric_names(
    ["hit_rate", "mrr", "precision", "recall", "ndcg"],
    retriever=index.as_retriever(similarity_top_k=3),
)
result = evaluator.evaluate(query="express delivery takes 1 to 2 days", expected_ids=["ship-express#0"])
result.metric_vals_dict       # {'hit_rate': 1.0, 'mrr': 1.0, 'precision': 0.333, 'recall': 1.0, 'ndcg': 1.0}
result.retrieved_ids          # ['ship-express#0', 'ship-standard#0', 'refund-physical#0']

from llama_index.core.evaluation import DatasetGenerator

golden = []
for node in nodes[:50]:
    gen = DatasetGenerator(nodes=[node], llm=judge, num_questions_per_chunk=2)
    for question in gen.generate_questions_from_nodes():
        golden.append({"query": question, "expected_ids": [node.node_id]})        # the source chunk is the answer

dataset = DatasetGenerator(nodes=nodes[:10], llm=judge, num_questions_per_chunk=2).generate_dataset_from_nodes(num=10)
dataset.qr_pairs                                                                   # [(question, answer)]
dataset.save_json("qa.json")                                                       # QueryResponseDataset

from llama_index.core.base.response.schema import Response
from llama_index.core.schema import MetadataMode


def to_record(question: str, response: Response, reference: str | None = None) -> dict:
    return {
        "user_input": question,
        "response": str(response),
        "retrieved_contexts": [n.node.get_content(metadata_mode=MetadataMode.LLM) for n in response.source_nodes],
        "reference": reference,
    }

# Ragas (see the Ragas guide for llm_factory and metric setup)
score = await Faithfulness(llm=ragas_llm).ascore(
    user_input=rec["user_input"], response=rec["response"], retrieved_contexts=rec["retrieved_contexts"]
)

# DeepEval
from deepeval import assert_test
from deepeval.metrics import FaithfulnessMetric
from deepeval.test_case import LLMTestCase

case = LLMTestCase(input=rec["user_input"], actual_output=rec["response"], retrieval_context=rec["retrieved_contexts"])
assert_test(case, [FaithfulnessMetric(threshold=0.8)])

import llama_index.core.instrumentation as instrument
from llama_index.core.instrumentation.event_handlers import BaseEventHandler
from llama_index.core.instrumentation.events.llm import LLMChatEndEvent
from llama_index.core.instrumentation.events.retrieval import RetrievalEndEvent
from llama_index.core.instrumentation.span_handlers import SimpleSpanHandler


class UsageHandler(BaseEventHandler):
    """Collects retrieved ids and token usage for each request."""

    retrieved: list = []
    tokens: int = 0

    @classmethod
    def class_name(cls) -> str:
        return "UsageHandler"

    def handle(self, event, **kwargs) -> None:
        if isinstance(event, RetrievalEndEvent):
            self.retrieved.append([n.node.node_id for n in event.nodes])
        elif isinstance(event, LLMChatEndEvent):
            usage = getattr(event.response.raw, "usage", None)           # OpenAI responses carry usage
            self.tokens += usage.total_tokens if usage else 0


handler, spans = UsageHandler(), SimpleSpanHandler()
dispatcher = instrument.get_dispatcher()                               # root dispatcher: sees everything
dispatcher.add_event_handler(handler)
dispatcher.add_span_handler(spans)

query_engine.query("express shipping time")
handler.retrieved, handler.tokens
spans.completed_spans            # SimpleSpan(id_, parent_id, start_time, end_time, duration)
spans.print_trace_trees()        # ASCII tree of the run (needs `treelib`)

import tiktoken
from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager, TokenCountingHandler

counter = TokenCountingHandler(tokenizer=tiktoken.encoding_for_model("gpt-4o-mini").encode)
Settings.callback_manager = CallbackManager([counter])
query_engine.query("express shipping time")
counter.prompt_llm_token_count, counter.completion_llm_token_count, counter.total_embedding_token_count
counter.reset_counts()

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter    # opentelemetry-exporter-otlp-proto-http
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:6006/v1/traces")))
LlamaIndexInstrumentor().instrument(tracer_provider=provider)             # once, before any LlamaIndex call

from llama_index.observability.otel import LlamaIndexOpenTelemetry
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

LlamaIndexOpenTelemetry(
    service_name_or_resource="kb-bot",
    span_exporter=OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces"),
    span_processor="batch",              # "simple" exports each span immediately (tests)
).start_registering()