DSPy 3 + GEPA：迄今最先进的 RAG 框架——自动推理与提示

上周，OpenAI 在半夜突发“飙车”，一度全网骚动。GPT-5.2 发布，全球 AI 王座再度易主。

四个月左右就迎来一次大版本更新并不常见。导火索是竞争压力。路透社报道(https://www.reuters.com/technology/openai-launches-gpt-52-ai-model-with-improved-capabilities-2025-12-11/) 称，Altman 在 12 月初下达了“code red”加速研发，而其背景正是应对 Google 的 Gemini 3。

OpenAI 本身也将其定位为：“（不是堆新功能）我们在 intelligence、code processing、long-form text comprehension 等方面显著提升，尤其擅长创建 spreadsheets、制作 presentations 以及其他复杂的多步骤任务。”

换句话说，GPT-5.2 不是“颠覆性大改”，而是一次面向实用场景的“精修版”：在 reliability、long-term context、tool execution、output generation 等维度全面加强。与其说是新玩具，不如说是更顺手的工作利器。

近年来，“Agentic AI” 正在执行一系列复杂动作：LLM 调用工具、推理，最后给出答案。为了优化这些动作，传统范式通常依赖 reinforcement learning（RL，强化学习）来“用 reward 学到好动作”。但问题在于：

• RL 只给出一个简单的标量 reward（答案对错），学习极其低效。
• 此外，fine-tuning 需要大量 rollout 与计算成本。

去年我做过一个关于 DSPy 的视频，之后它进步很大。DSPy 的核心思想是把语言模型视为一种“device”，类似深度学习中的 CPU、GPU。

在 DSPy 中，你只需声明所需的 “Natural Language Signatures”，而不必纠结 Prompt 的具体实现细节（事实上，一年实践后我们发现，过度纠结这些细节意义有限，LLM 输出的不稳定性并不会因此本质改变）。

可以这样理解 DSPy：基于这些 signatures，DSPy 会自动生成、优化、微调 Prompt，最终产出符合预期的结果。

GEPA 的想法：鼓励 LLM “反思自己的失败”。不同于 reinforcement learning，GEPA（Genetic-Pareto Prompt Optimizer）采用一种方式：让 LLM 自己用 natural language 分析自身行为，并提出下次如何改进。也就是说，与其调整模型参数，我们更是去“反思并进化”prompt 本身。

我会准备一个 SPACE_KNOWLEDGE。这种技术是一种替代训练方式，效果优于 reinforcement learning，并提出一个与太空有关的问题：“Which space telescope is most powerful?” 当你观察这个 chatbot 如何生成输出时，

你会看到 agent 使用 Term Frequency Inverse Document Frequency 来计算 term frequency（一个词在文档中出现的频率，以及它在所有文档中出现的稀有程度），然后用 cosine similarity 找出哪些 chunks 与你的问题真正相似，而不是仅仅出现了随机词匹配。检索出最相关的前三个 chunks 之后，

Agent 会用 confidence-based RAG 结合 chain-of-thought 生成答案与一个 confidence level，从而能坦诚地说“I don't have enough information”，而不是 hallucinate。与此同时，multi-hop RAG 会更进一步：先从 context 中抽取 bullet-pointed facts，再把这些事实综合为完整答案——这个两步流程对需要跨多来源整合信息的复杂问题至关重要，因为它能避免 AI Agent 混乱或漏掉关键关联。

接下来是 GEPA 的“杀手锏”：它不是手动调 prompt，也不是用过时的优化器（如 MIPROv2），而是采用 genetic algorithms。它会“组合优秀的 prompts”，生成更佳方案。

它引入 Pareto optimization，不是只保留一个 prompt，而是维护多个有效 prompt。同时，GEPA 引入 reflection：通过阅读 textual feedback，从错误中学习并修正。随着时间推进，GEPA 会自动产生越来越好的 prompts。

它会构建一棵 prompt evolution tree。每一次改进都像树上长出新的 branch。每个分支保留之前有效的做法，再做增量改良。一步步地，这些 prompts 会越来越接近适配 RAG 任务的“最佳指令”。而且，GEPA 比 MIPROv2 效率高 35 倍，生成的 prompts 长度更短（9 倍），表现还提升 10%。

GPT-5.2 有何过人之处？

先看最“爆”的数据。常用来评估 AI 能力的一个测试叫 “ARC-AGI-2”。

这是一个“抽象谜题即刻解（灵感）”的测试，不依赖“从过往数据里找答案（作弊）”。换言之，它测的是“先天智能”。看看分数：GPT-5.1：17.6%，Gemini 3 Pro：31.1%，GPT-5.2：52.9%（+35.3 分！）

这个提升太夸张了。是 5.1 的三倍多，几乎是 Gemini 的两倍。

如果说之前的 AI 更像“死记硬背教科书的天才”，那 GPT-5.2 已经进化为“能用巧思解决从未见过难题的天才”。那句常见的 AI 托词“我没学过所以不会”，正在成为过去式。

另一个值得关注的指标是 “GDPval”。它衡量的是在 research、planning、decision-making 等“真实世界任务”上的能力。GPT-5.1：38.8%，Gemini 3 Pro：53.5%，GPT-5.2：70.9%（+32.1 分！）

结果再次碾压。在 5.1 时，AI 还像是“等指令的新手实习生”；到了 5.2，则已升级为“能制定计划并管理项目的 manager”。

过去抱怨“AI 很聪明，但在工作中不好用”的人，会被 5.2 的“上岗能力”惊到。

GEPA 的独特之处？

GEPA 的核心来自人类学习的本质——reflection。

它不是“堆更多指令”，而是像经验丰富的导师，回看过往尝试，分析成功与不足，然后给出更好的解法。

GEPA 会构建一棵 prompt evolution tree，让每次优化像长枝一样累积改进，逐步逼近最优 prompt。

不同于传统的 reinforcement learning（RL），GEPA 利用 LLM 的反思能力，并结合 domain-specific textual feedback，而不是只依赖单一标量指标。

这就好比给模型一双“X 光之眼”，能洞察任务中的细微之处，仅用少量步骤就产出更强结果。

Let's start coding :

现在一步步看如何用 DSPy 3、GEPA Optimiser 与 Agentic RAG。我们先安装支撑所需的库。为此，执行 pip install requirements。

我需要说明，这里分享的只是部分代码。如需完整文件夹，可在我的 Patreonhttps://www.patreon.com/c/GaoDalie_AI获取。这份代码我花了相当多时间。

pip install requirements

Term Frequency Inverse Document Frequency.

我实现了一个 Term Frequency Inverse Document Frequency retriever，用来找到与用户问题最匹配的文档。首先，它存储所有文档，并将每个文档切分成简单的小写词项，去掉标点，使文本更干净、更易比较。

接着，它在全量语料上计算每个词的重要性：出现在很多文档里的词整体重要性更低，而只出现在少数文档里的词更重要。

当有查询进来时，会按同样方式清洗并分词，并基于词频与整体稀有度为每个词赋分。

retriever 随后将查询与每份文档进行比较，使用一种数学方法来衡量两者向量指向是否一致（cosine similarity）。

每个文档都会得到一个相似度分数，按从高到低排序，最后返回最相关的前若干文档给用户。

class TFIDFRetriever:
    """
    TF-IDF (Term Frequency - Inverse Document Frequency) retriever.
    
    This is smarter than simple keyword matching because:
    - TF: Words that appear often in a document are important for that document
    - IDF: Words that appear in many documents are less important overall
    
    Example: "the" appears everywhere (low IDF), but "astronaut" is specific (high IDF)
    """
    
    def __init__(self, documents: list[str], k: int = 3):
        self.documents = documents
        self.k = k
        self.doc_tokens = [self._tokenize(doc) fordocin documents]
        self.idf = self._compute_idf()
    
    def _tokenize(self, text: str) -> list[str]:
        """Convert text to lowercase tokens, removing punctuation."""
        import re
        text = text.lower()
        tokens = re.findall(r'\b[a-z]+\b', text)
        return tokens
    
    def _compute_idf(self) -> dict[str, float]:
        """Compute IDF for all terms in the corpus."""
        doc_count = len(self.documents)
        term_doc_counts = Counter()
        
        fortokensinself.doc_tokens:
            unique_tokens = set(tokens)
            fortokenin unique_tokens:
                term_doc_counts[token] += 1
        
        idf = {}
        forterm, count in term_doc_counts.items():
            # Standard IDF formula with smoothing
            idf[term] = math.log((doc_count + 1) / (count + 1)) + 1
        
        return idf
    
    def _compute_tfidf(self, tokens: list[str]) -> dict[str, float]:
        """Compute TF-IDF vector for a list of tokens."""
        tf = Counter(tokens)
        tfidf = {}
        forterm, count in tf.items():
            tfidf[term] = count * self.idf.get(term, 1.0)
        return tfidf
    
    def _cosine_similarity(self, vec1: dict, vec2: dict) -> float:
        """Compute cosine similarity between two sparse vectors."""
        common_terms = set(vec1.keys()) & set(vec2.keys())
        if not common_terms:
            return0.0
        
        dot_product = sum(vec1[t] * vec2[t] fortin common_terms)
        norm1 = math.sqrt(sum(v ** 2forvin vec1.values()))
        norm2 = math.sqrt(sum(v ** 2forvin vec2.values()))
        
        if norm1 == 0 or norm2 == 0:
            return0.0
        
        return dot_product / (norm1 * norm2)
    
    def __call__(self, query: str) -> list[str]:
        """Retrieve top-k documents most similar to the query."""
        query_tokens = self._tokenize(query)
        query_vec = self._compute_tfidf(query_tokens)
        
        scores = []
        fori, doc_tokens inenumerate(self.doc_tokens):
            doc_vec = self._compute_tfidf(doc_tokens)
            score = self._cosine_similarity(query_vec, doc_vec)
            scores.append((score, i, self.documents[i]))
        
        # Sort by score descending
        scores.sort(key=lambda x: x[0], reverse=True)
        
        return [doc forscore, idx, doc in scores[:self.k]]

Retrieve Argumentation Generation：

随后，我实现了两种基于 retrieval augmentation generation 的问答方式。第一种中，Agent 会接收问题，检索最相关的文档，将它们拼接为一个 context，然后生成答案，同时给出置信度。

它会保存使用过的文档，便于回溯答案来源。第二种针对需要更多推理的难题。

它先同样检索文档，然后只抽取与问题相关的关键 facts，最后再把这些 facts 组合成清晰的答案。

它还会同时保留已检索的文档与抽取的 facts，便于你检查每一步，理解最终答案是如何构建出来的。

class RAGWithConfidence(dspy.Module):
    """RAG that reports its confidence in the answer."""
    
    def__init__(self, retriever):
        super().__init__()
        self.retriever = retriever
        self.generate = dspy.ChainOfThought(AnswerWithConfidence)
    
    defforward(self, question: str):
        docs = self.retriever(question)
        context = "\n\n".join(docs)
        result = self.generate(context=context, question=question)
        result.retrieved_docs = docs
        return result


classMultiHopRAG(dspy.Module):
    """
    Multi-hop RAG: Extract facts first, then synthesize an answer.
    
    This helps with complex questions that require combining information
    from multiple sources.
    """
    
    def__init__(self, retriever):
        super().__init__()
        self.retriever = retriever
        self.extract = dspy.Predict(ExtractFacts)
        self.synthesize = dspy.Predict(SynthesizeAnswer)
    
    defforward(self, question: str):
        # Step 1: Retrieve
        docs = self.retriever(question)
        context = "\n\n".join(docs)
        
        # Step 2: Extract relevant facts
        extraction = self.extract(context=context, question=question)
        
        # Step 3: Synthesize answer from facts
        result = self.synthesize(facts=extraction.facts, question=question)
        
        # Attach intermediate results for inspection
        result.retrieved_docs = docs
        result.extracted_facts = extraction.facts
        
        return result

Reflective Prompt Evolution：

然后我用 GEPA 让系统逐步学习与改进答案。首先，metric 会用期望答案来评估模型输出：完全匹配则满分。

若只有部分正确，就给较低分，并解释缺失了什么。如果答案错误，则给低分并提供清晰的错误反馈。

这些反馈非常关键，因为 GEPA 会读取它们，并据此改进后续 prompts。Simple RAG 模块的工作流是：接收问题，检索相关文档，将其拼接为一个 context，并基于该 context 生成答案。

GEPA 会利用 metric 的分数与反馈，不断自动进化这个 RAG 系统的 prompts。

def gepa_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    """
    GEPA metric function with feedback.
    
    GEPA is special because it can use textual feedback to guide evolution.
    This function returns both a score AND feedback about what went wrong.
    """
    expected = gold.expected_answer.lower()
    actual = pred.answer.lower() ifhasattr(pred, 'answer') else""
    
    # Check if the key information is in the answer
    if expected in actual:
        return1.0# Perfect match
    
    # Partial credit for relevant answers
    expected_words = set(expected.split())
    actual_words = set(actual.split())
    overlap = len(expected_words & actual_words) / len(expected_words) if expected_words else0
    
    if overlap > 0.5:
        score = 0.7
        feedback = f"Partially correct. Expected '{gold.expected_answer}' but got related content."
    elif overlap > 0:
        score = 0.3
        feedback = f"Contains some relevant info but missing key details. Expected: '{gold.expected_answer}'"
    else:
        score = 0.0
        feedback = f"Incorrect. Expected answer to contain '{gold.expected_answer}' but got: '{actual[:100]}...'"
    
    # Return score with feedback for GEPA's reflection
    from dspy.teleprompt.gepa.gepa_utils import ScoreWithFeedback
    return ScoreWithFeedback(score=score, feedback=feedback)


classSimpleRAGForOptimization(dspy.Module):
    """A simple RAG module that GEPA will optimize."""
    
    def__init__(self, retriever):
        super().__init__()
        self.retriever = retriever
        self.generate = dspy.Predict("context, question -> answer")
    
    defforward(self, question: str):
        docs = self.retriever(question)
        context = "\n\n".join(docs)
        returnself.generate(context=context, question=question)

我的想法：

GPT-5.2 也许不是能做“全新魔法”的模型，但它能把“原本不敢交给 AI 的任务”，变成“可以放心委托的任务”。

未来仍有挑战，比如 multimodal 支持、real-time optimization、安全保障，但这同样意味着巨大的发展空间。

在 2026 之后，GEPA 有望催生诸如 self-correcting AI systems、neural-symbolic integration、meta-prompt engineering 等创新应用。GEPA 无疑会继续在 prompt 技术的未来中扮演核心角色。