eval.py

import asyncio
import json
import logging
import sys
import re
from dataclasses import dataclass, field
from typing import List, Dict, Any

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

from config import settings

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("lcqdrag.eval")


GOLDEN_DATASET: List[Dict[str, str]] = [
    {
        "question": "What is LangSmith?",
        "ground_truth": "LangSmith is a platform by LangChain for building, testing, evaluating, monitoring, and deploying LLM applications with observability into prompts, chains, agents, and retrieval pipelines.",
    },
    {
        "question": "How do I trace a LangChain run in LangSmith?",
        "ground_truth": "To trace a LangChain run, set the LANGCHAIN_TRACING_V2 environment variable to true and configure your LANGCHAIN_API_KEY; runs are then automatically logged to the LangSmith dashboard where you can inspect inputs, outputs, token usage, and latency per step.",
    },
    {
        "question": "What is a LangSmith dataset and how is it used?",
        "ground_truth": "A LangSmith dataset is a collection of labeled examples (inputs plus expected outputs or reference contexts) used to run offline evaluations of chains, agents, or RAG pipelines by comparing model outputs against ground truth with custom or built-in evaluators.",
    },
    {
        "question": "Can I run evaluations on LangSmith and what metrics are supported?",
        "ground_truth": "Yes, LangSmith supports both online and offline evaluation using custom evaluator functions and built-in metrics such as correctness, helpfulness, harmfulness, and reference-free LLM-as-a-judge scorers; results are aggregated per dataset version and compared across runs.",
    },
    {
        "question": "How does LangSmith help with debugging LLM applications?",
        "ground_truth": "LangSmith provides step-by-step tracing of every node in a chain or agent, showing exact inputs, outputs, latency, token counts, tool calls, and intermediate values per step; traces can be filtered by tags, metadata, or user ID and compared against golden runs to identify regressions.",
    },
    {
        "question": "What is the difference between a run and a trace in LangSmith?",
        "ground_truth": "A run is a single invocation of a component (LLM call, chain step, tool, or retriever) with its own inputs, outputs, and timing; a trace is the full tree of runs produced by one top-level execution, linked together by parent-child relationships so you can walk the whole call stack.",
    },
    {
        "question": "Does LangSmith support RAG-specific observability?",
        "ground_truth": "Yes, LangSmith captures retriever invocations as retrieval runs with the query, returned document chunks and their scores, metadata filters, and latency; downstream generation runs reference those retrieval events so end-to-end RAG fidelity can be audited and measured.",
    },
    {
        "question": "How can I compare two chain versions in LangSmith?",
        "ground_truth": "Upload or create a dataset, run evaluation experiments on each chain version against that dataset using the LangSmith SDK or UI, then open the Experiments tab to compare aggregate scores, per-example deltas, statistical significance, and side-by-side trace views.",
    },
    {
        "question": "What are LangSmith prompts and prompt versioning?",
        "ground_truth": "LangSmith prompts are versioned, named prompt templates stored in the Prompt Hub with metadata like tags and owner; each version captures the exact template string, input variables, and model config so applications can pin, A/B test, or roll back prompts without code changes.",
    },
    {
        "question": "How do I add custom metadata to a LangSmith trace?",
        "ground_truth": "Attach metadata to your runs using the RunnableConfig or by passing a metadata dictionary when invoking your chain; metadata fields are indexed and filterable in the LangSmith UI, letting you slice traces by user ID, session, feature flag, environment, or any custom dimension.",
    },
    {
        "question": "Is LangSmith open source?",
        "ground_truth": "LangSmith itself is a commercial platform offered by LangChain with both free and paid tiers; however, the tracing SDKs and the core LangChain framework that emits traces are open source, and traces can also be exported to external observability systems via OpenTelemetry.",
    },
    {
        "question": "How does LangSmith handle production monitoring and alerts?",
        "ground_truth": "LangSmith continuously ingests production traces and computes dataset-level and aggregate metrics over sliding windows; you can configure alerts on SLOs such as latency percentiles, error rates, or evaluator score thresholds delivered via email, Slack, or webhooks.",
    },
]


@dataclass
class EvalRow:
    question: str
    ground_truth: str
    answer: str = ""
    contexts: List[str] = field(default_factory=list)


async def _load_vectorstore_and_retriever():
    from langchain_ollama import OllamaEmbeddings
    from langchain_qdrant import QdrantVectorStore
    from qdrant_client import QdrantClient

    embeddings = OllamaEmbeddings(model=settings.models.embedding_model)
    client = QdrantClient(url=settings.qdrant.url)
    vectorstore = QdrantVectorStore(
        client=client,
        collection_name=settings.qdrant.collection_name,
        embedding=embeddings,
    )
    return vectorstore, embeddings


async def _retrieve_contexts(vectorstore, question: str) -> List[str]:
    from qdrant_client import QdrantClient

    client = QdrantClient(url=settings.qdrant.url)
    embeddings = vectorstore._embeddings if hasattr(vectorstore, "_embeddings") else vectorstore.embeddings
    query_embedding = await embeddings.aembed_query(question)

    search_k = settings.retrieval.top_k_final
    results = client.search(
        collection_name=settings.qdrant.collection_name,
        query_vector=query_embedding,
        limit=search_k,
        with_payload=True,
    )
    return [r.payload.get("page_content", "") for r in results]


