第一部分：DSPy自动优化

1.1 DSPy的核心创新

传统方式的问题：

Prompt = 字符串
优化 = 手动改字符串
评估 = 靠感觉

DSPy的突破：

Prompt = 可优化的参数
优化 = 搜索算法
评估 = 量化函数

类比深度学习：

• 神经网络权重 ≈ DSPy的Prompt
• 梯度下降 ≈ DSPy的优化器
• 损失函数 ≈ DSPy的评估函数

1.2 三步实现DSPy优化

Step 1：定义Program

import dspy

class GenerationAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        # 定义输入→输出
        self.generate = dspy.ChainOfThought(
            "query, context -> recommendation"
        )
    
    def forward(self, query, context):
        result = self.generate(query=query, context=context)
        return result.recommendation

Step 2：定义评估函数

def metric(example, pred, trace=None):
    # 相关性
    relevance = calculate_relevance(example.query, pred.recommendation)
    # 忠实度
    faithfulness = calculate_faithfulness(pred.recommendation, example.context)
    
    return relevance * 0.5 + faithfulness * 0.5

Step 3：自动优化

from dspy.teleprompt import BootstrapFewShot

optimizer = BootstrapFewShot(metric=metric, max_bootstrapped_demos=4)
optimized_agent = optimizer.compile(student=GenerationAgent(), trainset=trainset)

1.3 效果对比

指标	手动Prompt	DSPy优化	提升
准确率	72%	83%	+15%
调优时间	2小时	5分钟	-96%

注意：不是所有场景都能提升这么多，取决于：

• 训练数据质量（我用了200条）
• 评估函数设计
• 任务复杂度

1.4 我踩过的坑

坑1：训练数据太少

# 50条训练数据：准确率68%（比手动还差）
# 100条训练数据：准确率75%
# 200条训练数据：准确率83% ✓

教训：至少需要100-200条训练数据。

坑2：Few-shot示例过多

# 测试不同数量的示例
4个示例：准确率83% ← 最佳
8个示例：准确率81%
16个示例：准确率76%（过拟合）

教训：4-8个示例最佳。

坑3：评估函数设计不当

# 错误：只看准确率
def bad_metric(example, pred):
    return int(pred.recommendation == example.answer)

# 正确：多维度评估
def good_metric(example, pred):
    relevance = calculate_relevance(example.query, pred.recommendation)
    faithfulness = calculate_faithfulness(pred.recommendation, example.context)
    completeness = calculate_completeness(pred.recommendation, example.required_fields)
    
    return relevance * 0.4 + faithfulness * 0.4 + completeness * 0.2

教训：评估函数要全面，否则优化方向会偏。

1.5 完整实现示例

import dspy
from typing import List, Dict

class HouseRecommendationAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate = dspy.ChainOfThought(
            "query, context, memory -> recommendation, reasoning"
        )
    
    def forward(self, query: str, context: List[Dict], memory: str):
        # 格式化上下文
        context_str = "\n".join([
            f"房源{i+1}: {doc['title']}, {doc['price']}, {doc['location']}"
            for i, doc in enumerate(context)
        ])
        
        # 生成推荐
        result = self.generate(
            query=query,
            context=context_str,
            memory=memory
        )
        
        return {
            "recommendation": result.recommendation,
            "reasoning": result.reasoning
        }

# 训练数据准备
trainset = [
    dspy.Example(
        query="3室2厅，预算500万，靠近地铁",
        context=[...],  # 检索到的房源
        memory="用户偏好：现代风格，重视交通",
        answer="推荐房源A，理由：..."
    ).with_inputs("query", "context", "memory"),
    # ... 更多示例
]

# 优化
optimizer = dspy.teleprompt.BootstrapFewShot(
    metric=good_metric,
    max_bootstrapped_demos=4,
    max_labeled_demos=4
)

optimized_agent = optimizer.compile(
    student=HouseRecommendationAgent(),
    trainset=trainset
)

# 保存模型
optimized_agent.save("models/optimized_agent.json")

1.6 关键洞察

