在构建基于大语言模型的智能应用时RAG检索增强生成技术已成为连接私有知识与模型泛化能力的核心桥梁。然而随着应用场景从简单的单轮问答向复杂的多步推理演进传统RAG的静态、粗粒度检索模式逐渐暴露出其局限性。你是否遇到过这样的困境面对一个需要拆解、分析、综合多个知识片段才能回答的复杂问题RAG系统要么一次性召回大量无关信息淹没核心线索要么因检索粒度不当而遗漏关键细节导致最终答案的准确性和逻辑性大打折扣这正是当前Agentic RAG智能体驱动的RAG面临的核心“粒度痛点”。本文将从工程实战角度深入剖析这一痛点并系统介绍一个创新的解决方案——GRASPGRAined SPecific动态检索框架。GRASP的核心思想是让检索过程“活”起来使其能够智能地适配LLM的多步推理逻辑实现检索粒度与推理需求的动态对齐。我们将不仅探讨其原理更会提供一套从环境搭建、核心代码实现到生产级最佳实践的完整指南帮助开发者将这一前沿框架落地于自己的项目中。1. RAG的粒度困境与GRASP的破局思路在深入GRASP之前我们必须先理解传统RAG在复杂问答场景下为何会“失灵”。1.1 传统RAG的静态检索之殇传统的RAG流程通常是线性的将用户查询向量化在向量数据库中执行相似性搜索召回Top-K个相关的文档片段chunks然后将这些片段与问题一起拼接成提示词Prompt交给LLM生成答案。这里的“文档片段”是通过预先设定的固定规则如按字符数、句子或段落对原始文档进行分割得到的。这种模式在事实型、单点知识问答上表现良好。例如“Python中如何读取文件”这样的问题召回几个关于open()函数和文件模式的chunk就足够了。然而当问题变得复杂时例如“请对比Spring Boot和Django在微服务架构下的优缺点并给出一个从单体应用迁移到微服务的风险评估清单。” 这个问题隐含了多个子任务理解Spring Boot的微服务特性。理解Django的微服务特性可能需要第三方库支持。进行多维度对比。理解单体到微服务的迁移模式。识别并列举风险。此时固定粒度的检索会面临两难粗粒度如按整个章节检索可能一次性召回包含Spring Boot配置、Django ORM、迁移工具等混杂信息的超大文本块。大量无关信息会稀释关键内容增加LLM的处理负担和幻觉风险即“信息过载”。细粒度如按单个段落或句子检索可能无法召回那些分散在不同chunk中、但逻辑上紧密关联以完成对比或综合推理的信息。例如Spring Boot的一个优点和Django的一个对应缺点可能位于不同的chunk中细粒度检索可能只召回其中之一导致答案片面即“信息割裂”。1.2 Agentic RAG的挑战动态推理需要动态检索Agentic RAG引入了智能体Agent的概念通过规划Planning、工具调用Tool Use等能力将复杂问题分解为一系列子步骤Step-by-Step Reasoning。每个步骤可能关注知识库中不同的方面需要不同粒度或不同侧重点的检索支持。例如上述复杂问题可能被智能体分解为步骤1“检索Spring Boot的微服务支持特性”。需要中等粒度聚焦特性描述步骤2“检索Django的微服务支持特性及常用库”。需要中等粒度聚焦特性与生态步骤3“检索微服务架构的通用优缺点维度”。需要较粗粒度聚焦框架对比方法论步骤4“检索单体应用迁移到微服务的常见风险”。需要细粒度聚焦具体风险点显然为所有步骤使用同一种固定的检索粒度是低效甚至错误的。Agentic RAG的粒度痛点本质上是静态检索能力与动态推理需求之间的不匹配。1.3 GRASP框架的核心思想GRASP框架正是为解决此问题而生。其核心创新在于将检索过程本身也纳入到智能体的推理循环中实现检索策略的动态化、精细化。GRASP不是一个全新的RAG系统而是一个可插拔的“动态检索层”它可以集成到现有的Agentic RAG架构中。GRASP的工作流程可以概括为意图解析与步骤规划智能体或规划器首先解析复杂问题生成一个初步的多步推理计划。粒度感知与检索策略生成对于计划中的每一个推理步骤GRASP模块会分析该步骤的查询意图、所需信息的类型和范围动态决定最合适的检索粒度如文档级、章节级、段落级、句子级和检索策略如关键词加强、元数据过滤。动态执行与知识获取根据生成的策略从知识库中执行检索获取最适配当前步骤的知识片段。反馈与迭代将检索结果提供给LLM执行该步骤推理并根据推理结果或LLM的置信度决定是否需要调整后续步骤的检索策略甚至回溯重试。简而言之GRASP让RAG系统在回答问题时能够像人类一样“先瞄一眼大纲再细读某一章最后精研某一段”实现检索资源的最优分配。2. 环境准备与项目搭建我们将使用Python语言基于LangChain和LlamaIndex这两个流行的LLM应用框架来构建一个集成了GRASP思想的演示项目。选择它们是因为其生态丰富易于集成和扩展。2.1 基础环境与依赖确保你的Python版本在3.8以上。我们使用conda或venv创建独立环境。# 创建并激活虚拟环境 conda create -n grasp_rag python3.10 conda activate grasp_rag # 安装核心依赖 pip install langchain langchain-community langchain-openai pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai pip install openai tiktoken pip install pypdf # 用于处理PDF文档 pip install chromadb # 使用ChromaDB作为轻量级向量数据库版本说明本文示例基于langchain0.1.0以上版本及llama-index-core0.10.0以上版本其API与旧版如0.0.x有较大差异。请务必注意版本兼容性建议使用最新稳定版。2.2 项目结构创建一个清晰的项目目录便于管理代码和资源。grasp_rag_demo/ ├── data/ # 存放原始知识文档 │ └── spring_django_microservices.pdf ├── knowledge_base/ # 向量数据库存储目录 ├── src/ │ ├── __init__.py │ ├── grasp_retriever.py # GRASP动态检索器核心实现 │ ├── agent_planner.py # 智能体与规划器 │ └── main.py # 主程序入口 ├── config.py # 配置文件API密钥等 └── requirements.txt2.3 配置API密钥在config.py中配置你的大语言模型和嵌入模型访问密钥。本例使用OpenAI API你也可以替换为其他兼容接口的模型如Azure OpenAI, Groq, 本地部署的Ollama等。# config.py import os from dotenv import load_dotenv load_dotenv() # 从.env文件加载环境变量 # OpenAI配置 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) # 可配置代理或自定义端点 EMBEDDING_MODEL text-embedding-3-small # 性价比高的嵌入模型 LLM_MODEL gpt-4o-mini # 或 gpt-4-turbo, gpt-3.5-turbo # 向量数据库配置 PERSIST_DIR ./knowledge_base # ChromaDB持久化目录在项目根目录创建.env文件存放密钥切勿提交至版本控制系统OPENAI_API_KEYsk-your-openai-api-key-here3. GRASP动态检索器核心实现这是整个框架的灵魂。我们将实现一个GraspRetriever类它能够根据输入的“查询”和“当前推理步骤的上下文”来动态选择检索策略。3.1 定义检索粒度策略首先我们定义几种典型的检索粒度策略。# src/grasp_retriever.py from enum import Enum from typing import Dict, Any, List, Optional, Tuple from pydantic import BaseModel, Field from llama_index.core import VectorStoreIndex, Settings from llama_index.core.retrievers import BaseRetriever from llama_index.core.schema import NodeWithScore, QueryBundle from llama_index.core.postprocessor import SimilarityPostprocessor import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class RetrievalGranularity(str, Enum): 检索粒度枚举 DOCUMENT document # 文档级召回整个文档或大型片段 SECTION section # 章节/主题级 PARAGRAPH paragraph # 段落级 SENTENCE sentence # 句子级最高精度 HYBRID hybrid # 混合粒度用于需要概览和细节的场景 class RetrievalStrategy(BaseModel): 动态检索策略 granularity: RetrievalGranularity top_k: int Field(description召回数量根据粒度动态调整, ge1, le20) similarity_cutoff: Optional[float] Field(defaultNone, description相似度阈值) use_metadata_filter: bool Field(defaultFalse, description是否使用元数据过滤) metadata_filters: Optional[Dict[str, Any]] Field(defaultNone) emphasize_keywords: List[str] Field(default_factorylist, description需要强调的关键词列表) class Config: use_enum_values True3.2 实现策略决策器策略决策器是一个小型“控制器”它根据查询和上下文调用LLM来决策最佳的检索策略。这里我们使用LangChain的LCELLangChain Expression Language来构建一个可预测的决策链。# src/grasp_retriever.py (续) from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from langchain_core.exceptions import OutputParserException import json class StrategyDecider: 基于LLM的检索策略决策器 def __init__(self, llm_model: str gpt-4o-mini): self.llm ChatOpenAI(modelllm_model, temperature0.1) self.parser JsonOutputParser(pydantic_objectRetrievalStrategy) # 决策提示词模板 self.decision_prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息检索策略专家。你的任务是根据用户当前的问题和推理步骤决定从知识库中检索信息的最佳策略。 请从以下粒度中选择最合适的一种{granularity_options}。 考虑因素 1. **问题复杂性**简单事实查询用细粒度复杂分析/对比用粗粒度或混合粒度。 2. **信息范围**需要广泛背景如概述、定义用粗粒度需要具体细节、数据、步骤用细粒度。 3. **步骤类型**若为规划中的一步思考这一步具体需要什么信息。 请以JSON格式输出严格遵循以下schema {format_instructions} ), (human, 当前推理步骤或问题{query}\n\n当前步骤的上下文或目标{context}), ]) def decide_strategy(self, query: str, context: str ) - RetrievalStrategy: 决定检索策略 try: # 准备提示词 prompt_value self.decision_prompt.invoke({ granularity_options: , .join([g.value for g in RetrievalGranularity]), format_instructions: self.parser.get_format_instructions(), query: query, context: context }) # 调用LLM response self.llm.invoke(prompt_value.to_messages()) # 解析输出 strategy_dict self.parser.parse(response.content) return RetrievalStrategy(**strategy_dict) except (OutputParserException, json.JSONDecodeError) as e: logger.warning(fLLM策略解析失败使用默认策略。错误{e}) # 失败时返回一个稳健的默认策略 return RetrievalStrategy( granularityRetrievalGranularity.PARAGRAPH, top_k5, similarity_cutoff0.7 )3.3 实现多粒度检索器我们需要一个能够支持不同粒度检索的底层检索器。这里利用LlamaIndex的灵活性通过不同的节点解析和索引构建方式来实现。# src/grasp_retriever.py (续) from llama_index.core.node_parser import ( SentenceSplitter, SemanticSplitterNodeParser, HierarchicalNodeParser, SimpleNodeParser ) from llama_index.core import StorageContext, load_index_from_storage from llama_index.vector_stores.chroma import ChromaVectorStore import chromadb from chromadb.config import Settings as ChromaSettings class MultiGranularityRetriever: 支持多粒度检索的底层检索器 def __init__(self, persist_dir: str, embedding_model: str): self.persist_dir persist_dir self.embedding_model embedding_model # 初始化ChromaDB客户端和向量存储 chroma_client chromadb.PersistentClient( pathpersist_dir, settingsChromaSettings(anonymized_telemetryFalse) ) chroma_collection chroma_client.get_or_create_collection(grasp_rag) vector_store ChromaVectorStore(chroma_collectionchroma_collection) self.storage_context StorageContext.from_defaults(vector_storevector_store) # 加载或创建索引 try: self.index load_index_from_storage(storage_contextself.storage_context) logger.info(已加载现有向量索引。) except Exception as e: logger.info(f未找到现有索引将创建新索引。错误{e}) self.index None def build_index_from_documents(self, documents, chunk_size: int 512): 从文档构建索引初次使用或更新知识库时调用 from llama_index.core import Document from llama_index.embeddings.openai import OpenAIEmbedding if not isinstance(documents[0], Document): docs [Document(textdoc) for doc in documents] else: docs documents # 配置全局设置 Settings.embed_model OpenAIEmbedding(modelself.embedding_model) Settings.llm None # 索引构建不需要LLM # 使用分层节点解析器同时生成不同粒度的节点 node_parser HierarchicalNodeParser.from_defaults( chunk_sizes[chunk_size * 4, chunk_size * 2, chunk_size] # 大、中、小三种粒度 ) nodes node_parser.get_nodes_from_documents(docs) # 为节点添加元数据标识其粒度 for i, node in enumerate(nodes): # 根据节点长度等启发式方法简单判断粒度实际项目可用更复杂逻辑 if len(node.text) chunk_size * 3: node.metadata[granularity] RetrievalGranularity.DOCUMENT.value elif len(node.text) chunk_size * 1.5: node.metadata[granularity] RetrievalGranularity.SECTION.value else: node.metadata[granularity] RetrievalGranularity.PARAGRAPH.value node.metadata[node_id] i # 创建索引 self.index VectorStoreIndex( nodesnodes, storage_contextself.storage_context, show_progressTrue ) self.index.storage_context.persist(persist_dirself.persist_dir) logger.info(f索引构建完成共 {len(nodes)} 个节点。) return self.index def retrieve(self, query: str, strategy: RetrievalStrategy) - List[NodeWithScore]: 根据策略执行检索 if self.index is None: raise ValueError(索引未初始化请先构建或加载索引。) # 基础检索器 base_retriever self.index.as_retriever(similarity_top_kstrategy.top_k * 2) # 多召回一些用于后续过滤 # 执行检索 query_bundle QueryBundle(query_strquery) initial_nodes base_retriever.retrieve(query_bundle) # 应用粒度过滤 filtered_nodes [] for node in initial_nodes: node_granularity node.node.metadata.get(granularity, RetrievalGranularity.PARAGRAPH.value) # 根据策略的粒度要求进行匹配 if strategy.granularity RetrievalGranularity.HYBRID: # 混合模式接受所有粒度但可能根据其他策略调整排序 pass elif strategy.granularity ! node_granularity: # 粒度不匹配跳过这里简化处理实际可设置权重而非直接过滤 continue filtered_nodes.append(node) # 应用相似度阈值过滤 if strategy.similarity_cutoff is not None: filtered_nodes [n for n in filtered_nodes if n.score strategy.similarity_cutoff] # 应用元数据过滤 if strategy.use_metadata_filter and strategy.metadata_filters: filtered_nodes [ n for n in filtered_nodes if all(n.node.metadata.get(k) v for k, v in strategy.metadata_filters.items()) ] # 关键词强调简单实现将包含关键词的节点分数提高 if strategy.emphasize_keywords: for node in filtered_nodes: text_lower node.node.text.lower() for keyword in strategy.emphasize_keywords: if keyword.lower() in text_lower: node.score node.score * 1.2 # 权重提升20% break # 按分数排序并返回Top-K filtered_nodes.sort(keylambda x: x.score, reverseTrue) final_nodes filtered_nodes[:strategy.top_k] logger.info(f检索完成。策略{strategy.granularity}初始召回{len(initial_nodes)}条过滤后{len(final_nodes)}条。) return final_nodes3.4 整合成GRASP检索器最后我们将策略决策器和多粒度检索器整合在一起。# src/grasp_retriever.py (续) class GraspRetriever(BaseRetriever): GRASP动态检索器兼容LlamaIndex Retriever接口 def __init__(self, multi_retriever: MultiGranularityRetriever, strategy_decider: StrategyDecider): self.multi_retriever multi_retriever self.strategy_decider strategy_decider super().__init__() def _retrieve(self, query_bundle: QueryBundle) - List[NodeWithScore]: 核心检索方法 # 从query_bundle中提取查询和可能的上下文 query query_bundle.query_str # 这里可以扩展从query_bundle.metadata或其他地方获取推理步骤上下文 context query_bundle.custom_metadata.get(reasoning_step_context, ) if query_bundle.custom_metadata else # 1. 决策检索策略 strategy self.strategy_decider.decide_strategy(query, context) logger.info(f为查询『{query}』决策的策略{strategy.granularity}, top_k{strategy.top_k}) # 2. 执行检索 retrieved_nodes self.multi_retriever.retrieve(query, strategy) return retrieved_nodes # 为了兼容性也提供一个简单的retrieve方法 def retrieve_with_context(self, query: str, context: str ) - List[NodeWithScore]: 带上下文的检索供外部直接调用 from llama_index.core.schema import QueryBundle query_bundle QueryBundle( query_strquery, custom_metadata{reasoning_step_context: context} ) return self._retrieve(query_bundle)4. 构建集成GRASP的Agentic RAG智能体现在我们将GRASP检索器嵌入到一个具备多步推理能力的智能体工作流中。4.1 定义智能体规划与执行流程我们设计一个简单的两阶段智能体规划器Planner和执行器Executor。规划器将复杂问题分解执行器按步骤调用GRASP检索并推理。# src/agent_planner.py from typing import List, Dict, Any from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from src.grasp_retriever import GraspRetriever, NodeWithScore class ReasoningStep(BaseModel): 推理步骤定义 step_id: int Field(description步骤序号) description: str Field(description该步骤要完成的具体任务描述) query: str Field(description用于检索信息的查询语句) expected_output: str Field(description该步骤期望产出的中间结果) class Plan(BaseModel): 推理计划 original_question: str steps: List[ReasoningStep] final_synthesis_instruction: str Field(description如何综合各步骤结果形成最终答案的指令) class GraspEnhancedAgent: 集成GRASP检索的智能体 def __init__(self, grasp_retriever: GraspRetriever, llm_model: str gpt-4o-mini): self.retriever grasp_retriever self.llm ChatOpenAI(modelllm_model, temperature0.1) self.plan_parser JsonOutputParser(pydantic_objectPlan) # 规划器提示词 self.planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的复杂问题分解为一系列顺序执行的子步骤。 每个子步骤应该 1. 目标明确可独立检索信息并解答。 2. 其输出能为后续步骤提供输入或上下文。 3. 查询语句应具体便于从知识库中检索相关信息。 请以JSON格式输出计划。 {format_instructions} ), (human, 请为以下问题制定一个分步推理计划\n\n问题{question}), ]) # 步骤执行提示词集成检索结果 self.step_executor_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的分析助手。请基于提供的参考信息精确回答当前步骤的问题。 参考信息可能不完整或包含无关内容请仔细甄别只使用与问题直接相关的部分。 你的回答应聚焦、简洁为后续步骤提供清晰的基础。 如果参考信息不足以回答请明确指出缺少什么信息。 ), (human, 当前步骤任务{step_description} 步骤查询{step_query} 参考信息 {retrieved_context} 请基于以上信息完成本步骤的分析), ]) # 最终综合提示词 self.synthesizer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个总结与综合专家。请根据以下分步推理的结果整合成一份完整、连贯、专业的最终答案。 确保答案直接回应用户的原始问题逻辑清晰并涵盖所有关键子结论。 ), (human, 原始问题{original_question} 分步推理过程与结果 {step_results} 请给出最终答案), ]) def plan(self, question: str) - Plan: 生成推理计划 prompt_value self.planner_prompt.invoke({ format_instructions: self.plan_parser.get_format_instructions(), question: question }) response self.llm.invoke(prompt_value.to_messages()) plan_dict self.plan_parser.parse(response.content) return Plan(**plan_dict) def execute_step(self, step: ReasoningStep, previous_context: str ) - Dict[str, Any]: 执行单个推理步骤 # 1. 动态检索将上一步结果作为上下文传入影响本次检索策略 retrieved_nodes self.retriever.retrieve_with_context( querystep.query, contextf前序步骤上下文{previous_context}. 当前步骤目标{step.description} ) retrieved_text \n\n---\n\n.join([f[相关性{node.score:.3f}]\n{node.node.text} for node in retrieved_nodes]) # 2. 调用LLM基于检索结果进行步骤推理 executor_prompt self.step_executor_prompt.invoke({ step_description: step.description, step_query: step.query, retrieved_context: retrieved_text }) step_result self.llm.invoke(executor_prompt.to_messages()).content return { step: step, retrieved_nodes: retrieved_nodes, result: step_result } def run(self, question: str) - Dict[str, Any]: 运行完整Agent工作流 logger.info(f开始处理问题{question}) # 阶段1规划 logger.info(阶段1问题分解与规划...) plan self.plan(question) logger.info(f生成计划共{len(plan.steps)}个步骤。) # 阶段2顺序执行与推理 logger.info(阶段2顺序执行推理步骤...) step_results [] all_retrieved_info [] previous_step_result for step in plan.steps: logger.info(f 执行步骤 {step.step_id}: {step.description}) step_output self.execute_step(step, previous_step_result) step_results.append(step_output) all_retrieved_info.extend(step_output[retrieved_nodes]) previous_step_result step_output[result] # 将当前步骤结果作为下一步的部分上下文 logger.info(f 步骤{step.step_id}完成。结果摘要{step_output[result][:100]}...) # 阶段3综合 logger.info(阶段3综合各步骤结果生成最终答案...) step_results_formatted \n\n.join([ f步骤{res[step].step_id} ({res[step].description}):\n{res[result]} for res in step_results ]) synthesizer_prompt self.synthesizer_prompt.invoke({ original_question: question, step_results: step_results_formatted }) final_answer self.llm.invoke(synthesizer_prompt.to_messages()).content logger.info(处理完成。) return { original_question: question, plan: plan, step_executions: step_results, final_answer: final_answer, all_retrieved_nodes: all_retrieved_info }5. 完整实战从知识库构建到复杂问答让我们用一个完整的例子演示如何将上述所有模块串联起来解决一个复杂的对比类问题。5.1 准备知识库文档我们在data/目录下放置一个名为spring_django_microservices.pdf的文档内容需自行准备应包含Spring Boot和Django在微服务方面的介绍、特性、优缺点、迁移相关风险等内容。然后编写一个脚本加载文档并构建索引。# src/main.py (部分功能) import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.grasp_retriever import MultiGranularityRetriever, StrategyDecider, GraspRetriever from src.agent_planner import GraspEnhancedAgent from config import PERSIST_DIR, EMBEDDING_MODEL, OPENAI_API_KEY, LLM_MODEL import logging from llama_index.core import SimpleDirectoryReader logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def build_or_load_knowledge_base(): 构建或加载知识库索引 multi_retriever MultiGranularityRetriever( persist_dirPERSIST_DIR, embedding_modelEMBEDDING_MODEL ) # 检查索引是否存在不存在则构建 if multi_retriever.index is None: logger.info(未检测到现有索引开始从文档构建...) # 读取文档 documents SimpleDirectoryReader(./data).load_data() logger.info(f已加载 {len(documents)} 个文档。) # 构建索引 multi_retriever.build_index_from_documents(documents, chunk_size512) else: logger.info(已加载现有索引。) return multi_retriever def initialize_grasp_agent(): 初始化GRASP智能体 # 1. 准备知识库 multi_retriever build_or_load_knowledge_base() # 2. 初始化策略决策器 strategy_decider StrategyDecider(llm_modelLLM_MODEL) # 3. 初始化GRASP检索器 grasp_retriever GraspRetriever(multi_retriever, strategy_decider) # 4. 初始化智能体 agent GraspEnhancedAgent(grasp_retriever, llm_modelLLM_MODEL) return agent def ask_complex_question(question: str): 向智能体提问 agent initialize_grasp_agent() result agent.run(question) print(\n *60) print( 原始问题:) print(f {result[original_question]}) print(\n *60) print(️ 生成的推理计划:) for step in result[plan].steps: print(f 步骤{step.step_id}: {step.description}) print(f 查询: {step.query}) print(\n *60) print( 各步骤检索与推理摘要:) for exec in result[step_executions]: step exec[step] print(f\n 步骤{step.step_id} ({step.description}):) print(f 检索到 {len(exec[retrieved_nodes])} 个相关片段。) print(f 推理结果: {exec[result][:150]}...) print(\n *60) print( 最终答案:) print(result[final_answer]) print(*60) # 可选保存结果或进行进一步分析 return result if __name__ __main__: # 示例复杂问题 complex_question 请对比Spring Boot和Django在微服务架构下的优缺点并给出一个从单体应用迁移到微服务的风险评估清单。 # 运行 ask_complex_question(complex_question)5.2 运行与结果分析运行python src/main.py。控制台将输出详细的处理日志。通过日志你可以观察到规划阶段LLM将问题分解为多个步骤例如步骤1检索Spring Boot的微服务特性与优势。步骤2检索Django的微服务特性与常用方案。步骤3检索微服务架构的通用优缺点分析框架。步骤4检索单体迁移到微服务的常见风险与挑战。执行阶段对于每个步骤GRASP决策器会输出类似“为查询『Spring Boot微服务特性』决策的策略SECTION, top_k4”的信息表明它认为这一步需要章节级的中等粒度信息。检索器随后会召回相应粒度的文本片段。综合阶段LLM汇总各步骤的中间结论形成最终答案。与使用固定粒度检索的RAG相比GRASP引导下的回答通常具有更好的结构性和逻辑连贯性因为每个推理步骤都获得了“恰到好处”的信息支持避免了信息过载或割裂。6. 常见问题与排查思路在实际部署和运行GRASP框架时你可能会遇到以下问题问题现象可能原因排查思路与解决方案索引构建失败或速度慢1. 文档解析出错如PDF格式复杂。2. 嵌入模型API调用失败或超时。3. 文档过大节点过多。1. 检查文档格式尝试使用pypdf或pdfplumber等库确保文本提取正确。2. 检查网络和API密钥设置合理的超时和重试机制。3. 调整chunk_size或先对文档进行预处理如提取核心章节。策略决策器总是返回默认策略1. LLM调用失败或返回格式错误。2. 提示词Prompt设计不佳LLM不理解任务。3. 使用的LLM能力不足如gpt-3.5-turbo可能不稳定。1. 查看LLM调用日志和返回的原始内容确认是否被正确解析。2. 优化决策提示词提供更清晰的示例Few-shot。3. 升级到更强大的模型如gpt-4系列或在决策前加入一个校验步骤。检索结果与步骤意图不匹配1. 步骤分解的查询语句过于模糊或宽泛。2. 嵌入模型对领域术语表征能力不足。3. 粒度策略匹配逻辑过于简单。1. 在规划器提示词中要求生成更具体、包含关键实体的查询。2. 尝试使用领域微调的嵌入模型或在查询时加入同义词扩展。3. 改进MultiGranularityRetriever.retrieve中的粒度匹配逻辑例如使用软匹配或加权分数。智能体陷入循环或步骤无意义1. 规划器提示词导致步骤分解不合理。2. 某一步骤检索结果太差导致后续步骤失去方向。1. 在规划器提示词中增加约束如“步骤数不超过5个”、“每个步骤应有明确、可验证的输出”。2. 引入步骤执行质量评估如果某步骤结果置信度低可以触发重试或调整检索策略。整体响应时间过长1. 步骤过多串行执行。2. 每次检索都重新计算策略LLM调用开销大。3. 向量检索本身较慢。1. 评估步骤间的依赖性对无依赖的步骤尝试并行执行。2. 为策略决策添加缓存对相似的查询-上下文对复用策略。3. 考虑使用更高效的向量数据库如FAISS的IVF索引或对索引进行量化。7. 生产环境最佳实践与进阶优化将GRASP框架用于实际项目时需要考虑以下工程化问题7.1 性能优化策略缓存为StrategyDecider增加缓存层如使用functools.lru_cache或Redis缓存(query, context)到RetrievalStrategy的映射避免重复调用LLM。异步执行使用asyncio和异步客户端如openai.AsyncOpenAI并行执行多个步骤的检索和LLM调用大幅减少端到端延迟。索引优化对于海量知识库使用更高效的索引算法如HNSW并定期进行索引压缩和清理。7.2 检索质量提升混合检索Hybrid Search在MultiGranularityRetriever.retrieve方法中不仅使用向量相似度检索还可以融合关键词检索如BM25的结果提高召回率。重排序Re-ranking在初步召回后使用一个更精细的交叉编码器模型如bge-reranker对结果进行重排序将最相关的结果排到前面进一步提升精度。查询扩展Query Expansion在将步骤查询送入检索器前先用LLM生成几个相关的问法或关键词合并进行检索以覆盖更广的信息面。7.3 智能体逻辑增强反思与迭代Reflection在每一步执行后让LLM对当前结果进行自我评估和反思。如果置信度低或发现矛盾可以触发回溯重新规划或调整检索策略。工具扩展将GRASP检索器作为智能体的一个核心工具并与其他工具如计算器、代码执行器、网络搜索API结合处理更复杂的任务。记忆机制为智能体添加短期或长期记忆记住历史对话和推理过程避免在长对话中重复检索相同信息。7.4 可观测性与评估详细日志记录记录每一步的策略决策、检索到的节点ID与分数、LLM的输入输出。这对于调试和效果分析至关重要。构建评估集针对你的业务领域构建一个包含复杂问题的测试集并标注期望答案或关键点。定期运行评估监控GRASP框架的效果。A/B测试在生产环境中可以对比使用GRASP和固定粒度检索的答案质量用数据证明其价值。GRASP框架代表了一种更智能、更自适应的RAG演进方向。它通过将检索决策与推理过程深度耦合有效解决了复杂问答中的信息粒度失配问题。实现这一框架需要仔细设计策略决策、多粒度索引和智能体工作流之间的交互。本文提供的代码是一个完整的起点你可以根据具体业务需求对其中的检索策略、规划逻辑和提示词进行深度定制和优化。 GRASP动态检索框架：解决Agentic RAG的粒度痛点，实现智能多步推理
在构建基于大语言模型的智能应用时RAG检索增强生成技术已成为连接私有知识与模型泛化能力的核心桥梁。然而随着应用场景从简单的单轮问答向复杂的多步推理演进传统RAG的静态、粗粒度检索模式逐渐暴露出其局限性。你是否遇到过这样的困境面对一个需要拆解、分析、综合多个知识片段才能回答的复杂问题RAG系统要么一次性召回大量无关信息淹没核心线索要么因检索粒度不当而遗漏关键细节导致最终答案的准确性和逻辑性大打折扣这正是当前Agentic RAG智能体驱动的RAG面临的核心“粒度痛点”。本文将从工程实战角度深入剖析这一痛点并系统介绍一个创新的解决方案——GRASPGRAined SPecific动态检索框架。GRASP的核心思想是让检索过程“活”起来使其能够智能地适配LLM的多步推理逻辑实现检索粒度与推理需求的动态对齐。我们将不仅探讨其原理更会提供一套从环境搭建、核心代码实现到生产级最佳实践的完整指南帮助开发者将这一前沿框架落地于自己的项目中。1. RAG的粒度困境与GRASP的破局思路在深入GRASP之前我们必须先理解传统RAG在复杂问答场景下为何会“失灵”。1.1 传统RAG的静态检索之殇传统的RAG流程通常是线性的将用户查询向量化在向量数据库中执行相似性搜索召回Top-K个相关的文档片段chunks然后将这些片段与问题一起拼接成提示词Prompt交给LLM生成答案。这里的“文档片段”是通过预先设定的固定规则如按字符数、句子或段落对原始文档进行分割得到的。这种模式在事实型、单点知识问答上表现良好。例如“Python中如何读取文件”这样的问题召回几个关于open()函数和文件模式的chunk就足够了。然而当问题变得复杂时例如“请对比Spring Boot和Django在微服务架构下的优缺点并给出一个从单体应用迁移到微服务的风险评估清单。” 这个问题隐含了多个子任务理解Spring Boot的微服务特性。理解Django的微服务特性可能需要第三方库支持。进行多维度对比。理解单体到微服务的迁移模式。识别并列举风险。此时固定粒度的检索会面临两难粗粒度如按整个章节检索可能一次性召回包含Spring Boot配置、Django ORM、迁移工具等混杂信息的超大文本块。大量无关信息会稀释关键内容增加LLM的处理负担和幻觉风险即“信息过载”。细粒度如按单个段落或句子检索可能无法召回那些分散在不同chunk中、但逻辑上紧密关联以完成对比或综合推理的信息。例如Spring Boot的一个优点和Django的一个对应缺点可能位于不同的chunk中细粒度检索可能只召回其中之一导致答案片面即“信息割裂”。1.2 Agentic RAG的挑战动态推理需要动态检索Agentic RAG引入了智能体Agent的概念通过规划Planning、工具调用Tool Use等能力将复杂问题分解为一系列子步骤Step-by-Step Reasoning。每个步骤可能关注知识库中不同的方面需要不同粒度或不同侧重点的检索支持。例如上述复杂问题可能被智能体分解为步骤1“检索Spring Boot的微服务支持特性”。需要中等粒度聚焦特性描述步骤2“检索Django的微服务支持特性及常用库”。需要中等粒度聚焦特性与生态步骤3“检索微服务架构的通用优缺点维度”。需要较粗粒度聚焦框架对比方法论步骤4“检索单体应用迁移到微服务的常见风险”。需要细粒度聚焦具体风险点显然为所有步骤使用同一种固定的检索粒度是低效甚至错误的。Agentic RAG的粒度痛点本质上是静态检索能力与动态推理需求之间的不匹配。1.3 GRASP框架的核心思想GRASP框架正是为解决此问题而生。其核心创新在于将检索过程本身也纳入到智能体的推理循环中实现检索策略的动态化、精细化。GRASP不是一个全新的RAG系统而是一个可插拔的“动态检索层”它可以集成到现有的Agentic RAG架构中。GRASP的工作流程可以概括为意图解析与步骤规划智能体或规划器首先解析复杂问题生成一个初步的多步推理计划。粒度感知与检索策略生成对于计划中的每一个推理步骤GRASP模块会分析该步骤的查询意图、所需信息的类型和范围动态决定最合适的检索粒度如文档级、章节级、段落级、句子级和检索策略如关键词加强、元数据过滤。动态执行与知识获取根据生成的策略从知识库中执行检索获取最适配当前步骤的知识片段。反馈与迭代将检索结果提供给LLM执行该步骤推理并根据推理结果或LLM的置信度决定是否需要调整后续步骤的检索策略甚至回溯重试。简而言之GRASP让RAG系统在回答问题时能够像人类一样“先瞄一眼大纲再细读某一章最后精研某一段”实现检索资源的最优分配。2. 环境准备与项目搭建我们将使用Python语言基于LangChain和LlamaIndex这两个流行的LLM应用框架来构建一个集成了GRASP思想的演示项目。选择它们是因为其生态丰富易于集成和扩展。2.1 基础环境与依赖确保你的Python版本在3.8以上。我们使用conda或venv创建独立环境。# 创建并激活虚拟环境 conda create -n grasp_rag python3.10 conda activate grasp_rag # 安装核心依赖 pip install langchain langchain-community langchain-openai pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai pip install openai tiktoken pip install pypdf # 用于处理PDF文档 pip install chromadb # 使用ChromaDB作为轻量级向量数据库版本说明本文示例基于langchain0.1.0以上版本及llama-index-core0.10.0以上版本其API与旧版如0.0.x有较大差异。请务必注意版本兼容性建议使用最新稳定版。2.2 项目结构创建一个清晰的项目目录便于管理代码和资源。grasp_rag_demo/ ├── data/ # 存放原始知识文档 │ └── spring_django_microservices.pdf ├── knowledge_base/ # 向量数据库存储目录 ├── src/ │ ├── __init__.py │ ├── grasp_retriever.py # GRASP动态检索器核心实现 │ ├── agent_planner.py # 智能体与规划器 │ └── main.py # 主程序入口 ├── config.py # 配置文件API密钥等 └── requirements.txt2.3 配置API密钥在config.py中配置你的大语言模型和嵌入模型访问密钥。本例使用OpenAI API你也可以替换为其他兼容接口的模型如Azure OpenAI, Groq, 本地部署的Ollama等。# config.py import os from dotenv import load_dotenv load_dotenv() # 从.env文件加载环境变量 # OpenAI配置 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) # 可配置代理或自定义端点 EMBEDDING_MODEL text-embedding-3-small # 性价比高的嵌入模型 LLM_MODEL gpt-4o-mini # 或 gpt-4-turbo, gpt-3.5-turbo # 向量数据库配置 PERSIST_DIR ./knowledge_base # ChromaDB持久化目录在项目根目录创建.env文件存放密钥切勿提交至版本控制系统OPENAI_API_KEYsk-your-openai-api-key-here3. GRASP动态检索器核心实现这是整个框架的灵魂。我们将实现一个GraspRetriever类它能够根据输入的“查询”和“当前推理步骤的上下文”来动态选择检索策略。3.1 定义检索粒度策略首先我们定义几种典型的检索粒度策略。# src/grasp_retriever.py from enum import Enum from typing import Dict, Any, List, Optional, Tuple from pydantic import BaseModel, Field from llama_index.core import VectorStoreIndex, Settings from llama_index.core.retrievers import BaseRetriever from llama_index.core.schema import NodeWithScore, QueryBundle from llama_index.core.postprocessor import SimilarityPostprocessor import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class RetrievalGranularity(str, Enum): 检索粒度枚举 DOCUMENT document # 文档级召回整个文档或大型片段 SECTION section # 章节/主题级 PARAGRAPH paragraph # 段落级 SENTENCE sentence # 句子级最高精度 HYBRID hybrid # 混合粒度用于需要概览和细节的场景 class RetrievalStrategy(BaseModel): 动态检索策略 granularity: RetrievalGranularity top_k: int Field(description召回数量根据粒度动态调整, ge1, le20) similarity_cutoff: Optional[float] Field(defaultNone, description相似度阈值) use_metadata_filter: bool Field(defaultFalse, description是否使用元数据过滤) metadata_filters: Optional[Dict[str, Any]] Field(defaultNone) emphasize_keywords: List[str] Field(default_factorylist, description需要强调的关键词列表) class Config: use_enum_values True3.2 实现策略决策器策略决策器是一个小型“控制器”它根据查询和上下文调用LLM来决策最佳的检索策略。这里我们使用LangChain的LCELLangChain Expression Language来构建一个可预测的决策链。# src/grasp_retriever.py (续) from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from langchain_core.exceptions import OutputParserException import json class StrategyDecider: 基于LLM的检索策略决策器 def __init__(self, llm_model: str gpt-4o-mini): self.llm ChatOpenAI(modelllm_model, temperature0.1) self.parser JsonOutputParser(pydantic_objectRetrievalStrategy) # 决策提示词模板 self.decision_prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息检索策略专家。你的任务是根据用户当前的问题和推理步骤决定从知识库中检索信息的最佳策略。 请从以下粒度中选择最合适的一种{granularity_options}。 考虑因素 1. **问题复杂性**简单事实查询用细粒度复杂分析/对比用粗粒度或混合粒度。 2. **信息范围**需要广泛背景如概述、定义用粗粒度需要具体细节、数据、步骤用细粒度。 3. **步骤类型**若为规划中的一步思考这一步具体需要什么信息。 请以JSON格式输出严格遵循以下schema {format_instructions} ), (human, 当前推理步骤或问题{query}\n\n当前步骤的上下文或目标{context}), ]) def decide_strategy(self, query: str, context: str ) - RetrievalStrategy: 决定检索策略 try: # 准备提示词 prompt_value self.decision_prompt.invoke({ granularity_options: , .join([g.value for g in RetrievalGranularity]), format_instructions: self.parser.get_format_instructions(), query: query, context: context }) # 调用LLM response self.llm.invoke(prompt_value.to_messages()) # 解析输出 strategy_dict self.parser.parse(response.content) return RetrievalStrategy(**strategy_dict) except (OutputParserException, json.JSONDecodeError) as e: logger.warning(fLLM策略解析失败使用默认策略。错误{e}) # 失败时返回一个稳健的默认策略 return RetrievalStrategy( granularityRetrievalGranularity.PARAGRAPH, top_k5, similarity_cutoff0.7 )3.3 实现多粒度检索器我们需要一个能够支持不同粒度检索的底层检索器。这里利用LlamaIndex的灵活性通过不同的节点解析和索引构建方式来实现。# src/grasp_retriever.py (续) from llama_index.core.node_parser import ( SentenceSplitter, SemanticSplitterNodeParser, HierarchicalNodeParser, SimpleNodeParser ) from llama_index.core import StorageContext, load_index_from_storage from llama_index.vector_stores.chroma import ChromaVectorStore import chromadb from chromadb.config import Settings as ChromaSettings class MultiGranularityRetriever: 支持多粒度检索的底层检索器 def __init__(self, persist_dir: str, embedding_model: str): self.persist_dir persist_dir self.embedding_model embedding_model # 初始化ChromaDB客户端和向量存储 chroma_client chromadb.PersistentClient( pathpersist_dir, settingsChromaSettings(anonymized_telemetryFalse) ) chroma_collection chroma_client.get_or_create_collection(grasp_rag) vector_store ChromaVectorStore(chroma_collectionchroma_collection) self.storage_context StorageContext.from_defaults(vector_storevector_store) # 加载或创建索引 try: self.index load_index_from_storage(storage_contextself.storage_context) logger.info(已加载现有向量索引。) except Exception as e: logger.info(f未找到现有索引将创建新索引。错误{e}) self.index None def build_index_from_documents(self, documents, chunk_size: int 512): 从文档构建索引初次使用或更新知识库时调用 from llama_index.core import Document from llama_index.embeddings.openai import OpenAIEmbedding if not isinstance(documents[0], Document): docs [Document(textdoc) for doc in documents] else: docs documents # 配置全局设置 Settings.embed_model OpenAIEmbedding(modelself.embedding_model) Settings.llm None # 索引构建不需要LLM # 使用分层节点解析器同时生成不同粒度的节点 node_parser HierarchicalNodeParser.from_defaults( chunk_sizes[chunk_size * 4, chunk_size * 2, chunk_size] # 大、中、小三种粒度 ) nodes node_parser.get_nodes_from_documents(docs) # 为节点添加元数据标识其粒度 for i, node in enumerate(nodes): # 根据节点长度等启发式方法简单判断粒度实际项目可用更复杂逻辑 if len(node.text) chunk_size * 3: node.metadata[granularity] RetrievalGranularity.DOCUMENT.value elif len(node.text) chunk_size * 1.5: node.metadata[granularity] RetrievalGranularity.SECTION.value else: node.metadata[granularity] RetrievalGranularity.PARAGRAPH.value node.metadata[node_id] i # 创建索引 self.index VectorStoreIndex( nodesnodes, storage_contextself.storage_context, show_progressTrue ) self.index.storage_context.persist(persist_dirself.persist_dir) logger.info(f索引构建完成共 {len(nodes)} 个节点。) return self.index def retrieve(self, query: str, strategy: RetrievalStrategy) - List[NodeWithScore]: 根据策略执行检索 if self.index is None: raise ValueError(索引未初始化请先构建或加载索引。) # 基础检索器 base_retriever self.index.as_retriever(similarity_top_kstrategy.top_k * 2) # 多召回一些用于后续过滤 # 执行检索 query_bundle QueryBundle(query_strquery) initial_nodes base_retriever.retrieve(query_bundle) # 应用粒度过滤 filtered_nodes [] for node in initial_nodes: node_granularity node.node.metadata.get(granularity, RetrievalGranularity.PARAGRAPH.value) # 根据策略的粒度要求进行匹配 if strategy.granularity RetrievalGranularity.HYBRID: # 混合模式接受所有粒度但可能根据其他策略调整排序 pass elif strategy.granularity ! node_granularity: # 粒度不匹配跳过这里简化处理实际可设置权重而非直接过滤 continue filtered_nodes.append(node) # 应用相似度阈值过滤 if strategy.similarity_cutoff is not None: filtered_nodes [n for n in filtered_nodes if n.score strategy.similarity_cutoff] # 应用元数据过滤 if strategy.use_metadata_filter and strategy.metadata_filters: filtered_nodes [ n for n in filtered_nodes if all(n.node.metadata.get(k) v for k, v in strategy.metadata_filters.items()) ] # 关键词强调简单实现将包含关键词的节点分数提高 if strategy.emphasize_keywords: for node in filtered_nodes: text_lower node.node.text.lower() for keyword in strategy.emphasize_keywords: if keyword.lower() in text_lower: node.score node.score * 1.2 # 权重提升20% break # 按分数排序并返回Top-K filtered_nodes.sort(keylambda x: x.score, reverseTrue) final_nodes filtered_nodes[:strategy.top_k] logger.info(f检索完成。策略{strategy.granularity}初始召回{len(initial_nodes)}条过滤后{len(final_nodes)}条。) return final_nodes3.4 整合成GRASP检索器最后我们将策略决策器和多粒度检索器整合在一起。# src/grasp_retriever.py (续) class GraspRetriever(BaseRetriever): GRASP动态检索器兼容LlamaIndex Retriever接口 def __init__(self, multi_retriever: MultiGranularityRetriever, strategy_decider: StrategyDecider): self.multi_retriever multi_retriever self.strategy_decider strategy_decider super().__init__() def _retrieve(self, query_bundle: QueryBundle) - List[NodeWithScore]: 核心检索方法 # 从query_bundle中提取查询和可能的上下文 query query_bundle.query_str # 这里可以扩展从query_bundle.metadata或其他地方获取推理步骤上下文 context query_bundle.custom_metadata.get(reasoning_step_context, ) if query_bundle.custom_metadata else # 1. 决策检索策略 strategy self.strategy_decider.decide_strategy(query, context) logger.info(f为查询『{query}』决策的策略{strategy.granularity}, top_k{strategy.top_k}) # 2. 执行检索 retrieved_nodes self.multi_retriever.retrieve(query, strategy) return retrieved_nodes # 为了兼容性也提供一个简单的retrieve方法 def retrieve_with_context(self, query: str, context: str ) - List[NodeWithScore]: 带上下文的检索供外部直接调用 from llama_index.core.schema import QueryBundle query_bundle QueryBundle( query_strquery, custom_metadata{reasoning_step_context: context} ) return self._retrieve(query_bundle)4. 构建集成GRASP的Agentic RAG智能体现在我们将GRASP检索器嵌入到一个具备多步推理能力的智能体工作流中。4.1 定义智能体规划与执行流程我们设计一个简单的两阶段智能体规划器Planner和执行器Executor。规划器将复杂问题分解执行器按步骤调用GRASP检索并推理。# src/agent_planner.py from typing import List, Dict, Any from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from src.grasp_retriever import GraspRetriever, NodeWithScore class ReasoningStep(BaseModel): 推理步骤定义 step_id: int Field(description步骤序号) description: str Field(description该步骤要完成的具体任务描述) query: str Field(description用于检索信息的查询语句) expected_output: str Field(description该步骤期望产出的中间结果) class Plan(BaseModel): 推理计划 original_question: str steps: List[ReasoningStep] final_synthesis_instruction: str Field(description如何综合各步骤结果形成最终答案的指令) class GraspEnhancedAgent: 集成GRASP检索的智能体 def __init__(self, grasp_retriever: GraspRetriever, llm_model: str gpt-4o-mini): self.retriever grasp_retriever self.llm ChatOpenAI(modelllm_model, temperature0.1) self.plan_parser JsonOutputParser(pydantic_objectPlan) # 规划器提示词 self.planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的复杂问题分解为一系列顺序执行的子步骤。 每个子步骤应该 1. 目标明确可独立检索信息并解答。 2. 其输出能为后续步骤提供输入或上下文。 3. 查询语句应具体便于从知识库中检索相关信息。 请以JSON格式输出计划。 {format_instructions} ), (human, 请为以下问题制定一个分步推理计划\n\n问题{question}), ]) # 步骤执行提示词集成检索结果 self.step_executor_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的分析助手。请基于提供的参考信息精确回答当前步骤的问题。 参考信息可能不完整或包含无关内容请仔细甄别只使用与问题直接相关的部分。 你的回答应聚焦、简洁为后续步骤提供清晰的基础。 如果参考信息不足以回答请明确指出缺少什么信息。 ), (human, 当前步骤任务{step_description} 步骤查询{step_query} 参考信息 {retrieved_context} 请基于以上信息完成本步骤的分析), ]) # 最终综合提示词 self.synthesizer_prompt ChatPromptTemplate.from_messages([ (system, 你是一个总结与综合专家。请根据以下分步推理的结果整合成一份完整、连贯、专业的最终答案。 确保答案直接回应用户的原始问题逻辑清晰并涵盖所有关键子结论。 ), (human, 原始问题{original_question} 分步推理过程与结果 {step_results} 请给出最终答案), ]) def plan(self, question: str) - Plan: 生成推理计划 prompt_value self.planner_prompt.invoke({ format_instructions: self.plan_parser.get_format_instructions(), question: question }) response self.llm.invoke(prompt_value.to_messages()) plan_dict self.plan_parser.parse(response.content) return Plan(**plan_dict) def execute_step(self, step: ReasoningStep, previous_context: str ) - Dict[str, Any]: 执行单个推理步骤 # 1. 动态检索将上一步结果作为上下文传入影响本次检索策略 retrieved_nodes self.retriever.retrieve_with_context( querystep.query, contextf前序步骤上下文{previous_context}. 当前步骤目标{step.description} ) retrieved_text \n\n---\n\n.join([f[相关性{node.score:.3f}]\n{node.node.text} for node in retrieved_nodes]) # 2. 调用LLM基于检索结果进行步骤推理 executor_prompt self.step_executor_prompt.invoke({ step_description: step.description, step_query: step.query, retrieved_context: retrieved_text }) step_result self.llm.invoke(executor_prompt.to_messages()).content return { step: step, retrieved_nodes: retrieved_nodes, result: step_result } def run(self, question: str) - Dict[str, Any]: 运行完整Agent工作流 logger.info(f开始处理问题{question}) # 阶段1规划 logger.info(阶段1问题分解与规划...) plan self.plan(question) logger.info(f生成计划共{len(plan.steps)}个步骤。) # 阶段2顺序执行与推理 logger.info(阶段2顺序执行推理步骤...) step_results [] all_retrieved_info [] previous_step_result for step in plan.steps: logger.info(f 执行步骤 {step.step_id}: {step.description}) step_output self.execute_step(step, previous_step_result) step_results.append(step_output) all_retrieved_info.extend(step_output[retrieved_nodes]) previous_step_result step_output[result] # 将当前步骤结果作为下一步的部分上下文 logger.info(f 步骤{step.step_id}完成。结果摘要{step_output[result][:100]}...) # 阶段3综合 logger.info(阶段3综合各步骤结果生成最终答案...) step_results_formatted \n\n.join([ f步骤{res[step].step_id} ({res[step].description}):\n{res[result]} for res in step_results ]) synthesizer_prompt self.synthesizer_prompt.invoke({ original_question: question, step_results: step_results_formatted }) final_answer self.llm.invoke(synthesizer_prompt.to_messages()).content logger.info(处理完成。) return { original_question: question, plan: plan, step_executions: step_results, final_answer: final_answer, all_retrieved_nodes: all_retrieved_info }5. 完整实战从知识库构建到复杂问答让我们用一个完整的例子演示如何将上述所有模块串联起来解决一个复杂的对比类问题。5.1 准备知识库文档我们在data/目录下放置一个名为spring_django_microservices.pdf的文档内容需自行准备应包含Spring Boot和Django在微服务方面的介绍、特性、优缺点、迁移相关风险等内容。然后编写一个脚本加载文档并构建索引。# src/main.py (部分功能) import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.grasp_retriever import MultiGranularityRetriever, StrategyDecider, GraspRetriever from src.agent_planner import GraspEnhancedAgent from config import PERSIST_DIR, EMBEDDING_MODEL, OPENAI_API_KEY, LLM_MODEL import logging from llama_index.core import SimpleDirectoryReader logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def build_or_load_knowledge_base(): 构建或加载知识库索引 multi_retriever MultiGranularityRetriever( persist_dirPERSIST_DIR, embedding_modelEMBEDDING_MODEL ) # 检查索引是否存在不存在则构建 if multi_retriever.index is None: logger.info(未检测到现有索引开始从文档构建...) # 读取文档 documents SimpleDirectoryReader(./data).load_data() logger.info(f已加载 {len(documents)} 个文档。) # 构建索引 multi_retriever.build_index_from_documents(documents, chunk_size512) else: logger.info(已加载现有索引。) return multi_retriever def initialize_grasp_agent(): 初始化GRASP智能体 # 1. 准备知识库 multi_retriever build_or_load_knowledge_base() # 2. 初始化策略决策器 strategy_decider StrategyDecider(llm_modelLLM_MODEL) # 3. 初始化GRASP检索器 grasp_retriever GraspRetriever(multi_retriever, strategy_decider) # 4. 初始化智能体 agent GraspEnhancedAgent(grasp_retriever, llm_modelLLM_MODEL) return agent def ask_complex_question(question: str): 向智能体提问 agent initialize_grasp_agent() result agent.run(question) print(\n *60) print( 原始问题:) print(f {result[original_question]}) print(\n *60) print(️ 生成的推理计划:) for step in result[plan].steps: print(f 步骤{step.step_id}: {step.description}) print(f 查询: {step.query}) print(\n *60) print( 各步骤检索与推理摘要:) for exec in result[step_executions]: step exec[step] print(f\n 步骤{step.step_id} ({step.description}):) print(f 检索到 {len(exec[retrieved_nodes])} 个相关片段。) print(f 推理结果: {exec[result][:150]}...) print(\n *60) print( 最终答案:) print(result[final_answer]) print(*60) # 可选保存结果或进行进一步分析 return result if __name__ __main__: # 示例复杂问题 complex_question 请对比Spring Boot和Django在微服务架构下的优缺点并给出一个从单体应用迁移到微服务的风险评估清单。 # 运行 ask_complex_question(complex_question)5.2 运行与结果分析运行python src/main.py。控制台将输出详细的处理日志。通过日志你可以观察到规划阶段LLM将问题分解为多个步骤例如步骤1检索Spring Boot的微服务特性与优势。步骤2检索Django的微服务特性与常用方案。步骤3检索微服务架构的通用优缺点分析框架。步骤4检索单体迁移到微服务的常见风险与挑战。执行阶段对于每个步骤GRASP决策器会输出类似“为查询『Spring Boot微服务特性』决策的策略SECTION, top_k4”的信息表明它认为这一步需要章节级的中等粒度信息。检索器随后会召回相应粒度的文本片段。综合阶段LLM汇总各步骤的中间结论形成最终答案。与使用固定粒度检索的RAG相比GRASP引导下的回答通常具有更好的结构性和逻辑连贯性因为每个推理步骤都获得了“恰到好处”的信息支持避免了信息过载或割裂。6. 常见问题与排查思路在实际部署和运行GRASP框架时你可能会遇到以下问题问题现象可能原因排查思路与解决方案索引构建失败或速度慢1. 文档解析出错如PDF格式复杂。2. 嵌入模型API调用失败或超时。3. 文档过大节点过多。1. 检查文档格式尝试使用pypdf或pdfplumber等库确保文本提取正确。2. 检查网络和API密钥设置合理的超时和重试机制。3. 调整chunk_size或先对文档进行预处理如提取核心章节。策略决策器总是返回默认策略1. LLM调用失败或返回格式错误。2. 提示词Prompt设计不佳LLM不理解任务。3. 使用的LLM能力不足如gpt-3.5-turbo可能不稳定。1. 查看LLM调用日志和返回的原始内容确认是否被正确解析。2. 优化决策提示词提供更清晰的示例Few-shot。3. 升级到更强大的模型如gpt-4系列或在决策前加入一个校验步骤。检索结果与步骤意图不匹配1. 步骤分解的查询语句过于模糊或宽泛。2. 嵌入模型对领域术语表征能力不足。3. 粒度策略匹配逻辑过于简单。1. 在规划器提示词中要求生成更具体、包含关键实体的查询。2. 尝试使用领域微调的嵌入模型或在查询时加入同义词扩展。3. 改进MultiGranularityRetriever.retrieve中的粒度匹配逻辑例如使用软匹配或加权分数。智能体陷入循环或步骤无意义1. 规划器提示词导致步骤分解不合理。2. 某一步骤检索结果太差导致后续步骤失去方向。1. 在规划器提示词中增加约束如“步骤数不超过5个”、“每个步骤应有明确、可验证的输出”。2. 引入步骤执行质量评估如果某步骤结果置信度低可以触发重试或调整检索策略。整体响应时间过长1. 步骤过多串行执行。2. 每次检索都重新计算策略LLM调用开销大。3. 向量检索本身较慢。1. 评估步骤间的依赖性对无依赖的步骤尝试并行执行。2. 为策略决策添加缓存对相似的查询-上下文对复用策略。3. 考虑使用更高效的向量数据库如FAISS的IVF索引或对索引进行量化。7. 生产环境最佳实践与进阶优化将GRASP框架用于实际项目时需要考虑以下工程化问题7.1 性能优化策略缓存为StrategyDecider增加缓存层如使用functools.lru_cache或Redis缓存(query, context)到RetrievalStrategy的映射避免重复调用LLM。异步执行使用asyncio和异步客户端如openai.AsyncOpenAI并行执行多个步骤的检索和LLM调用大幅减少端到端延迟。索引优化对于海量知识库使用更高效的索引算法如HNSW并定期进行索引压缩和清理。7.2 检索质量提升混合检索Hybrid Search在MultiGranularityRetriever.retrieve方法中不仅使用向量相似度检索还可以融合关键词检索如BM25的结果提高召回率。重排序Re-ranking在初步召回后使用一个更精细的交叉编码器模型如bge-reranker对结果进行重排序将最相关的结果排到前面进一步提升精度。查询扩展Query Expansion在将步骤查询送入检索器前先用LLM生成几个相关的问法或关键词合并进行检索以覆盖更广的信息面。7.3 智能体逻辑增强反思与迭代Reflection在每一步执行后让LLM对当前结果进行自我评估和反思。如果置信度低或发现矛盾可以触发回溯重新规划或调整检索策略。工具扩展将GRASP检索器作为智能体的一个核心工具并与其他工具如计算器、代码执行器、网络搜索API结合处理更复杂的任务。记忆机制为智能体添加短期或长期记忆记住历史对话和推理过程避免在长对话中重复检索相同信息。7.4 可观测性与评估详细日志记录记录每一步的策略决策、检索到的节点ID与分数、LLM的输入输出。这对于调试和效果分析至关重要。构建评估集针对你的业务领域构建一个包含复杂问题的测试集并标注期望答案或关键点。定期运行评估监控GRASP框架的效果。A/B测试在生产环境中可以对比使用GRASP和固定粒度检索的答案质量用数据证明其价值。GRASP框架代表了一种更智能、更自适应的RAG演进方向。它通过将检索决策与推理过程深度耦合有效解决了复杂问答中的信息粒度失配问题。实现这一框架需要仔细设计策略决策、多粒度索引和智能体工作流之间的交互。本文提供的代码是一个完整的起点你可以根据具体业务需求对其中的检索策略、规划逻辑和提示词进行深度定制和优化。