RAGAS框架：RAG端到端评测

# 简化版 RAGAS 端到端评测# 用迷你语料库模拟知识库, 用关键词重叠模拟检索与生成 import re  class MiniRagas:    def __init__(self, corpus, retriever_topk=2):        self.corpus = corpus  # 知识库: 段落列表        self.topk = retriever_topk        self.stopwords = set(["the", "a", "an", "is", "are", "of", "in",                              "on", "to", "and", "with", "for"])     def tokenize(self, text):        words = re.findall(r"[a-z0-9]+", text.lower())        return [w for w in words if w not in self.stopwords]     def retrieve(self, question):        q_tokens = set(self.tokenize(question))        scored = []        for i, doc in enumerate(self.corpus):            overlap = len(q_tokens & set(self.tokenize(doc)))            scored.append((overlap, i, doc))        scored.sort(key=lambda x: x[0], reverse=True)        return [item[1] for item in scored[:self.topk]]     def generate(self, question, contexts):        for doc in contexts:            first = re.split(r"[.!]", doc)[0]            if len(self.tokenize(first)) > 3:                return first + "."        return "I cannot find enough evidence in the pantry."     def split_claims(self, answer):        return [c.strip() for c in re.split(r"[.!]", answer)                if len(self.tokenize(c.strip())) > 0]     def faithfulness(self, answer, contexts):        claims = self.split_claims(answer)        if not claims:            return 0.0        context_text = " ".join(contexts).lower()        supported = 0        for claim in claims:            key_tokens = [w for w in self.tokenize(claim) if len(w) > 3]            if not key_tokens:                continue            hit = sum(1 for w in key_tokens if w in context_text)            if hit / len(key_tokens) >= 0.5:                supported += 1        return supported / len(claims)     def answer_relevance(self, question, answer):        q_tokens = set(self.tokenize(question))        a_tokens = set(self.tokenize(answer))        if not q_tokens or not a_tokens:            return 0.0        return len(q_tokens & a_tokens) / len(q_tokens | a_tokens)     def context_relevance(self, context, question):        sentences = [s for s in re.split(r"[.!]", context)                     if len(self.tokenize(s)) > 0]        if not sentences:            return 0.0        q_tokens = set(self.tokenize(question))        useful = sum(1 for s in sentences if q_tokens & set(self.tokenize(s)))        return useful / len(sentences)     def evaluate(self, test_set):        metric_map = {"faithfulness": [], "answer_relevance": [],                      "context_relevance": []}        report = []        for item in test_set:            question = item["question"]            gold = item["gold_context"]            retrieved = self.retrieve(question)            retrieved_docs = [self.corpus[i] for i in retrieved]            answer = self.generate(question, retrieved_docs)            metric_map["faithfulness"].append(                self.faithfulness(answer, retrieved_docs))            metric_map["answer_relevance"].append(                self.answer_relevance(question, answer))            metric_map["context_relevance"].append(                self.context_relevance(retrieved_docs[0], question))            oracle_faith = self.faithfulness(answer, [gold])            delta = oracle_faith - metric_map["faithfulness"][-1]            report.append({                "question": question,                "faithfulness": metric_map["faithfulness"][-1],                "answer_relevance": metric_map["answer_relevance"][-1],                "context_relevance": metric_map["context_relevance"][-1],                "oracle_delta": round(delta, 3),            })        summary = {key: sum(vals) / len(vals)                   for key, vals in metric_map.items()}        summary["e2e_score"] = round(            0.4 * summary["faithfulness"] +            0.3 * summary["answer_relevance"] +            0.3 * summary["context_relevance"], 3)        return summary, report  def build_corpus():    return [        "The restaurant pantry stores fresh ingredients, including vegetables,"        " cheese, and smoked fish, organized by daily menu needs.",        "The head chef starts cooking only after the purchaser returns with"        " selected ingredients, following the recipe strictly.",        "A Michelin taster scores the final dish on the table, not the"        " half-finished ingredients in the kitchen.",        "The retrieval module is like a purchaser who picks relevant food from"        " the warehouse for the current order.",        "Faithfulness means every claim on the plate must come from the"        " purchased ingredients, without added fabrication.",        "Answer relevance checks whether the cooked dish actually matches what"        " the customer ordered.",        "Context relevance measures whether the purchased ingredients are truly"        " useful for the final dish.",        "End-to-end evaluation treats purchasing and cooking as one complete"        " service chain from order to table.",    ]  def main():    corpus = build_corpus()    evaluator = MiniRagas(corpus, retriever_topk=2)    test_set = [        {"question": "who scores the final dish in a michelin restaurant?",         "gold_context": corpus[2]},        {"question": "what does faithfulness require for each claim?",         "gold_context": corpus[4]},        {"question": "how does the head chef start cooking?",         "gold_context": corpus[1]},    ]    summary, report = evaluator.evaluate(test_set)    for row in report:        print(row)    print("summary:", summary)    avg_delta = sum(r["oracle_delta"] for r in report) / len(report)    if avg_delta > 0.15:        print("bottleneck: retrieval is the main bottleneck")    else:        print("bottleneck: generation side should be checked next")  if __name__ == "__main__":    main()