💡 洞察1：Prompt是可优化的参数，不是固定字符串

💡 洞察2：Few-shot示例的选择比数量更重要（4个精选 > 16个随机）

💡 洞察3：评估函数决定优化方向（多指标平衡很重要）

💡 洞察4：训练数据质量最重要（至少100-200条）

💡 洞察5：不是所有场景都适合DSPy（简单任务手动Prompt就够了）

---

第二部分：Agentic RAG智能检索

2.1 三大核心技术

技术1：Self-RAG（检索后反思）

核心思想：检索后评估质量，过滤无关文档

class SelfRAG:
    def retrieve_and_reflect(self, query):
        # 1. 检索
        documents = self.retriever.retrieve(query)
        
        # 2. 反思：评估相关性
        relevance_scores = self._score_relevance(query, documents)
        
        # 3. 过滤
        relevant_docs = [doc for doc, score in zip(documents, relevance_scores)
                        if score > 0.7]
        
        return relevant_docs

效果：

• 检索相关性：0.72 → 0.85（+18%）
• 过滤掉约60%的无关文档

技术2：Corrective RAG（查询改写）

核心思想：检索质量低时，改写查询重试

class CorrectiveRAG:
    def retrieve_with_correction(self, query, max_attempts=2):
        for attempt in range(max_attempts):
            documents = self.retriever.retrieve(query)
            quality = self._evaluate_quality(query, documents)
            
            if quality > 0.75:
                return documents
            
            # 改写查询
            query = self._rewrite_query(query)
        
        return documents

效果：

• 检索成功率：75% → 87%（+16%）
• 平均重试1.2次

技术3：Adaptive Routing（自适应策略）

核心思想：根据查询复杂度选择策略

class AdaptiveRetriever:
    def retrieve(self, query):
        complexity = self._analyze_complexity(query)
        
        if complexity == "简单":
            return self._fast_retrieve(query, top_k=5)
        else:
            return self._deep_retrieve(query, top_k=20)

效果：

• 简单查询：响应时间-40%
• 复杂查询：准确率+25%

2.2 整体效果

方法	准确率	响应时间
传统RAG	72%	650ms
+ Self-RAG	78%	700ms
+ Corrective RAG	81%	600ms
+ Adaptive Routing	83%	480ms

关键发现：

• Self-RAG贡献最大（+6%）
• Corrective RAG提升稳定性（+3%）
• Adaptive Routing优化性能（+2%，-120ms）

2.3 我踩过的坑

坑4：重试次数过多

# 1次重试：成功率+12%，响应时间+50ms
# 2次重试：成功率+16%，响应时间+100ms ✓
# 3次重试：成功率+16%，响应时间+200ms（边际收益为0）

教训：最多重试2次。

坑5：Self-RAG阈值太高

# 阈值0.9：过滤90%文档，只剩1个 ✗
# 阈值0.7：过滤60%文档，剩5-8个 ✓
# 阈值0.5：过滤30%文档，效果不明显

教训：阈值0.7左右最佳。

坑6：查询改写过度

# 原始查询："3室2厅，靠近地铁"
# 改写1次："三居室两厅，地铁沿线" ✓
# 改写2次："多卧室住宅，公共交通便利" ✗（语义漂移）

教训：改写要保持原意，不能过度泛化。

2.4 完整实现示例

