GRASP：基于强化学习的粒度感知搜索策略，让RAG系统学会“思考”

1. 项目概述当RAG遇上“思考”GRASP如何重塑检索策略如果你最近在折腾大模型应用尤其是RAG检索增强生成系统那你肯定对“幻觉”和“召回不准”这两个老问题深恶痛绝。传统的RAG流程很直接用户提问 - 检索相关文档片段 - 扔给大模型生成答案。但这里有个核心矛盾检索的“粒度”和“意图”往往是不匹配的。用户问一个复杂、需要多步推理的问题比如“对比A公司和B公司在过去五年云计算战略上的异同并分析其背后的技术驱动力”你用一个固定的、通常是基于语义相似度的检索策略去捞一堆可能相关但信息冗余或碎片化的文档块结果就是大模型要么“吃撑了”信息过载要么“没吃饱”关键信息缺失生成的内容自然不尽人意。这就是“Agentic RAG”智能体驱动的RAG要解决的问题让检索过程本身也“智能”起来具备规划、决策和迭代的能力。而GRASPGRanularity-Aware Search Policy粒度感知搜索策略正是这个前沿方向上一个极具代表性的技术思路。它不像传统RAG那样“一锤子买卖”而是引入了一个搜索智能体Search Agent这个智能体能够根据当前对问题的理解状态动态地决定下一步该怎么搜是应该用更粗的粒度如章节、摘要来快速定位大致方向还是该用更细的粒度如具体段落、句子来精确获取细节亦或是需要转换查询词、甚至结合多个来源进行交叉验证简单说GRASP试图让RAG系统学会“像人一样思考和研究”先看目录粗粒度找到相关章节再精读段落细粒度必要时还去翻参考文献或附录。整个过程是一个由策略Policy驱动的、多步的、动态调整的序列决策过程。这背后的核心技术正是强化学习Reinforcement Learning, RL。智能体在“问题求解”这个环境中不断尝试不同的搜索动作调整粒度、改写查询等根据最终生成的答案质量获得奖励或惩罚从而学习出一套最优的搜索策略。这不仅仅是工程上的优化更是对RAG核心工作流的一次范式升级。2. GRASP核心架构与工作原理拆解要理解GRASP我们不能把它看成一个黑盒而是需要拆解其内部的关键组件和它们之间的交互逻辑。一个典型的GRASP框架通常包含以下几个核心部分它们共同构成了一个可学习的、闭环的搜索决策系统。2.1 智能体Agent、环境Environment与状态State在强化学习的框架下GRASP将整个搜索-问答过程建模为一个马尔可夫决策过程MDP。智能体Agent 这就是GRASP的核心决策大脑。它的观察对象是当前的环境状态输出是一个具体的“搜索动作”。这个智能体通常是一个神经网络如基于Transformer的小型策略网络其参数需要通过训练来优化。环境Environment 环境包括了整个外部世界用户的问题Query、庞大的文档知识库Knowledge Base、底层的检索器Retriever如基于向量数据库的稠密检索器或关键词检索器以及最终评估答案质量的大模型LLM或评判器Judge。智能体的动作会作用于环境引起环境状态的变化。状态State 这是智能体做决策的依据。在GRASP中状态的设计非常关键它需要编码当前搜索的“进度”和“上下文”。一个典型的状态表示可能包括原始用户问题。截至目前为止智能体已经检索到的所有文档片段或它们的摘要表示。上一步搜索动作及其结果例如上次用“细粒度”搜索了“技术驱动力”这个词返回了5个片段。当前对问题的部分理解或中间答案可以由一个小型LLM或编码器生成。 状态通常被编码成一个固定长度的向量供策略网络处理。2.2 动作空间Action Space搜索策略的操控杆智能体可以执行的动作定义了其能力范围。GRASP的核心创新在于其动作空间是粒度感知的。常见的动作包括粒度选择Granularity SelectionACTION_COARSE 以粗粒度进行检索。例如针对文档的章节标题、摘要、或通过聚类得到的大主题块进行搜索。目的是快速缩小范围把握全局结构。代价是信息不够精确。ACTION_FINE 以细粒度进行检索。例如针对原始的段落或句子块进行搜索。目的是获取精确的细节和证据。代价是可能陷入局部错过全局关联。ACTION_MIXED 混合粒度检索。例如同时检索粗粒度摘要和细粒度段落并进行融合。查询重构Query ReformulationACTION_EXPAND 基于当前已收集的信息扩展或泛化查询词。例如从“云计算战略”扩展到“IaaS, PaaS, SaaS 市场布局”。ACTION_SPECIFY 具体化或细化查询词。例如从“技术驱动力”具体到“芯片自研能力”或“分布式数据库技术”。ACTION_REPHRASE 用不同的表述方式重写查询以应对语义相似度检索的局限性。终止TerminationACTION_STOP 智能体认为当前收集的信息已足够回答问题停止搜索将收集到的所有上下文传递给答案生成器LLM。注意动作空间的设计需要平衡灵活性与复杂性。动作太多会导致训练困难探索空间巨大动作太少则无法体现智能体的优势。在实际实现中通常会从一个精简的动作集开始。2.3 奖励函数Reward Function好策略的指挥棒奖励函数是强化学习的“灵魂”它告诉智能体什么是好的行为。设计一个好的奖励函数是GRASP项目成败的关键。奖励通常是稀疏的在智能体执行ACTION_STOP并生成最终答案后才给出。常见的奖励信号来源包括答案质量奖励 这是最主要的奖励。可以通过以下方式计算基于参考答案的评分 使用ROUGE、BLEU或BERTScore等指标对比智能体生成的答案与人工标注的标准答案。基于LLM的评判 使用一个更强的LLM如GPT-4作为裁判根据问题、检索到的上下文和生成的答案从准确性、完整性、相关性等方面进行打分。这种方式更灵活但成本较高。基于事实一致性的奖励 检查生成答案中的关键事实是否都能从检索到的上下文中找到支持减少幻觉。效率惩罚 为了鼓励高效搜索通常会引入一个小的负奖励惩罚与智能体采取的搜索步数成正比。这防止智能体无休止地检索无关信息。信息增益奖励 在每一步可以评估新检索到的文档片段带来了多少“新信息”例如通过与已有上下文计算相似度低相似度意味着高信息增益。这可以作为一种中间奖励引导智能体探索新区域。奖励函数的设计需要仔细权衡。过分强调答案质量可能导致智能体过度搜索步数太多而过分强调效率则可能导致答案不完整。2.4 策略网络Policy Network与训练流程策略网络是一个参数化的函数通常为神经网络它接收当前状态s输出在动作空间上的概率分布π(a|s)。智能体根据这个分布采样动作来执行。训练GRASP是一个典型的强化学习过程通常采用策略梯度方法如REINFORCE或Actor-Critic架构初始化 随机初始化策略网络参数。采样轨迹 用当前策略在多个问题上运行即从初始状态开始根据策略选择动作与环境交互直到触发ACTION_STOP收集大量的轨迹数据(s1, a1, r1, s2, a2, r2, ..., sT)。这里的rT是最终的稀疏奖励中间步骤的奖励可能为0或包含信息增益等辅助奖励。计算回报 对于轨迹中的每个时间步t计算其累计回报Gt未来奖励的折扣和。更新策略 使用策略梯度定理计算梯度并更新策略网络参数目标是最大化期望回报。在Actor-Critic中还会同时训练一个价值网络Critic来估计状态的价值以降低梯度估计的方差。迭代 重复步骤2-4直到策略收敛。实操心得 训练一个有效的GRASP智能体挑战很大。稀疏奖励和长序列决策是两大难点。实践中我们常采用“课程学习”Curriculum Learning先让智能体在简单问题上学习例如只需1-2步检索就能回答的问题再逐步过渡到复杂问题。此外使用一个预训练的检索器或语言模型来初始化策略网络的某些部分即“热启动”也能显著加速训练。3. 从零到一构建一个简易GRASP系统的实操指南理论说了这么多我们来点实际的。下面我将勾勒一个简化版的GRASP系统实现方案使用Python和主流开源库帮助你理解每个模块如何拼接。这个示例旨在揭示核心流程而非生产级代码。3.1 环境准备与知识库构建假设我们有一个关于科技公司的文档集PDF、HTML等。第一步是构建可被检索的知识库。# 环境准备安装必要库 # pip install langchain chromadb openai tiktoken transformers sentence-transformers import os from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings # 或用开源的sentence-transformers # 1. 加载文档 loader DirectoryLoader(./tech_company_docs/, glob**/*.pdf, loader_clsPyPDFLoader) documents loader.load() # 2. 文档切片 - 这里体现“粒度”我们准备两种粒度。 text_splitter_coarse RecursiveCharacterTextSplitter( chunk_size1000, # 大块粗粒度 chunk_overlap200, separators[\n\n, \n, 。, , ] ) text_splitter_fine RecursiveCharacterTextSplitter( chunk_size250, # 小块细粒度 chunk_overlap50, separators[\n, 。, , , ] ) chunks_coarse text_splitter_coarse.split_documents(documents) chunks_fine text_splitter_fine.split_documents(documents) # 为不同粒度添加元数据标签 for chunk in chunks_coarse: chunk.metadata[granularity] coarse for chunk in chunks_fine: chunk.metadata[granularity] fine all_chunks chunks_coarse chunks_fine # 3. 向量化并存入向量数据库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 或使用 HuggingFaceEmbeddings vectorstore Chroma.from_documents( documentsall_chunks, embeddingembeddings, persist_directory./chroma_db_granular ) # 现在向量库中每个片段都有‘granularity’元数据3.2 定义GRASP环境与状态我们创建一个简单的环境类来封装与向量数据库和LLM的交互。import numpy as np from langchain_openai import ChatOpenAI from typing import List, Dict, Any, Tuple class GraspEnvironment: def __init__(self, vectorstore, llm, max_steps5): self.vectorstore vectorstore self.llm llm # 用于生成最终答案和评估的LLM self.max_steps max_steps self.reset() def reset(self, query: str): 重置环境开始一个新的问题 self.query query self.retrieved_chunks: List[str] [] # 历史检索到的所有片段内容 self.retrieved_metadata: List[Dict] [] # 对应的元数据如粒度 self.current_step 0 self.done False # 初始状态可以只是问题本身或者加上一个空上下文 self.state self._get_state_representation() return self.state def _get_state_representation(self) - np.ndarray: 将当前状态编码为向量。简化版使用查询已检索文本的嵌入均值 # 使用一个句子编码器例如与向量库相同的embedding模型 from sentence_transformers import SentenceTransformer encoder SentenceTransformer(all-MiniLM-L6-v2) context_text .join(self.retrieved_chunks[-3:]) if self.retrieved_chunks else # 只看最近3个片段 full_text fQuery: {self.query}. Context: {context_text} state_vec encoder.encode(full_text) return state_vec def step(self, action: Dict) - Tuple[np.ndarray, float, bool, Dict]: 执行动作。 action: 字典例如 {type: retrieve, granularity: fine, query: 具体化的查询词} if self.done or self.current_step self.max_steps: return self.state, 0.0, True, {} self.current_step 1 reward 0.0 info {} if action[type] retrieve: # 执行检索 search_query action.get(query, self.query) # 允许智能体重写查询 target_granularity action.get(granularity, None) # 构建过滤器 filter None if target_granularity: filter {granularity: target_granularity} # 检索 docs self.vectorstore.similarity_search( search_query, k3, # 每次检索3个片段 filterfilter ) new_chunks [doc.page_content for doc in docs] new_meta [doc.metadata for doc in docs] # 去重简单版根据内容哈希 for content, meta in zip(new_chunks, new_meta): if content not in self.retrieved_chunks: self.retrieved_chunks.append(content) self.retrieved_metadata.append(meta) info[retrieved] new_chunks info[granularity] target_granularity # 中间奖励基于信息增益简化版用新片段与旧上下文的相似度 if len(self.retrieved_chunks) len(new_chunks): # 如果有新增 # 计算新增内容的平均嵌入与现有上下文的平均嵌入的余弦相似度 # 相似度越低信息增益越大给予小的正奖励 # 这里省略具体计算仅为示意 reward 0.01 # 微小鼓励 elif action[type] stop: self.done True # 生成最终答案并计算最终奖励 final_answer, final_reward self._generate_and_evaluate() reward final_reward info[final_answer] final_answer info[final_reward] final_reward else: # 无效动作 reward - 0.1 # 更新状态 self.state self._get_state_representation() # 效率惩罚每一步都扣一点分鼓励尽快停止 step_penalty -0.02 reward step_penalty return self.state, reward, self.done, info def _generate_and_evaluate(self) - Tuple[str, float]: 生成最终答案并评估简化评估 context \n\n.join(self.retrieved_chunks) prompt f基于以下信息回答问题。如果信息不足请说明。 信息 {context} 问题{self.query} 答案 response self.llm.invoke(prompt) final_answer response.content # 简化评估使用一个评估LLM或规则 # 这里用一个非常简单的规则检查答案长度和是否包含关键词仅作演示 reward 0.0 if len(final_answer) 20: # 答案不能太短 reward 0.5 # 可以加入更复杂的评估如调用GPT-4作为裁判 # 这里我们假设有一个理想答案计算相似度模拟 ideal_answer 这是一个模拟的理想答案。 # 计算简单的词重叠作为奖励实际应用应使用更健壮的指标 overlap len(set(final_answer.split()) set(ideal_answer.split())) / len(set(ideal_answer.split())) reward overlap * 2.0 # 重叠度奖励 return final_answer, reward3.3 实现策略网络与训练循环我们将使用一个简单的神经网络作为策略网络并采用REINFORCE算法进行训练。import torch import torch.nn as nn import torch.optim as optim from torch.distributions import Categorical class PolicyNetwork(nn.Module): def __init__(self, state_dim, action_dims): super(PolicyNetwork, self).__init__() # action_dims: 一个字典定义每个动作子空间的维度 # 例如{type: 2, granularity: 3, query_shift: 5} self.fc nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 64), nn.ReLU(), ) # 为每个动作头创建输出层 self.action_heads nn.ModuleDict() for name, dim in action_dims.items(): self.action_heads[name] nn.Linear(64, dim) def forward(self, state): x self.fc(state) logits {} for name, head in self.action_heads.items(): logits[name] head(x) return logits # 返回每个动作头的logits # 定义动作空间 ACTION_SPACE { type: [retrieve, stop], # 动作类型检索或停止 granularity: [coarse, fine, None], # 检索粒度None表示不指定 # 查询重构可以简化为一个分类动作例如[keep, expand, specify] query_shift: [keep, expand, specify] } def action_to_dict(action_indices): 将网络输出的动作索引转换为环境可执行的字典 # 这里需要根据采样到的索引映射回具体的动作值 # 简化处理假设已经映射好 return action_indices def train_grasp(env, policy_net, optimizer, num_episodes1000): policy_net.train() for episode in range(num_episodes): # 采样一个问题在实际中应从数据集中采样 query 请分析A公司云计算战略的核心技术优势 state env.reset(query) episode_log_probs [] episode_rewards [] done False while not done: state_tensor torch.FloatTensor(state).unsqueeze(0) action_logits_dict policy_net(state_tensor) # 采样动作 actions {} log_probs [] for name, logits in action_logits_dict.items(): dist Categorical(logitslogits) action_idx dist.sample() log_prob dist.log_prob(action_idx) actions[name] action_idx.item() log_probs.append(log_prob) # 执行动作 action_dict action_to_dict(actions) # 转换为环境需要的格式 next_state, reward, done, info env.step(action_dict) # 存储数据 episode_log_probs.append(sum(log_probs)) # 简化求和所有动作的log prob episode_rewards.append(reward) state next_state # REINFORCE更新 returns [] R 0 for r in episode_rewards[::-1]: # 从后向前计算回报 R r 0.99 * R # 折扣因子0.99 returns.insert(0, R) returns torch.FloatTensor(returns) # 归一化回报减少方差 returns (returns - returns.mean()) / (returns.std() 1e-8) policy_loss [] for log_prob, R in zip(episode_log_probs, returns): policy_loss.append(-log_prob * R) # 负号因为我们要最大化回报 policy_loss torch.stack(policy_loss).sum() optimizer.zero_grad() policy_loss.backward() optimizer.step() if episode % 100 0: print(fEpisode {episode}, Total Reward: {sum(episode_rewards):.2f}, Loss: {policy_loss.item():.4f})注意事项 以上代码是高度简化的教学示例。真实训练中你需要处理更复杂的状态表示如使用RNN或Transformer编码历史、更合理的动作空间查询重构可能需要一个独立的语言模型、更稳定的奖励函数如使用基于LLM的评判器以及更先进的强化学习算法如PPO、A2C。此外收集高质量的对话或问答轨迹数据用于训练也是一个重大挑战。4. GRASP实战中的挑战、技巧与未来方向即便理解了原理和框架真正让一个GRASP系统跑起来并达到实用水平依然充满挑战。下面分享一些从实验和论文中总结出的关键点。4.1 核心挑战与应对策略训练数据与模拟环境挑战GRASP需要大量的问题多步搜索轨迹最终答案三元组数据进行训练。人工标注成本极高。应对使用现有QA数据集合成轨迹 对于已有标准答案的问题可以设计启发式规则如先检索粗粒度定位再检索细粒度找证据来生成“伪轨迹”作为初始训练数据。构建模拟器Simulator 创建一个简化但可控的环境模拟器。例如假设一个文档库预先定义好不同粒度片段之间的关系如段落属于哪个章节智能体在模拟器中互动奖励根据预设规则计算。这可以快速生成大量训练数据。离线强化学习 利用已有的、非策略的搜索日志数据如搜索引擎日志进行训练无需在线交互。奖励函数的稀疏性与设计难题挑战 最终答案质量的奖励非常稀疏且延迟。智能体很难将最终的成功/失败归因到早期具体的搜索动作上。应对分层奖励Hierarchical Reward 除了最终奖励设计合理的中间奖励。例如当智能体检索到的片段被最终答案生成器“引用”时给予正向奖励。或者使用一个“相关性预测模型”对每一步检索到的片段进行即时评分。好奇心驱动探索 引入内在好奇心模块奖励智能体探索到新的、信息丰富的文档区域即使这些区域暂时看来与最终答案无关。逆强化学习 不直接设计奖励函数而是从专家演示可能是人工或启发式规则生成的优质轨迹中反推出奖励函数。策略的泛化能力挑战 在一个特定知识库上训练的策略能否迁移到结构、主题不同的新知识库应对元学习Meta-Learning 在多个不同领域或结构的知识库上进行训练让策略学会快速适应新环境。状态表示的泛化性 设计不依赖于特定知识库元数据的、更通用的状态表示。例如使用检索结果本身的语义特征而非其所在的章节ID。模块化设计 将策略网络分解为通用模块如查询分析模块和领域特定模块。在新领域只需微调少量参数。4.2 性能优化与工程实践技巧动作空间的剪枝与分层 不要一开始就设计过于复杂的动作空间。可以从最简单的“检索固定粒度”和“停止”开始。随着策略稳定再逐步引入“粒度切换”和“查询重构”。考虑使用分层强化学习高层策略决定宏观阶段如“探索阶段”、“精炼阶段”底层策略执行具体动作。利用预训练模型进行热启动策略网络初始化 可以使用在大量文本上预训练过的语言模型如BERT、T5的小型版本作为策略网络的基础编码器。它已经具备了强大的语言理解能力。价值函数初始化 在Actor-Critic框架中Critic网络可以初始化为一个能预测答案质量的回归模型在监督数据上预训练。推理时的加速与稳定性束搜索Beam Search 在智能体推理时不单单采样一个动作而是保留概率最高的Top-K个动作序列进行探索最后选择整体期望回报最高的序列。这能提高决策质量但会增加计算量。提前停止Early Stopping 设定一个置信度阈值当策略网络对ACTION_STOP的输出概率超过阈值时强制终止避免不必要的搜索。缓存机制 对相同的中间状态和动作其检索结果是确定的。实现一个缓存可以极大加快训练和推理速度。4.3 未来演进方向GRASP代表了RAG系统向更自主、更智能方向迈出的重要一步。它的未来演进可能会与以下几个方向深度融合与工具调用Tool Use结合 智能体的动作空间不局限于文本检索可以扩展为调用各种API工具例如计算器、代码解释器、专业数据库查询等实现真正意义上的“智能体即操作系统”。多模态检索 当前的GRASP主要针对文本。未来的版本需要处理图像、表格、图表等多模态信息动作可能包括“聚焦图像区域”、“提取表格某列”等。终身学习与在线适应 系统在部署后能持续从用户反馈如点赞、纠错中学习动态调整其搜索策略适应知识库的更新和用户偏好的变化。可解释性与可控性 让智能体能够解释它为什么选择某个粒度或改写查询并提供给用户一些干预的接口例如用户可以提示“请从财务角度再深入搜索一下”。构建一个强大的GRASP系统绝非一日之功它需要机器学习、信息检索、系统工程等多方面的知识。但它的潜力是巨大的——它有望将RAG从一种被动的、静态的检索工具转变为一个主动的、动态的研究伙伴。对于任何希望构建下一代知识密集型AI应用的人来说深入理解并实践Agentic RAG与GRASP这类技术将是保持竞争力的关键。从今天开始尝试在你的下一个RAG项目中加入哪怕是最简单的一步决策循环你或许就能立刻感受到其带来的不同。