pip install deepeval chromadb tiktoken pandas

export OPENAI_API_KEY="sk-xxxxxxx"

text = """
Crows are among the smartest birds, capable of using tools and recognizing human faces even after years.
In contrast, the archerfish displays remarkable precision, shooting jets of water to knock insects off branches.
Meanwhile, in the world of physics, superconductors can carry electric current with zero resistance -- a phenomenon
discovered over a century ago but still unlocking new technologies like quantum computers today.
...
"""

with open("example.txt", "w") as f:
    f.write(text)

from deepeval.synthesizer import Synthesizer

synthesizer = Synthesizer(model="gpt-4.1-nano")

# 從文檔中生成合成數據
synthesizer.generate_goldens_from_docs(
    document_paths=["example.txt"],
    include_expected_output=True
)

# 打印部分結果
for golden in synthesizer.synthetic_goldens[:3]:  
    print(golden, "\n")

Input: Evaluate the cognitive abilities of corvids in facial recognition tasks.
Expected Output: Crows can recognize human faces and remember them for years, showing advanced memory and problem-solving.
Context: "Crows are among the smartest birds..."

from deepeval.synthesizer.config import EvolutionConfig, Evolution

evolution_config = EvolutionConfig(
    evolutions={
        Evolution.REASONING: 1/5,
        Evolution.MULTICONTEXT: 1/5,
        Evolution.COMPARATIVE: 1/5,
        Evolution.HYPOTHETICAL: 1/5,
        Evolution.IN_BREADTH: 1/5,
    },
    num_evolutions=3
)

synthesizer = Synthesizer(evolution_config=evolution_config)
synthesizer.generate_goldens_from_docs(["example.txt"])

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
rag_iterative_eval_full.py
完整示例：迭代評測循環（RAG 改進閉環）
功能：
  - 生成/讀取文檔
  - 生成合成 goldens（DeepEval / OpenAI / 規則化）
  - 構建檢索器（OpenAI embeddings 或 TF-IDF）
  - 使用檢索到的上下文調用 LLM 生成答案（OpenAI 或簡單拼接回復）
  - 計算 grounding / context_relevance / faithfulness 指標
  - 基于指標自動調整 top_k 與 temperature（形成閉環）
  - 保存與打印每輪結果
