连云港市网站建设_网站建设公司_服务器部署_seo优化

引言：传统 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 适用场景

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

获取完整代码： 本文为了精简篇幅，省略了部分实现细节（如查询重写、策略代理、文档预处理等）。完整的生产级代码（含详细注释）请后台回复“DeepThinkingRAG”即可获取。

参考资源：

• 论文：DeepRAG - Deep Reasoning for Retrieval-Augmented Generation
• 框架：LangGraph 官方文档
• 评估：RAGAs Evaluation Framework

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

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