async def _generate_answer(llm, question: str, contexts: List[str]) -> str:
    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
    from langchain_core.messages import HumanMessage

    qa_system_prompt = (
        "You are an expert AI assistant. Answer the user's question using ONLY the provided context below.\n"
        "CRITICAL RULE: Do NOT include any '<think>' tags, internal reasoning, or chain-of-thought blocks in your output. "
        "Provide ONLY the direct, final answer to the user's question.\n\n"
        "If you do not know the answer or if it's not in the context, say 'I am sorry, but I do not have that information.'\n\n"
        "<context>\n{context}\n</context>"
    )
    qa_prompt = ChatPromptTemplate.from_messages([
        ("system", qa_system_prompt),
        ("human", "{question}"),
    ])

    context_text = "\n\n".join(contexts)
    chain = qa_prompt | llm
    response = await chain.ainvoke({
        "context": context_text,
        "question": question,
    })
    clean_answer = re.sub(r'<think>.*?</think>', '', response.content, flags=re.DOTALL).strip()
    return clean_answer


async def run_evaluation() -> int:
    logger.info(f"Starting RAGAS evaluation with {len(GOLDEN_DATASET)} golden examples")
    logger.info(f"Thresholds — faithfulness >= {settings.eval.faithfulness_threshold}, "
                f"relevancy >= {settings.eval.relevancy_threshold}")

    try:
        from langchain_groq import ChatGroq
        from langchain_ollama import ChatOllama
    except ImportError as e:
        logger.error(f"Missing imports: {e}")
        return 2

    vectorstore, embeddings = await _load_vectorstore_and_retriever()

    try:
        generator_llm = ChatGroq(
            model_name=settings.models.llm_model,
            temperature=0.0,
        )
        await generator_llm.ainvoke("ping")
    except Exception:
        logger.warning("Groq generator LLM unavailable, falling back to Ollama llama3.1:8b")
        generator_llm = ChatOllama(model="llama3.1:8b", temperature=0.0)

    evaluator_llm = generator_llm

    rows: List[EvalRow] = [EvalRow(question=item["question"], ground_truth=item["ground_truth"]) for item in GOLDEN_DATASET]

    for i, row in enumerate(rows):
        logger.info(f"  [{i+1}/{len(rows)}] Evaluating: {row.question[:80]}...")
        try:
            row.contexts = await _retrieve_contexts(vectorstore, row.question)
            row.answer = await _generate_answer(generator_llm, row.question, row.contexts)
        except Exception as e:
            logger.error(f"    Error processing example {i+1}: {e}")
            row.answer = ""
            row.contexts = []

    data_dict = {
        "question": [r.question for r in rows],
        "answer": [r.answer for r in rows],
        "contexts": [r.contexts for r in rows],
        "ground_truth": [r.ground_truth for r in rows],
    }
    dataset = Dataset.from_dict(data_dict)

    try:
        result = evaluate(
            dataset=dataset,
            metrics=[faithfulness, answer_relevancy],
            llm=evaluator_llm,
            embeddings=embeddings,
            raise_exceptions=False,
        )
    except Exception as e:
        logger.error(f"RAGAS evaluation failed: {e}", exc_info=True)
        return 3

    logger.info("\n===== RAGAS Evaluation Results =====")
    df = result.to_pandas()
    for i, row in df.iterrows():
        logger.info(f"  [{i+1}] Q: {str(row.get('question', ''))[:80]}")
        logger.info(f"       Faithfulness: {row.get('faithfulness', 'N/A')} | "
                    f"Answer Relevancy: {row.get('answer_relevancy', 'N/A')}")
        logger.info(f"       Answer: {str(row.get('answer', ''))[:160]}")
        logger.info("")

    overall = {}
    try:
        for metric_name in ["faithfulness", "answer_relevancy"]:
            if metric_name in result:
                overall[metric_name] = float(result[metric_name])
    except Exception:
        overall = {}

    logger.info("===== Overall Scores =====")
    for k, v in overall.items():
        logger.info(f"  {k}: {v:.4f}")

    failed_metrics = []
    if "faithfulness" in overall and overall["faithfulness"] < settings.eval.faithfulness_threshold:
        failed_metrics.append(f"faithfulness={overall['faithfulness']:.4f} < {settings.eval.faithfulness_threshold}")
    if "answer_relevancy" in overall and overall["answer_relevancy"] < settings.eval.relevancy_threshold:
        failed_metrics.append(f"answer_relevancy={overall['answer_relevancy']:.4f} < {settings.eval.relevancy_threshold}")

    summary_path = "eval_results.json"
    with open(summary_path, "w") as f:
        json.dump({
            "thresholds": {
                "faithfulness": settings.eval.faithfulness_threshold,
                "answer_relevancy": settings.eval.relevancy_threshold,
            },
            "overall": overall,
            "per_example": data_dict,
            "failed": failed_metrics,
        }, f, indent=2, default=str)
    logger.info(f"Full results written to {summary_path}")

    if failed_metrics:
        logger.error(f"EVALUATION FAILED — metrics below threshold: {'; '.join(failed_metrics)}")
        return 1

    logger.info("EVALUATION PASSED — all metrics meet thresholds")
    return 0


def main():
    try:
        exit_code = asyncio.run(run_evaluation())
    except KeyboardInterrupt:
        logger.info("Evaluation interrupted by user")
        sys.exit(130)
    except Exception as e:
        logger.error(f"Evaluation crashed: {e}", exc_info=True)
        sys.exit(99)
    sys.exit(exit_code)


if __name__ == "__main__":
    main()