德州市网站建设_网站建设公司_UX设计_seo优化

Deep Thinking RAG是一种将RAG与Agent技术融合的新型架构，突破了传统RAG的局限性。它通过四个智能模块（规划代理、检索监督者、多阶段检索漏斗、策略代理）实现从线性链到循环图的跃迁，支持多跳推理、动态知识边界和自适应检索策略。文章详细介绍了其实现方法，并通过量化评估证明其在Context Precision、Context Recall和Answer Correctness等方面显著优于传统RAG，适用于企业问答、金融研报等复杂场景。

引言：传统 RAG 的瓶颈与突破

在大模型应用的浪潮中，RAG（Retrieval-Augmented Generation）已成为构建知识密集型 AI 系统的主流范式。然而，传统的"检索→增强→生成"三段式流水线在面对复杂查询时暴露出明显的局限性：

• 无法处理多跳推理：需要综合多个文档片段才能回答的问题
• 静态知识边界：无法获取实时信息或外部数据源
• 检索策略单一：对所有问题采用相同的检索方法
• 缺乏自我纠错：无法判断检索到的信息是否充分

Deep Thinking RAG架构正是为了突破这些瓶颈而生，这是一种具备自主推理能力的新型 RAG 系统，通过 LangGraph 编排的循环推理机制，实现了从"简单检索"到"深度思考"的跨越。它本质上是将 RAG 与 Agent 技术深度融合——具体 Agent 的构建与策略编排，我们已在之前的“智能体实战合集”(持续更新)中有过大量深入实践，而本文将是 RAG 检索增强工程与 Agent 推理框架的首次融合串联，实现从“工具”向“智能体”的范式跃迁。

由于受限于文章长度和照顾大家的阅读体验，本文仅会列出主要部分代码，完整实战代码在文末提供。视界君创作不易，觉得有所帮助烦劳关注转发点赞！

一、Deep Thinking RAG 核心架构

1.1 系统设计哲学

许多复杂的查询需要推理、思考和明智的决策，才能决定何时采取行动，这与我们面对问题时检索信息的方式非常相似。这就是 RAG pipeline中由Agent驱动的操作发挥作用的地方。下图是Deep Thinking RAG pipeline示意图：

▲Deep Thinking RAG pipeline

1. 计划：首先，代理将复杂的用户查询分解为结构化的多步骤研究计划，并决定每个步骤需要哪个工具（内部文档搜索或网络搜索）。
2. 检索：对于每个步骤，它执行自适应的多阶段检索流程，使用监督器动态选择最佳搜索策略（向量、关键词或混合）。
3. 精炼：然后使用高精度交叉编码器对初始结果进行重新排序，并使用蒸馏代理将最佳证据压缩成简洁的上下文。
4. 反思：每完成一步，智能体都会总结其发现并更新其研究历史，从而逐步建立对问题的理解。
5. 评论：政策制定者随后会审查这段历史，做出战略决策，要么继续进行下一步研究，要么在遇到瓶颈时修改计划，要么结束研究。
6. 综合：研究完成后，最终决策者将从所有来源收集到的所有证据综合成一个单一、全面且可引用的答案。

Deep Thinking RAG 的核心思想是将 RAG 系统从无状态的线性链升级为有状态的循环图。精简一下就是系统主要通过以下四个智能模块协同工作：

用户查询 ↓ ┌─────────────────────┐ │ Planning Agent │ ──→ 分解为多步子任务，选择工具 └──────────┬──────────┘ ↓ ┌─────────────────────────────┐ │ Retrieval Supervisor │ ──→ 动态选择检索策略 │ (Vector/Keyword/Hybrid) │ └──────────┬──────────────────┘ ↓ ┌─────────────────────┐ │ Multi-Stage │ ──→ 检索 → 重排 → 蒸馏 │ Retrieval Funnel │ └──────────┬──────────┘ ↓ ┌─────────────────────┐ │ Policy Agent │ ──→ 判断：继续 or 完成 └─────────────────────┘ ↓ 最终答案（含引用）