作者：jilolo
日期：2025-10
"""

import os
import json
import time
import math
import random
import hashlib
from typing import List, Dict, Any, Tuple
from collections import defaultdict, Counter

# optional imports
try:
    import openai
except Exception:
    openai = None

try:
    import numpy as np
    from numpy.linalg import norm
    NUMPY_AVAILABLE = True
except Exception:
    NUMPY_AVAILABLE = False

try:
    from sklearn.feature_extraction.text import TfidfVectorizer
    from sklearn.metrics.pairwise import cosine_similarity
    SKLEARN_AVAILABLE = True
except Exception:
    SKLEARN_AVAILABLE = False

try:
    from tqdm import tqdm
    TQDM_AVAILABLE = True
except Exception:
    TQDM_AVAILABLE = False

# -------------------------
# CONFIG
# -------------------------
CONFIG = {
    "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY", ""),
    "OPENAI_EMBEDDING_MODEL": "text-embedding-3-small",
    "OPENAI_COMPLETION_MODEL": "gpt-4o-mini",  # change to available model
    "DOC_PATH": "example.txt",
    "NUM_GOLDENS": 12,
    "ITERATIONS": 6,
    "INITIAL_TOP_K": 3,
    "MAX_TOP_K": 8,
    "MIN_TOP_K": 1,
    "TEMPERATURE_OPTIONS": [0.0, 0.2, 0.5],
    "SEED": 42,
    "REPORT_FILE": "rag_eval_report.json",
    "SAVE_DIR": "rag_eval_runs",
    "PROMPT_TEMPLATE": (
        "You are a knowledgeable assistant. Use only the provided context snippets to answer the question. "
        "If the information is not present in the context, respond with 'Insufficient information in context.'\n\n"
        "Context:\n{context}\n\nQuestion: {question}\n\nAnswer:"
    ),
    # metric thresholds for increasing/decreasing top_k
    "GROUNDING_GOOD": 0.7,
    "GROUNDING_BAD": 0.45,
    "FAITHFULNESS_GOOD": 0.7,
    "FAITHFULNESS_BAD": 0.45,
    "CONTEXT_RELEVANCE_GOOD": 0.7,
    "CONTEXT_RELEVANCE_BAD": 0.45,
}

random.seed(CONFIG["SEED"])
if openai and CONFIG["OPENAI_API_KEY"]:
    openai.api_key = CONFIG["OPENAI_API_KEY"]

# -------------------------
# Utilities
# -------------------------
def safe_print(*args, **kwargs):
    print(*args, **kwargs)

def ensure_dir(path: str):
    if not os.path.exists(path):
        os.makedirs(path, exist_ok=True)

def sha1_snippet(s: str) -> str:
    return hashlib.sha1(s.encode("utf-8")).hexdigest()[:10]

# -------------------------
# Example document (will write if missing)
# -------------------------
SAMPLE_TEXT = """Crows are among the smartest birds, capable of using tools and recognizing human faces even after years.
The archerfish displays remarkable precision, shooting jets of water to knock insects off branches.
Superconductors can carry electric current with zero resistance -- a phenomenon discovered over a century ago but still unlocking new technologies like quantum computers today.
The Library of Alexandria was once the largest center of learning, but much of its collection was lost in fires and wars.
Voyager 1 probe, launched in 1977, has left the solar system, carrying a golden record with sounds and images of Earth.
The Amazon rainforest produces roughly 20% of the world's oxygen.
Coral reefs support nearly 25% of all marine life despite covering less than 1% of the ocean floor.
MRI scanners use strong magnetic fields and radio waves to generate detailed images of organs without harmful radiation.
Moore's Law observed that the number of transistors on microchips doubles roughly every two years.
The Mariana Trench is the deepest part of Earth's oceans, reaching nearly 11,000 meters below sea level.
Ancient civilizations like the Sumerians and Egyptians invented mathematical systems thousands of years ago.
"""

def ensure_example_doc(path: str):
    if not os.path.exists(path):
        with open(path, "w", encoding="utf-8") as f:
            f.write(SAMPLE_TEXT)
        safe_print(f"[INFO] Wrote sample doc to {path}")

# -------------------------
# Synthetic golden generation (fallback-first approach)
# -------------------------
def simple_rule_based_goldens(doc_path: str, num: int = 12) -> List[Dict[str, str]]:
    """
    Very simple fallback: split document into sentences/paragraphs and craft simple Q/A.
    """
    with open(doc_path, "r", encoding="utf-8") as f:
        txt = f.read()
    paras = [p.strip() for p in txt.split("\n") if p.strip()]
    goldens = []
    for p in paras:
        q = f"What is one key fact from the following sentence: '{p[:120]}...'? "
        a = p
        goldens.append({"input": q, "expected_output": a, "context": p})
        if len(goldens) >= num:
            break
    return goldens

def openai_synthesize_goldens(doc_path: str, num: int = 12, model: str = CONFIG["OPENAI_COMPLETION_MODEL"]) -> List[Dict[str, str]]:
    """
    Try to use OpenAI to synthesize question-answer pairs.
    If OpenAI is not configured or API call fails, fall back to rule-based generation.
    """
    if openai is None or not getattr(openai, "api_key", None):
        safe_print("[WARN] OpenAI key not found - using rule-based goldens")
        return simple_rule_based_goldens(doc_path, num)
    with open(doc_path, "r", encoding="utf-8") as f:
        doc = f.read()

    prompt = (
        f"You are a dataset creator. Given the document below, produce {num} question-answer pairs. "
        f"For each pair, provide 'question', 'answer' (concise and grounded in the doc), and 'context' (the snippet). "
        f"Return a JSON array of objects.\n\nDocument:\n{doc}\n\n"
    )

    try:
        resp = openai.ChatCompletion.create(
            model=model,
            messages=[
                {"role": "system", "content": "You generate QA pairs."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.0,
            max_tokens=1500
        )
        text = resp["choices"][0]["message"]["content"]
        # find JSON in text
        start = text.find("[")
        if start >= 0:
            json_text = text[start:]
            try:
                arr = json.loads(json_text)
                goldens = []
                for item in arr[:num]:
                    q = item.get("question") or item.get("input") or item.get("q") or ""
                    a = item.get("answer") or item.get("expected_output") or ""
                    c = item.get("context") or ""
                    goldens.append({"input": q.strip(), "expected_output": a.strip(), "context": c.strip()})
                safe_print(f"[INFO] OpenAI synthesized {len(goldens)} goldens.")
                return goldens
            except Exception as e:
                safe_print("[WARN] Failed to parse JSON from OpenAI output:", e)
                return simple_rule_based_goldens(doc_path, num)
        else:
            safe_print("[WARN] OpenAI response lacking JSON - using rule-based fallback.")
            return simple_rule_based_goldens(doc_path, num)
    except Exception as e:
        safe_print("[ERROR] OpenAI call failed:", e)
        return simple_rule_based_goldens(doc_path, num)

def generate_goldens(doc_path: str, num: int = 12) -> List[Dict[str, str]]:
    # Attempt DeepEval if installed (not required here); else OpenAI; else rule-based
    # To keep dependencies light in this script we skip DeepEval auto-call.
    return openai_synthesize_goldens(doc_path, num)

# -------------------------
# Retriever: TF-IDF (fallback) and Embedding based (OpenAI)
# -------------------------
class TFIDFRetriever:
    def __init__(self, docs: List[str]):
        if not SKLEARN_AVAILABLE:
            raise RuntimeError("sklearn not available for TF-IDF retriever.")
        self.docs = docs
        self.vectorizer = TfidfVectorizer(ngram_range=(1,2), stop_words='english')
        self.doc_matrix = self.vectorizer.fit_transform(self.docs)

    def retrieve(self, query: str, top_k: int = 3) -> List[Tuple[int, float]]:
        qv = self.vectorizer.transform([query])
        sims = cosine_similarity(qv, self.doc_matrix)[0]
        idx_scores = list(enumerate(sims))
        idx_scores.sort(key=lambda x: x[1], reverse=True)
        return idx_scores[:top_k]

class OpenAIEmbeddingRetriever:
    def __init__(self, docs: List[str], embedding_model: str = CONFIG["OPENAI_EMBEDDING_MODEL"]):
        self.docs = docs
        self.embedding_model = embedding_model
        self.embeddings = []
        # compute embeddings
        self._build()

    def _embed_text(self, text: str):
        if openai is None or not getattr(openai, "api_key", None):
            # fallback: random vector (deterministic via hash)
            if NUMPY_AVAILABLE:
                h = int(hashlib_sha1_int(text))
                rng = np.random.RandomState(h % (2**32))
                return rng.normal(size=(1536,)).tolist()  # fake dim
            else:
                return [random.random() for _ in range(512)]
        try:
            resp = openai.Embedding.create(model=self.embedding_model, input=text)
            return resp["data"][0]["embedding"]
        except Exception as e:
            safe_print("[WARN] OpenAI embedding failed:", e)
            # fallback deterministic pseudo-random
            if NUMPY_AVAILABLE:
                h = int(hashlib_sha1_int(text))
                rng = np.random.RandomState(h % (2**32))
                return rng.normal(size=(1536,)).tolist()
            else:
                return [random.random() for _ in range(512)]

    def _build(self):
        self.embeddings = [self._embed_text(d) for d in self.docs]

    def retrieve(self, query: str, top_k: int = 3) -> List[Tuple[int, float]]:
        q_emb = self._embed_text(query)
        # compute cosine similarities
        if NUMPY_AVAILABLE:
            qv = np.array(q_emb, dtype=float)
            sims = []
            for emb in self.embeddings:
                ev = np.array(emb, dtype=float)
                denom = (norm(qv) * norm(ev))
                sim = float(np.dot(qv, ev) / denom) if denom > 0 else 0.0
                sims.append(sim)
            idx_scores = list(enumerate(sims))
            idx_scores.sort(key=lambda x: x[1], reverse=True)
            return idx_scores[:top_k]
        else:
            sims = []
            for emb in self.embeddings:
                sim = sum(a*b for a,b in zip(q_emb, emb)) / (len(q_emb) or 1)
                sims.append(sim)
            idx_scores = list(enumerate(sims))
            idx_scores.sort(key=lambda x: x[1], reverse=True)
            return idx_scores[:top_k]

# helper hashing for fallback embeddings
def hashlib_sha1_int(s: str) -> int:
    return int(hashlib.sha1(s.encode('utf-8')).hexdigest()[:16], 16)

# -------------------------
# Generator (LLM call) with fallback
# -------------------------
def call_openai_chat(question: str, contexts: List[str], temperature: float = 0.0, model: str = CONFIG["OPENAI_COMPLETION_MODEL"]) -> str:
    if openai is None or not getattr(openai, "api_key", None):
        # fallback: naive rule - if any context contains a sentence with overlap words, return that sentence; else "Insufficient"
        combined = " ".join(contexts)
        q_words = set([w.lower() for w in question.split() if len(w) > 3])
        best_sent = None
        best_overlap = 0
        for s in combined.split("."):
            wset = set([w.lower().strip(" ,;:()[]") for w in s.split() if len(w)>3])
            overlap = len(q_words & wset)
            if overlap > best_overlap:
                best_overlap = overlap
                best_sent = s.strip()
        if best_sent and best_overlap >= 1:
            return best_sent + "."
        return"Insufficient information in context."
    # try call
    prompt = CONFIG["PROMPT_TEMPLATE"].format(context="\n\n".join(contexts), question=question)
    try:
        resp = openai.ChatCompletion.create(
            model=model,
            messages=[
                {"role": "system", "content": "You are a precise assistant."},
                {"role": "user", "content": prompt}
            ],
            temperature=temperature,
            max_tokens=512,
        )
        text = resp["choices"][0]["message"]["content"].strip()
        return text
    except Exception as e:
        safe_print("[WARN] OpenAI ChatCompletion failed:", e)
        # fallback naive
        combined = " ".join(contexts)
        q_words = set([w.lower() for w in question.split() if len(w) > 3])
        best_sent = None
        best_overlap = 0
        for s in combined.split("."):
            wset = set([w.lower().strip(" ,;:()[]") for w in s.split() if len(w)>3])
            overlap = len(q_words & wset)
            if overlap > best_overlap:
                best_overlap = overlap
                best_sent = s.strip()
        if best_sent and best_overlap >= 1:
            return best_sent + "."
        return"Insufficient information in context."

# -------------------------
# Metrics implementations
# -------------------------
def compute_context_relevance(retrieved_idxs_scores: List[Tuple[int, float]]) -> float:
    """
    Simple metric: average similarity score (score between 0-1)
    """
    if not retrieved_idxs_scores:
        return 0.0
    scores = [s for _, s in retrieved_idxs_scores]
    # ensure in [0,1]
    clipped = [max(0.0, min(1.0, float(x))) for x in scores]
    return sum(clipped) / len(clipped)

def compute_grounding(answer: str, contexts: List[str]) -> float:
    """
    Heuristic: fraction of answer tokens that have overlap with context tokens.
    Returns 0-1.
    """
    a_words = [w.strip(" ,.;:()[]'\"").lower() for w in answer.split() if len(w) > 2]
    if not a_words:
        return 0.0
    context_text = " ".join(contexts).lower()
    hits = sum(1 for w in a_words if w in context_text)
    return hits / len(a_words)

def compute_faithfulness(answer: str, expected: str) -> float:
    """
    Very simple normalized similarity:
    - overlap ratio of important tokens (set intersection over union)
    """
    a_set = set([w.strip(" ,.;:()[]'\"").lower() for w in answer.split() if len(w)>2])
    e_set = set([w.strip(" ,.;:()[]'\"").lower() for w in expected.split() if len(w)>2])
    if not a_set and not e_set:
        return 1.0
    if not a_set or not e_set:
        return 0.0
    inter = a_set & e_set
    union = a_set | e_set
    return len(inter) / len(union)

# -------------------------
# Single-run RAG evaluation on list of goldens
# -------------------------
def run_rag_eval(
    goldens: List[Dict[str, str]],
    docs: List[str],
    retriever,
    top_k: int,
    temperature: float
) -> Dict[str, Any]:
    """
    Run through goldens, for each:
      - retrieve top_k contexts
      - call generator
      - compute metrics
    Return aggregated metrics and per-sample results
    """
    per_samples = []
    total_grounding = 0.0
    total_context_rel = 0.0
    total_faith = 0.0

    iterator = goldens if not TQDM_AVAILABLE else tqdm(goldens, desc=f"Eval top_k={top_k}, temp={temperature}")

    for g in iterator:
        q = g["input"]
        expected = g.get("expected_output", "")
        # retrieve
        retrieved = retriever.retrieve(q, top_k=top_k)
        contexts = [docs[idx] for idx, _ in retrieved]
        ctx_scores = [score for _, score in retrieved]

        # call generator
        answer = call_openai_chat(q, contexts, temperature=temperature)

        # compute metrics
        context_rel = compute_context_relevance(retrieved)
        grounding = compute_grounding(answer, contexts)
        faith = compute_faithfulness(answer, expected)

        total_context_rel += context_rel
        total_grounding += grounding
        total_faith += faith

        per_samples.append({
            "question": q,
            "expected": expected,
            "answer": answer,
            "retrieved": [{"idx": idx, "score": float(score), "snippet_hash": sha1_snippet(docs[idx])} for idx, score in retrieved],
            "metrics": {"context_relevance": context_rel, "grounding": grounding, "faithfulness": faith}
        })

    n = len(goldens)
    agg = {
        "avg_context_relevance": total_context_rel / n if n else 0.0,
        "avg_grounding": total_grounding / n if n else 0.0,
        "avg_faithfulness": total_faith / n if n else 0.0
    }
    return {"aggregate": agg, "samples": per_samples}

# -------------------------
# Iterative parameter adjustment logic
# -------------------------
def adjust_params(current_top_k: int, metrics: Dict[str, float]) -> int:
    """
    Very simple policy:
      - If grounding low -> increase top_k (more context)
      - If grounding high and context relevance low -> increase top_k
      - If grounding high & context relevance high -> try reduce top_k to optimize
    Bound by min/max.
    """
    g = metrics.get("avg_grounding", 0.0)
    cr = metrics.get("avg_context_relevance", 0.0)
    fa = metrics.get("avg_faithfulness", 0.0)
    new_top_k = current_top_k

    # if grounding is very low, expand context
    if g < CONFIG["GROUNDING_BAD"]:
        new_top_k = min(CONFIG["MAX_TOP_K"], current_top_k + 2)
    elif cr < CONFIG["CONTEXT_RELEVANCE_BAD"] and g < CONFIG["GROUNDING_GOOD"]:
        new_top_k = min(CONFIG["MAX_TOP_K"], current_top_k + 1)
    elif g > CONFIG["GROUNDING_GOOD"] and cr > CONFIG["CONTEXT_RELEVANCE_GOOD"]:
        # try shrink to save cost
        new_top_k = max(CONFIG["MIN_TOP_K"], current_top_k - 1)
    # small adjustments if faithfulness very low
    if fa < CONFIG["FAITHFULNESS_BAD"]:
        new_top_k = min(CONFIG["MAX_TOP_K"], new_top_k + 1)
    # ensure bounds
    new_top_k = max(CONFIG["MIN_TOP_K"], min(CONFIG["MAX_TOP_K"], new_top_k))
    return new_top_k

def pick_temperature(candidate_list: List[float], metrics: Dict[str, float]) -> float:
    """
    Simple heuristic: if faithfulness low, use lower temp (more deterministic).
    If faithfulness high and grounding high, allow slightly higher temp for diversity.
    """
    fa = metrics.get("avg_faithfulness", 0.0)
    g = metrics.get("avg_grounding", 0.0)
    if fa < 0.4 or g < 0.4:
        return min(candidate_list)
    if fa > 0.75 and g > 0.7:
        return max(candidate_list)
    return candidate_list[len(candidate_list)//2]

# -------------------------
# Main pipeline
# -------------------------
def main():
    safe_print("=== RAG Iterative Evaluation Demo ===")
    ensure_example_doc(CONFIG["DOC_PATH"])
    ensure_dir(CONFIG["SAVE_DIR"])

    # load docs and split into chunks (naive paragraph chunking)
    with open(CONFIG["DOC_PATH"], "r", encoding="utf-8") as f:
        doc_text = f.read()
    paragraphs = [p.strip() for p in doc_text.split("\n") if p.strip()]
    # if paragraphs too short, split sentences
    if len(paragraphs) < 5:
        # attempt sentence split
        sents = [s.strip() for s in doc_text.replace("\n", " ").split(".") if s.strip()]
        # group per 1-2 sentences
        paragraphs = []
        i = 0
        while i < len(sents):
            chunk = sents[i]
            if i+1 < len(sents):
                if random.random() < 0.5:
                    chunk = chunk + ". " + sents[i+1]
                    i += 2
                else:
                    i += 1
            else:
                i += 1
            paragraphs.append(chunk + ".")
    docs = paragraphs

    safe_print(f"[INFO] Loaded {len(docs)} document chunks for retrieval.")

    # generate goldens
    goldens = generate_goldens(CONFIG["DOC_PATH"], CONFIG["NUM_GOLDENS"])
    safe_print(f"[INFO] Generated {len(goldens)} goldens for evaluation.")

    # choose retriever: prefer OpenAI embeddings if available, else TF-IDF
    retriever = None
    use_embedding = False
    if openai and getattr(openai, "api_key", None) and NUMPY_AVAILABLE:
        try:
            retriever = OpenAIEmbeddingRetriever(docs)
            use_embedding = True
            safe_print("[INFO] Using OpenAI embedding retriever.")
        except Exception as e:
            safe_print("[WARN] OpenAIEmbeddingRetriever failed, falling back to TF-IDF:", e)
    if retriever is None:
        if SKLEARN_AVAILABLE:
            retriever = TFIDFRetriever(docs)
            safe_print("[INFO] Using TF-IDF retriever.")
        else:
            # fallback: naive substring search retriever
            class NaiveRetriever:
                def __init__(self, docs):
                    self.docs = docs
                def retrieve(self, query, top_k=3):
                    qs = query.lower()
                    scores = []
                    for i, d in enumerate(self.docs):
                        s = sum(1 for w inset(qs.split()) if w in d.lower())
                        scores.append((i, float(s)))
                    scores.sort(key=lambda x: x[1], reverse=True)
                    return scores[:top_k]
            retriever = NaiveRetriever(docs)
            safe_print("[INFO] Using naive substring retriever.")

    # iterative loop
    cur_top_k = CONFIG["INITIAL_TOP_K"]
    cur_temp = CONFIG["TEMPERATURE_OPTIONS"][0]
    history = []
    for itr in range(1, CONFIG["ITERATIONS"] + 1):
        safe_print(f"\n--- Iteration {itr} | top_k={cur_top_k} | temp={cur_temp} ---")
        result = run_rag_eval(goldens, docs, retriever, top_k=cur_top_k, temperature=cur_temp)
        agg = result["aggregate"]
        safe_print(f"[RESULT] avg_context_relevance={agg['avg_context_relevance']:.3f}, avg_grounding={agg['avg_grounding']:.3f}, avg_faithfulness={agg['avg_faithfulness']:.3f}")
        # save per-iteration
        run_record = {
            "iteration": itr,
            "top_k": cur_top_k,
            "temperature": cur_temp,
            "aggregate": agg,
            "timestamp": time.time(),
            "samples_count": len(result["samples"])
        }
        history.append(run_record)
        # adapt params
        new_top_k = adjust_params(cur_top_k, agg)
        new_temp = pick_temperature(CONFIG["TEMPERATURE_OPTIONS"], agg)
        safe_print(f"[ADAPT] next_top_k={new_top_k}, next_temp={new_temp}")
        # if no change and already good metrics, we can stop early
        if new_top_k == cur_top_k and new_temp == cur_temp and agg["avg_grounding"] > 0.8 and agg["avg_faithfulness"] > 0.8:
            safe_print("[INFO] Metrics are good and stable - stopping early.")
            break
        cur_top_k = new_top_k
        cur_temp = new_temp

    # produce final report
    report = {
        "config": CONFIG,
        "docs_count": len(docs),
        "goldens_count": len(goldens),
        "history": history
    }
    report_path = os.path.join(CONFIG["SAVE_DIR"], CONFIG["REPORT_FILE"])
    with open(report_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    safe_print(f"\n[FINISH] Saved report to {report_path}")
    safe_print("=== End ===")

if __name__ == "__main__":
    main()