class AgenticRAGPipeline:
    def __init__(self, retriever, rewriter, verifier):
        self.retriever = retriever
        self.rewriter = rewriter
        self.verifier = verifier
    
    def retrieve(self, query: str, max_attempts: int = 2):
        """完整的Agentic RAG流程"""
        
        # 1. 分析查询复杂度
        complexity = self._analyze_complexity(query)
        top_k = 20 if complexity == "复杂" else 5
        
        # 2. 检索 + 重试
        for attempt in range(max_attempts):
            documents = self.retriever.retrieve(query, top_k=top_k)
            
            # 3. Self-RAG：评估相关性
            scored_docs = []
            for doc in documents:
                score = self.verifier.score_relevance(query, doc)
                if score > 0.7:
                    scored_docs.append((doc, score))
            
            # 4. 检查质量
            if len(scored_docs) >= 3:
                # 按分数排序
                scored_docs.sort(key=lambda x: x[1], reverse=True)
                return [doc for doc, _ in scored_docs[:5]]
            
            # 5. Corrective RAG：改写查询
            if attempt < max_attempts - 1:
                query = self.rewriter.rewrite(query, documents)
                print(f"改写查询: {query}")
        
        # 6. 兜底：返回最佳结果
        return [doc for doc, _ in scored_docs[:3]] if scored_docs else documents[:3]
    
    def _analyze_complexity(self, query: str) -> str:
        """分析查询复杂度"""
        # 多条件 = 复杂
        conditions = len([w for w in ["预算", "位置", "面积", "户型", "楼层"] if w in query])
        return "复杂" if conditions >= 3 else "简单"

# 使用示例
pipeline = AgenticRAGPipeline(retriever, rewriter, verifier)
results = pipeline.retrieve("3室2厅，预算500万，靠近地铁，楼层不要太高")

2.5 关键洞察

💡 洞察1：检索不是一次性的，需要反思和调整

💡 洞察2：检索质量比数量更重要（5个高质量 > 20个低质量）

💡 洞察3：不同查询需要不同策略（简单查询快速检索，复杂查询深度检索）

💡 洞察4：重试次数要控制（1-2次最佳）

💡 洞察5：性能和准确率要平衡（不能为了准确率牺牲太多性能）

---

第三部分：记忆系统集成

3.1 集成要点

记忆系统与DSPy集成：

# 记忆上下文作为DSPy的输入
memory_context = build_memory_context(memory_manager, entity_memory, bias_manager)

result = optimized_agent(
    query=query,
    context=context,
    memory=memory_context  # 包含用户偏好、历史对话
)

记忆系统与Agentic RAG集成：

• 检索记忆：缓存结果，避免重复检索
• 偏置记忆：个性化检索策略
• 实体记忆：指代消解

3.2 效果

指标	无记忆	有记忆	提升
多轮对话准确率	68%	81%	+19%
重复推荐率	38%	0%	-100%
响应时间	650ms	480ms	-26%

---

第四部分：实验结果

4.1 整体效果

指标	基线	优化后	提升
整体准确率	72%	83%	+15%
复杂查询准确率	65%	81%	+25%
检索相关性	0.72	0.85	+18%
响应时间	650ms	480ms	-26%
重复推荐率	38%	0%	-100%

说明：

• 数据基于200条训练集、50条测试集
• 运行环境：本地CPU（16GB内存）
• API成本：约$3（使用GPT-3.5）

4.2 消融实验

配置	准确率	提升
基线	72%	-
+ DSPy	77%	+5%
+ Self-RAG	80%	+3%
+ Corrective RAG	82%	+2%
+ 记忆系统	83%	+1%

结论：

• DSPy贡献最大（+5%）
• Self-RAG是核心（+3%）
• 其他技术锦上添花（+3%）

4.3 实战经验

经验1：优化器选择

• 简单任务用BootstrapFewShot（快10倍）
• 复杂任务用MIPROv2（准5%）

经验2：缓存策略

• 必须缓存优化结果
• 否则每次都重新优化（耗时5分钟）
• 方案：保存为JSON文件，TTL=7天

经验3：记忆容量控制

• 短期记忆：5轮
• 实体记忆：最近5个
• 总Token：<2000

---

总结

从手动调优到自动优化，从被动检索到主动思考，我们的系统经历了质的飞跃。

核心收获：

• DSPy：Prompt调优时间-96%，准确率+15%
• Agentic RAG：检索相关性+18%，复杂查询准确率+25%
• 记忆系统：多轮对话准确率+19%，重复推荐率-100%

但也要注意：

• 不是所有场景都能达到这个效果
• 数据质量很重要（垃圾进，垃圾出）
• 要根据实际情况调整参数

最重要的是：系统现在能自我优化了！

---