1.2 状态管理：系统的"记忆中枢"

系统通过RAGState维护全局状态，记录推理历程：

from typing import List, Dict, TypedDict, Literal, Optional from langchain_core.pydantic_v1 import BaseModel, Field class Step(BaseModel): """单步任务定义""" sub_question: str = Field(description="具体的子问题") tool: Literal["search_10k", "search_web"] = Field(description="工具选择") keywords: List[str] = Field(description="关键检索词") document_section: Optional[str] = Field(description="目标文档章节") class Plan(BaseModel): """执行计划""" steps: List[Step] class RAGState(TypedDict): """全局状态""" original_question: str # 原始问题 plan: Plan # 执行计划 past_steps: List[Dict] # 历史推理记录 current_step_index: int # 当前步骤 retrieved_docs: List[Document] # 检索结果 reranked_docs: List[Document] # 重排后文档 synthesized_context: str # 蒸馏上下文 final_answer: str # 最终答案

核心价值：

• past_steps保存完整的推理链条
• tool字段实现内部文档与外部搜索的智能路由
• 状态持久化支持断点续推和轨迹分析

二、Deep Thinking RAG 关键组件

2.1 工具感知的规划代理

规划代理是系统的"战略大脑"，负责查询分解与工具选择：

from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI planner_prompt = ChatPromptTemplate.from_messages([ ("system", """你是研究策略专家。将复杂查询拆解为多步计划。 可用工具： 1. search_10k：检索 NVIDIA 2023 年 10-K 财报（历史数据） 2. search_web：搜索互联网最新信息（实时数据） 要求： - 每个子问题选择最合适的工具 - 对 search_10k 指定目标章节（如 'Item 1A. Risk Factors'）"""), ("human", "查询: {question}") ]) reasoning_llm = ChatOpenAI(model="gpt-4o", temperature=0) planner_agent = planner_prompt | reasoning_llm.with_structured_output(Plan)

实际效果示例：

输入查询：

“基于 NVIDIA 2023 年 10-K，识别竞争风险。然后查找 2024 年 AMD AI 芯片战略，分析其对 NVIDIA 风险的影响。”

输出计划：

步骤 1: 工具=search_10k, 章节=Item 1A. Risk Factors 子问题: NVIDIA 列出了哪些竞争相关风险？ 步骤 2: 工具=search_web 子问题: AMD 在 2024 年推出了哪些 AI 芯片产品？ 步骤 3: 工具=search_10k, 章节=Item 1A. Risk Factors 子问题: AMD 新战略如何影响 NVIDIA 的竞争风险？

2.2 自适应检索监督者

不同问题需要不同检索策略。监督者通过 LLM 分析查询特征，动态选择最优方法：

class RetrievalDecision(BaseModel): strategy: Literal["vector_search", "keyword_search", "hybrid_search"] justification: str retrieval_supervisor_prompt = ChatPromptTemplate.from_messages([ ("system", """选择最佳检索策略： - vector_search：概念性查询（如"公司对气候变化的态度"） - keyword_search：精确查询（如"Item 1A"、"Hopper 架构"） - hybrid_search：复杂查询（结合语义与关键词）"""), ("human", "查询: {sub_question}") ]) retrieval_supervisor = retrieval_supervisor_prompt | \\ reasoning_llm.with_structured_output(RetrievalDecision)

决策示例：

# 查询 1："2023 财年 Compute 部门营收" # 决策：keyword_search（包含特定术语，需精确匹配） # 查询 2："市场竞争的总体情绪" # 决策：vector_search（概念性，需语义理解）

2.3 多策略检索 + 高精度重排

系统实现三种检索策略，并通过交叉编码器（Cross-Encoder）进行高精度重排：

