泰州市网站建设_网站建设公司_前后端分离_seo优化

文章介绍Deep Thinking RAG架构，将RAG与Agent技术融合，通过LangGraph实现的循环推理机制，解决传统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 适用场景

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

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

AI时代，未来的就业机会在哪里？

答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具，到自然语言处理、计算机视觉、多模态等核心领域，技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。

掌握大模型技能，就是把握高薪未来。

那么，普通人如何抓住大模型风口？

AI技术的普及对个人能力提出了新的要求，在AI时代，持续学习和适应新技术变得尤为重要。无论是企业还是个人，都需要不断更新知识体系，提升与AI协作的能力，以适应不断变化的工作环境。

因此，这里给大家整理了一份《2025最新大模型全套学习资源》，包括2025最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题等，带你从零基础入门到精通，快速掌握大模型技术！

由于篇幅有限，有需要的小伙伴可以扫码获取！

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 大模型项目实战

学以致用，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

5. 大模型行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

为什么大家都在学AI大模型？

随着AI技术的发展，企业对人才的需求从“单一技术”转向 “AI+行业”双背景。企业对人才的需求从“单一技术”转向 “AI+行业”双背景。金融+AI、制造+AI、医疗+AI等跨界岗位薪资涨幅达30%-50%。

同时很多人面临优化裁员，近期科技巨头英特尔裁员2万人，传统岗位不断缩减，因此转行AI势在必行！

这些资料有用吗？

这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

大模型全套学习资料已整理打包，有需要的小伙伴可以微信扫描下方CSDN官方认证二维码，免费领取【保证100%免费】