from langchain_community.vectorstores import Chroma from rank_bm25 import BM25Okapi from sentence_transformers import CrossEncoder # 策略 1：向量检索（支持元数据过滤） def vector_search_only(query: str, section_filter: str = None, k: int = 10): filter_dict = {"section": section_filter} if section_filter else None return vector_store.similarity_search(query, k=k, filter=filter_dict) # 策略 2：BM25 关键词检索 def bm25_search_only(query: str, k: int = 10): tokenized_query = query.split(" ") scores = bm25.get_scores(tokenized_query) top_k_indices = np.argsort(scores)[::-1][:k] return [doc_map[doc_ids[i]] for i in top_k_indices] # 策略 3：混合检索（RRF 融合） def hybrid_search(query: str, section_filter: str = None, k: int = 10): bm25_docs = bm25_search_only(query, k=k) semantic_docs = vector_search_only(query, section_filter, k=k) # 倒数排名融合（Reciprocal Rank Fusion） return rrf_fusion(bm25_docs, semantic_docs, k) # 高精度重排序 reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') def rerank_documents(query: str, documents: List[Document], top_n: int = 3): pairs = [(query, doc.page_content) for doc in documents] scores = reranker.predict(pairs) doc_scores = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True) return [doc for doc, _ in doc_scores[:top_n]]

关键优势：

• 第一阶段（广度）：检索 Top-K (如 10) 个候选文档，最大化召回
• 第二阶段（精度）：交叉编码器深度对比 query-document，筛选 Top-N (如 3) 高质量文档
• 元数据过滤：通过section_filter精确定位目标章节

2.4 上下文蒸馏与外部工具

# 上下文蒸馏代理 distiller_prompt = ChatPromptTemplate.from_messages([ ("system", "将检索片段合成简洁上下文，直接回答：{question}"), ("human", "检索文档:\\n{context}") ]) distiller_agent = distiller_prompt | reasoning_llm | StrOutputParser() # 外部搜索工具 from langchain_community.tools.tavily_search import TavilySearchResults web_search_tool = TavilySearchResults(k=3) def web_search_function(query: str) -> List[Document]: results = web_search_tool.invoke({"query": query}) return [Document(page_content=res["content"], metadata={"source": res["url"]}) for res in results]

三、LangGraph 编排：构建推理闭环

3.1 核心图节点定义

from langgraph.graph import StateGraph, END def plan_node(state: RAGState): """生成执行计划""" plan = planner_agent.invoke({"question": state["original_question"]}) return {"plan": plan, "current_step_index": 0, "past_steps": []} def retrieval_node(state: RAGState): """内部文档检索（自适应策略）""" step = state["plan"].steps[state["current_step_index"]] # 1. 查询重写 rewritten_query = query_rewriter_agent.invoke({ "sub_question": step.sub_question, "keywords": step.keywords, "past_context": get_past_context(state['past_steps']) }) # 2. 策略选择 decision = retrieval_supervisor.invoke({"sub_question": rewritten_query}) # 3. 执行检索 if decision.strategy == 'vector_search': docs = vector_search_only(rewritten_query, step.document_section, k=10) elif decision.strategy == 'keyword_search': docs = bm25_search_only(rewritten_query, k=10) else: docs = hybrid_search(rewritten_query, step.document_section, k=10) return {"retrieved_docs": docs} def web_search_node(state: RAGState): """外部网络搜索""" step = state["plan"].steps[state["current_step_index"]] docs = web_search_function(step.sub_question) return {"retrieved_docs": docs} def rerank_node(state: RAGState): """文档重排序""" step = state["plan"].steps[state["current_step_index"]] reranked = rerank_documents(step.sub_question, state["retrieved_docs"], top_n=3) return {"reranked_docs": reranked} def compression_node(state: RAGState): """上下文蒸馏""" step = state["plan"].steps[state["current_step_index"]] context = "\\n\\n".join([doc.page_content for doc in state["reranked_docs"]]) synthesized = distiller_agent.invoke({ "question": step.sub_question, "context": context }) return {"synthesized_context": synthesized} def reflection_node(state: RAGState): """总结当前步骤""" summary = reflection_agent.invoke({ "sub_question": state["plan"].steps[state["current_step_index"]].sub_question, "context": state['synthesized_context'] }) new_past_step = { "step_index": state["current_step_index"] + 1, "summary": summary, "retrieved_docs": state['reranked_docs'] } return { "past_steps": state["past_steps"] + [new_past_step], "current_step_index": state["current_step_index"] + 1 } def final_answer_node(state: RAGState): """生成最终答案（含引用）""" # 整合所有证据 final_context = aggregate_all_evidence(state['past_steps']) final_answer = final_answer_agent.invoke({ "question": state['original_question'], "context": final_context }) return {"final_answer": final_answer}

3.2 条件边与控制策略

def route_by_tool(state: RAGState) -> str: """根据工具类型路由""" return state["plan"].steps[state["current_step_index"]].tool def should_continue(state: RAGState) -> str: """判断是否继续推理""" # 检查 1：计划是否完成 if state["current_step_index"] >= len(state["plan"].steps): return "finish" # 检查 2：是否超过最大迭代 if state["current_step_index"] >= 7: return "finish" # 检查 3：LLM 自我判断 history = get_past_context(state['past_steps']) decision = policy_agent.invoke({ "question": state["original_question"], "history": history }) return "finish" if decision.next_action == "FINISH" else "continue"

3.3 图构建与编译

# 创建状态图 graph = StateGraph(RAGState) # 添加节点 graph.add_node("plan", plan_node) graph.add_node("retrieve_10k", retrieval_node) graph.add_node("retrieve_web", web_search_node) graph.add_node("rerank", rerank_node) graph.add_node("compress", compression_node) graph.add_node("reflect", reflection_node) graph.add_node("generate_final_answer", final_answer_node) # 设置入口 graph.set_entry_point("plan") # 条件路由：根据工具类型 graph.add_conditional_edges( "plan", route_by_tool, {"search_10k": "retrieve_10k", "search_web": "retrieve_web"} ) # 处理流程 graph.add_edge("retrieve_10k", "rerank") graph.add_edge("retrieve_web", "rerank") graph.add_edge("rerank", "compress") graph.add_edge("compress", "reflect") # 循环判断 graph.add_conditional_edges( "reflect", should_continue, {"continue": "plan", "finish": "generate_final_answer"} ) graph.add_edge("generate_final_answer", END) # 编译 deep_thinking_rag = graph.compile()

四、实战对比：Baseline vs Deep Thinking

4.1 测试查询

complex_query = """基于 NVIDIA 2023 年 10-K 财报，识别其与竞争相关的关键风险。 然后，查找 2024 年（财报发布后）关于 AMD 的 AI 芯片战略的最新新闻， 并分析这一新战略如何直接加剧或应对 NVIDIA 所述的竞争风险。"""

4.2 Baseline RAG（失败案例）

baseline_result = baseline_rag_chain.invoke(complex_query)

输出：

基于提供的上下文，NVIDIA 在 2023 年 10-K 中提到了激烈的市场竞争… 然而，关于 AMD 在 2024 年的 AI 芯片战略，提供的文档中没有相关信息。 因此我无法分析其对 NVIDIA 风险的影响。

失败原因：

• 静态知识库无法获取 2024 年信息
• 单次检索无法分解多跳逻辑

4.3 Deep Thinking RAG（成功案例）

# 执行推理循环 final_state = None for chunk in deep_thinking_rag.stream({"original_question": complex_query}): final_state = chunk print(final_state['final_answer'])

输出：

NVIDIA 在其 2023 年 10-K 财报的"Item 1A. Risk Factors"中明确指出，半导体行业竞争极为激烈，技术变革迅速，且面临来自 AMD、Intel 等竞争对手在 GPU 和 AI 加速器市场的压力 [来源: Item 1A. Risk Factors]。

根据 2024 年的最新报道，AMD 推出了 MI300X AI 加速器，该产品直接对标 NVIDIA 的 H100，并已被 Microsoft Azure、Oracle Cloud 等主要云服务商采用[来源: https://techcrunch.com/amd-mi300x-2024]。

这一战略 …

成功要素：

✅ 自动分解为 3 个子任务

✅ 正确路由到内部文档（步骤1、3）和外部搜索（步骤2）

✅ 成功综合跨源信息，生成连贯分析

✅ 提供完整的信息溯源

五、量化评估：RAGAs 验证

5.1 评估框架

from datasets import Dataset from ragas import evaluate from ragas.metrics import ( context_precision, # 检索精度 context_recall, # 检索召回 faithfulness, # 答案忠实度 answer_correctness # 答案正确性 ) # 构建评估数据 eval_data = { 'question': [complex_query, complex_query], 'answer': [baseline_result, final_state['final_answer']], 'contexts': [baseline_contexts, advanced_contexts], 'ground_truth': [ground_truth, ground_truth] } result = evaluate(Dataset.from_dict(eval_data), metrics=[context_precision, context_recall, faithfulness, answer_correctness])

5.2 评估结果

核心发现：

1. Context Precision（+100%）：

• Baseline：检索到大量无关章节（精度 50%）
• Deep Thinking：通过元数据过滤精确定位（精度 100%）

2. Context Recall（+203%）：

• Baseline：完全遗漏 2024 年外部信息（召回 33%）
• Deep Thinking：通过工具路由获取所有必需信息（召回 100%）

3. Answer Correctness（+148%）：

• Baseline：只完成了部分任务（40% 正确）
• Deep Thinking：实现近乎完美的多源综合分析（99% 正确）

六、生产化建议

6.1 性能优化

# 1. 启用 Redis 缓存（降低 70% 重复查询成本） from langchain.cache import RedisCache set_llm_cache(RedisCache(redis_client)) # 2. LangSmith 可观测性（追踪所有推理轨迹） os.environ["LANGSMITH_TRACING"] = "true" os.environ["LANGSMITH_PROJECT"] = "DeepThinking-RAG-Prod" # 3. 策略模型蒸馏（将 GPT-4o 替换为微调的 Llama-3-8B） # 推理速度提升 10x，成本降低 95%

6.2 容错与降级

def safe_retrieval_node(state: RAGState): try: return retrieval_node(state) except Exception as e: logger.error(f"检索失败: {e}") # 降级：返回缓存的相关文档 return {"retrieved_docs": get_cached_docs(state)}

6.3 扩展方向

1. 多模态支持：整合图表、视频等非文本信息
2. 并行执行：独立子任务并行处理，降低延迟
3. 强化学习：通过用户反馈优化策略选择
4. 个性化定制：为不同场景训练专属策略模型

七、总结

7.1 Deep Thinking RAG 的核心价值

7.2 技术突破点

1. 从工具到智能体：系统具备规划、决策、反思能力
2. 从单源到多源：无缝整合内部知识库与外部实时信息
3. 从固定到自适应：根据任务特性动态优化检索策略
4. 从结果到过程：每个结论都附带完整的推理轨迹

7.3 适用场景

• 企业智能问答：结合内部文档与行业动态
• 金融研报生成：整合财报、新闻、市场数据
• 法律案例分析：关联法条、判例、最新裁决
• 科研文献综述：跨论文、跨时间的知识图谱构建

如何学习AI大模型？

如果你对AI大模型入门感兴趣，那么你需要的话可以点击这里大模型重磅福利：入门进阶全套104G学习资源包免费分享！

这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

这是一份大模型从零基础到进阶的学习路线大纲全览，小伙伴们记得点个收藏！

第一阶段：从大模型系统设计入手，讲解大模型的主要方法；

第二阶段：在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段：大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段：大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段：大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段：以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段：以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

100套AI大模型商业化落地方案

大模型全套视频教程

200本大模型PDF书籍

👉学会后的收获：👈

• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

LLM面试题合集

大模型产品经理资源合集

大模型项目实战合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