企业官网建设流程全解析

梁平网站建设,网络管理是什么,wordpress js时钟,wordpress 搜索过滤​ 【个人主页#xff1a;玄同765】 大语言模型#xff08;LLM#xff09;开发工程师#xff5c;中国传媒大学数字媒体技术#xff08;智能交互与游戏设计#xff09; 深耕领域#xff1a;大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调 技术栈#xff1a;Pyth…​【个人主页玄同765】大语言模型LLM开发工程师中国传媒大学·数字媒体技术智能交互与游戏设计深耕领域大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调技术栈Python / LangChain/RAGDifyRedisMilvus| SQL/NumPy | FastAPIDocker ️工程能力专注模型工程化部署、知识库构建与优化擅长全流程解决方案「让AI交互更智能让技术落地更高效」欢迎技术探讨/项目合作 关注我解锁大模型与智能交互的无限可能核心内容本文将深入解析传统RAG与Agentic RAG的核心差异通过对比架构设计、实现方式和实战案例帮助你理解为什么Agentic RAG是下一代AI应用的关键技术。一、RAG技术概述1.1 什么是RAGRAGRetrieval-Augmented Generation检索增强生成是一种将外部知识检索与大语言模型生成能力相结合的技术架构。它通过从外部知识库中检索相关信息并将其作为上下文提供给LLM从而生成更准确、更可靠的回答。1.2 RAG的核心价值┌─────────────────────────────────────────────────────────────┐ │ RAG核心价值 │ ├─────────────────────────────────────────────────────────────┤ │ ✅ 解决知识截止问题 - 访问最新、私有数据 │ │ ✅ 减少幻觉生成 - 基于检索到的真实信息 │ │ ✅ 提供可溯源回答 - 引用具体文档来源 │ │ ✅ 降低微调成本 - 无需重新训练模型 │ │ ✅ 支持动态更新 - 知识库可实时更新 │ └─────────────────────────────────────────────────────────────┘1.3 RAG的典型应用场景场景说明示例企业知识库问答基于内部文档的智能客服产品手册查询、HR政策咨询法律文档分析合同审查、案例检索法规查询、判例匹配医疗辅助诊断医学文献检索症状分析、治疗方案推荐学术研究助手论文检索与综述文献综述、研究趋势分析代码智能助手技术文档查询API使用、最佳实践查询二、传统RAG架构详解2.1 传统RAG的工作流程传统RAG采用线性管道架构遵循固定的处理流程┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ 用户查询 │ -- │ 向量检索 │ -- │ 生成回答 │ │ (Query) │ │ (Retrieve) │ │ (Generate) │ └──────────────┘ └──────────────┘ └──────────────┘ │ v ┌──────────────┐ │ 向量数据库 │ │ (Vector DB) │ └──────────────┘标准流程索引阶段离线文档加载 → 文本分割 → 嵌入生成 → 向量存储查询阶段在线查询向量化 → 相似度检索 → 上下文构建 → LLM生成2.2 传统RAG的实现代码from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_classic.chains.combine_documents import create_stuff_documents_chain from langchain_classic.chains.retrieval import create_retrieval_chain from langchain_core.prompts import ChatPromptTemplate # 1. 加载和分割文档 loader TextLoader(knowledge_base.txt) docs loader.load() splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks splitter.split_documents(docs) # 2. 创建向量数据库 embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(chunks, embeddings) # 3. 创建检索器 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 4. 创建RAG链 prompt ChatPromptTemplate.from_messages([ (system, 基于以下上下文回答问题:\n{context}), (human, {input}) ]) document_chain create_stuff_documents_chain(ChatOpenAI(), prompt) rag_chain create_retrieval_chain(retriever, document_chain) # 5. 执行查询 response rag_chain.invoke({input: 什么是RAG技术}) print(response[answer])2.3 传统RAG的局限性局限性说明影响单次检索只执行一次检索无法迭代优化复杂问题检索不充分静态流程固定流程无法根据问题调整简单问题过度检索无自我纠错无法验证生成内容的准确性可能产生幻觉缺乏推理不能进行多步推理和规划复杂查询处理能力弱上下文限制检索结果一次性输入LLM长文档处理能力受限三、Agentic RAG下一代RAG架构3.1 什么是Agentic RAGAgentic RAG智能体化RAG是将AI Agent的智能决策能力引入RAG系统的下一代架构。它不再遵循固定的线性流程而是通过智能体动态决策每个步骤的执行方式。3.2 Agentic RAG vs 传统RAG┌─────────────────────────────────────────────────────────────────────┐ │ 架构对比传统RAG vs Agentic RAG │ ├─────────────────────────────────────────────────────────────────────┤ │ │ │ 传统RAG线性管道 Agentic RAG智能决策网络 │ │ ┌─────────┐ ┌─────────┐ │ │ │ 查询 │ │ 查询 │ │ │ └────┬────┘ └────┬────┘ │ │ │ │ │ │ v v │ │ ┌─────────┐ ┌───────────────┐ │ │ │ 检索 │ │ 查询分析Agent │ ← 判断是否需要检索 │ │ └────┬────┘ └───────┬───────┘ │ │ │ │ │ │ v v │ │ ┌─────────┐ ┌───────────────┐ │ │ │ 生成 │ │ 检索Agent │ ← 决定检索策略 │ │ └─────────┘ └───────┬───────┘ │ │ │ │ │ v │ │ ┌───────────────┐ │ │ │ 验证Agent │ ← 检查结果质量 │ │ └───────┬───────┘ │ │ │ │ │ v │ │ ┌───────────────┐ │ │ │ 生成Agent │ ← 生成最终回答 │ │ └───────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘3.3 Agentic RAG的核心特征特征传统RAGAgentic RAG决策能力无固定流程有动态决策检索次数单次可多次迭代查询改写无智能改写优化自我验证无幻觉检测与纠正工具使用仅向量检索多工具协同推理能力无多步推理规划四、Agentic RAG架构深度解析4.1 Agentic RAG的核心组件# Agentic RAG核心组件 class AgenticRAGComponents: Agentic RAG的核心组件 def __init__(self): # 1. 查询分析器 - 理解用户意图 self.query_analyzer QueryAnalyzer() # 2. 检索策略选择器 - 选择最佳检索方式 self.retrieval_selector RetrievalSelector() # 3. 文档评估器 - 评估检索结果质量 self.doc_evaluator DocumentEvaluator() # 4. 答案生成器 - 生成最终回答 self.answer_generator AnswerGenerator() # 5. 幻觉检测器 - 验证答案准确性 self.hallucination_checker HallucinationChecker() # 6. 查询改写器 - 优化检索查询 self.query_rewriter QueryRewriter()4.2 Agentic RAG的工作流程┌─────────────────────────────────────────────────────────────────┐ │ Agentic RAG 工作流程 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ │ │ │ 用户输入 │ │ │ └────┬─────┘ │ │ │ │ │ v │ │ ┌─────────────────────────────────────┐ │ │ │ Step 1: 查询分析 (Query Analysis) │ │ │ │ - 理解查询意图 │ │ │ │ - 判断复杂度 │ │ │ │ - 决定是否需要检索 │ │ │ └──────────────┬──────────────────────┘ │ │ │ │ │ ┌─────────┴─────────┐ │ │ │ │ │ │ v v │ │ ┌─────────┐ ┌─────────┐ │ │ │直接回答 │ │需要检索 │ │ │ └────┬────┘ └────┬────┘ │ │ │ │ │ │ v v │ │ ┌──────────┐ ┌─────────────────────────────────────┐ │ │ │ 生成答案 │ │ Step 2: 检索策略选择 │ │ │ └──────────┘ │ - 选择检索工具 │ │ │ │ - 确定检索参数 │ │ │ └──────────────┬──────────────────────┘ │ │ │ │ │ v │ │ ┌─────────────────────────────────────┐ │ │ │ Step 3: 执行检索 │ │ │ │ - 向量检索 │ │ │ │ - 关键词检索 │ │ │ │ - 网络搜索 │ │ │ └──────────────┬──────────────────────┘ │ │ │ │ │ v │ │ ┌─────────────────────────────────────┐ │ │ │ Step 4: 结果评估 │ │ │ │ - 评估文档相关性 │ │ │ │ - 判断是否足够 │ │ │ └──────────────┬──────────────────────┘ │ │ │ │ │ ┌────────────┴────────────┐ │ │ │ │ │ │ v v │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 结果充足 │ │ 结果不足 │ │ │ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ v v │ │ ┌──────────┐ ┌─────────────────────┐ │ │ │生成答案 │ │ Step 5: 查询改写 │ │ │ └──────────┘ │ - 优化查询词 │ │ │ │ - 重新检索 │ │ │ └──────────┬──────────┘ │ │ │ │ │ └──────→ (回到Step 3)│ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ Step 6: 答案验证 (Hallucination Check) │ │ │ │ - 验证答案是否基于检索内容 │ │ │ │ - 检测幻觉 │ │ │ └────────────────────────┬────────────────────────────────┘ │ │ │ │ │ ┌─────────┴─────────┐ │ │ │ │ │ │ v v │ │ ┌──────────┐ ┌──────────┐ │ │ │ 验证通过 │ │ 验证失败 │ │ │ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ v v │ │ ┌──────────┐ ┌──────────┐ │ │ │ 返回答案 │ │ 重新生成 │ │ │ └──────────┘ └──────────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘4.3 Agentic RAG的关键技术4.3.1 查询分析与路由from typing import Literal from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class QueryAnalysis(BaseModel): 查询分析结果 query_type: Literal[factual, analytical, creative, ambiguous] Field( description查询类型事实性、分析性、创造性、模糊 ) complexity: Literal[simple, medium, complex] Field( description复杂度级别 ) needs_retrieval: bool Field( description是否需要检索外部知识 ) reasoning_steps: int Field( description预估需要的推理步骤数 ) def analyze_query(query: str) - QueryAnalysis: 分析用户查询 prompt ChatPromptTemplate.from_messages([ (system, 分析用户查询的特征包括 1. 查询类型事实性(需要具体信息)、分析性(需要推理)、创造性(需要生成)、模糊(需要澄清) 2. 复杂度简单(直接回答)、中等(需要一些推理)、复杂(需要多步推理) 3. 是否需要检索判断是否需要外部知识 4. 推理步骤预估需要的推理步骤数), (human, 查询: {query}) ]) llm ChatOpenAI(modelgpt-4o-mini) structured_llm llm.with_structured_output(QueryAnalysis) chain prompt | structured_llm return chain.invoke({query: query}) # 使用示例 result analyze_query(比较Transformer和RNN在序列建模中的优缺点) print(f查询类型: {result.query_type}) print(f复杂度: {result.complexity}) print(f需要检索: {result.needs_retrieval})4.3.2 自适应检索策略from typing import List, Optional from langchain_core.documents import Document class AdaptiveRetriever: 自适应检索器 def __init__(self, vectorstore, web_search_toolNone): self.vectorstore vectorstore self.web_search web_search_tool def retrieve(self, query: str, strategy: str auto) - List[Document]: 根据策略执行检索 Args: query: 查询字符串 strategy: 检索策略 (vector, keyword, web, hybrid, auto) if strategy auto: strategy self._select_strategy(query) if strategy vector: return self._vector_search(query) elif strategy keyword: return self._keyword_search(query) elif strategy web: return self._web_search(query) elif strategy hybrid: return self._hybrid_search(query) else: return self._vector_search(query) def _select_strategy(self, query: str) - str: 自动选择检索策略 # 简单启发式规则 if any(word in query.lower() for word in [latest, recent, news, 2024, 2025]): return web # 需要最新信息 elif any(word in query.lower() for word in [compare, difference, vs]): return hybrid # 需要对比使用混合检索 elif len(query.split()) 5: return keyword # 短查询关键词检索 else: return vector # 默认向量检索 def _vector_search(self, query: str) - List[Document]: 向量相似度搜索 retriever self.vectorstore.as_retriever(search_kwargs{k: 5}) return retriever.invoke(query) def _keyword_search(self, query: str) - List[Document]: 关键词搜索使用BM25 from langchain_community.retrievers import BM25Retriever # 实现关键词检索逻辑 pass def _web_search(self, query: str) - List[Document]: 网络搜索 if self.web_search: results self.web_search.invoke(query) return [Document(page_contentr[content], metadata{source: r[url]}) for r in results] return [] def _hybrid_search(self, query: str) - List[Document]: 混合检索向量 关键词 vector_results self._vector_search(query) keyword_results self._keyword_search(query) # 融合两种检索结果 return self._merge_results(vector_results, keyword_results) def _merge_results(self, results1: List[Document], results2: List[Document]) - List[Document]: 合并检索结果并去重 seen set() merged [] for doc in results1 results2: key doc.page_content[:100] # 使用内容前100字符作为去重键 if key not in seen: seen.add(key) merged.append(doc) return merged[:10] # 返回前10个结果4.3.3 迭代检索与查询改写class IterativeRetriever: 迭代检索器 - 支持多轮检索 def __init__(self, llm, retriever, max_iterations3): self.llm llm self.retriever retriever self.max_iterations max_iterations def retrieve_with_refinement(self, query: str) - List[Document]: 通过迭代优化进行检索 策略 1. 初始检索 2. 评估结果质量 3. 如果质量不足改写查询并重新检索 4. 重复直到满意或达到最大迭代次数 all_docs [] current_query query for i in range(self.max_iterations): # 执行检索 docs self.retriever.invoke(current_query) all_docs.extend(docs) # 评估检索质量 quality_score self._evaluate_quality(docs, query) if quality_score 0.8: # 质量足够 break if i self.max_iterations - 1: # 还有迭代次数 # 改写查询 current_query self._rewrite_query(current_query, docs, query) print(f迭代 {i1}: 查询改写为 - {current_query}) # 去重并返回 return self._deduplicate(all_docs) def _evaluate_quality(self, docs: List[Document], query: str) - float: 评估检索结果质量 if not docs: return 0.0 # 使用LLM评估相关性 prompt f评估以下文档与用户查询的相关性0-1分 查询: {query} 文档: {docs[0].page_content[:500]} 只返回一个0到1之间的数字表示相关性得分。 try: response self.llm.invoke(prompt) score float(response.content.strip()) return min(max(score, 0.0), 1.0) # 确保在0-1范围内 except: return 0.5 # 默认中等质量 def _rewrite_query(self, current_query: str, docs: List[Document], original_query: str) - str: 基于已有结果改写查询 prompt f基于原始查询和已检索到的信息生成一个新的查询以获取更多相关信息。 原始查询: {original_query} 当前查询: {current_query} 已检索到的信息片段: {chr(10).join([f- {doc.page_content[:200]}... for doc in docs[:3]])} 请生成一个更具体、更有针对性的查询以获取缺失的信息。只返回查询字符串。 response self.llm.invoke(prompt) return response.content.strip() def _deduplicate(self, docs: List[Document]) - List[Document]: 文档去重 seen set() unique [] for doc in docs: key hash(doc.page_content[:200]) if key not in seen: seen.add(key) unique.append(doc) return unique4.3.4 幻觉检测与自我纠正from typing import Tuple class HallucinationDetector: 幻觉检测器 def __init__(self, llm): self.llm llm def check(self, answer: str, context: List[Document]) - Tuple[bool, str]: 检测答案中是否存在幻觉 Returns: (是否存在幻觉, 解释说明) context_text \n\n.join([doc.page_content for doc in context]) prompt f验证以下答案是否完全基于提供的上下文是否存在幻觉编造信息。 上下文: {context_text[:2000]} 答案: {answer} 请分析 1. 答案中的每个事实是否都能在上下文中找到依据 2. 是否存在上下文之外的信息 3. 是否存在过度推断 以JSON格式返回 {{ has_hallucination: true/false, explanation: 详细解释, unsupported_claims: [具体的不受支持的陈述] }} try: response self.llm.invoke(prompt) # 解析JSON响应 import json result json.loads(response.content) return result.get(has_hallucination, False), result.get(explanation, ) except: # 如果解析失败保守起见认为可能存在幻觉 return True, 无法验证答案 def correct(self, answer: str, context: List[Document]) - str: 纠正幻觉基于上下文重新生成答案 context_text \n\n.join([doc.page_content for doc in context]) prompt f基于以下上下文重新生成答案。确保答案完全基于上下文不要添加外部信息。 上下文: {context_text[:2000]} 原答案可能包含幻觉: {answer} 请生成一个准确、基于上下文的答案 response self.llm.invoke(prompt) return response.content class SelfCorrectingRAG: 自我纠正的RAG系统 def __init__(self, llm, retriever, max_retries2): self.llm llm self.retriever retriever self.detector HallucinationDetector(llm) self.max_retries max_retries def generate(self, query: str) - dict: 生成答案包含自我纠正机制 # 检索相关文档 docs self.retriever.invoke(query) context \n\n.join([doc.page_content for doc in docs]) # 初始生成 answer self._generate_answer(query, context) # 幻觉检测与纠正 for i in range(self.max_retries): has_hallucination, explanation self.detector.check(answer, docs) if not has_hallucination: break print(f检测到幻觉正在纠正... (尝试 {i1}/{self.max_retries})) print(f原因: {explanation}) # 重新生成 answer self.detector.correct(answer, docs) return { answer: answer, context: docs, has_hallucination: has_hallucination if has_hallucination in locals() else False } def _generate_answer(self, query: str, context: str) - str: 基于上下文生成答案 prompt f基于以下上下文回答问题。 上下文: {context} 问题: {query} 答案: response self.llm.invoke(prompt) return response.content五、使用LangGraph构建Agentic RAG5.1 LangGraph简介LangGraph是LangChain的扩展库专门用于构建复杂的多智能体工作流。它使用图Graph结构来定义状态机非常适合实现Agentic RAG。5.2 Agentic RAG状态机设计from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langchain_core.documents import Document import operator class AgenticRAGState(TypedDict): Agentic RAG的状态定义 query: str # 用户查询 query_type: str # 查询类型 documents: List[Document] # 检索到的文档 rewritten_queries: List[str] # 改写后的查询历史 iteration: int # 当前迭代次数 answer: str # 生成的答案 needs_retrieval: bool # 是否需要检索 quality_score: float # 答案质量评分 has_hallucination: bool # 是否存在幻觉 def create_agentic_rag_graph(llm, retriever): 创建Agentic RAG的状态图 # 1. 查询分析节点 def analyze_query_node(state: AgenticRAGState) - AgenticRAGState: query state[query] # 分析查询 analysis analyze_query(query) return { **state, query_type: analysis.query_type, needs_retrieval: analysis.needs_retrieval, iteration: 0, rewritten_queries: [] } # 2. 检索节点 def retrieve_node(state: AgenticRAGState) - AgenticRAGState: query state[query] # 如果有改写查询使用最新的 if state[rewritten_queries]: query state[rewritten_queries][-1] # 执行检索 docs retriever.invoke(query) return { **state, documents: docs, iteration: state[iteration] 1 } # 3. 评估节点 def evaluate_node(state: AgenticRAGState) - AgenticRAGState: docs state[documents] query state[query] # 评估检索质量 detector HallucinationDetector(llm) quality_score detector._evaluate_quality(docs, query) return { **state, quality_score: quality_score } # 4. 查询改写节点 def rewrite_node(state: AgenticRAGState) - AgenticRAGState: query state[query] docs state[documents] # 改写查询 iterative_retriever IterativeRetriever(llm, retriever) rewritten iterative_retriever._rewrite_query( state[rewritten_queries][-1] if state[rewritten_queries] else query, docs, query ) return { **state, rewritten_queries: state[rewritten_queries] [rewritten] } # 5. 生成答案节点 def generate_node(state: AgenticRAGState) - AgenticRAGState: query state[query] docs state[documents] context \n\n.join([doc.page_content for doc in docs]) # 生成答案 prompt f基于以下上下文回答问题。 上下文: {context} 问题: {query} 请提供详细、准确的答案 response llm.invoke(prompt) return { **state, answer: response.content } # 6. 幻觉检测节点 def hallucination_check_node(state: AgenticRAGState) - AgenticRAGState: answer state[answer] docs state[documents] detector HallucinationDetector(llm) has_hallucination, _ detector.check(answer, docs) return { **state, has_hallucination: has_hallucination } # 7. 直接回答节点无需检索 def direct_answer_node(state: AgenticRAGState) - AgenticRAGState: query state[query] response llm.invoke(f回答以下问题{query}) return { **state, answer: response.content, documents: [] } # 构建图 workflow StateGraph(AgenticRAGState) # 添加节点 workflow.add_node(analyze, analyze_query_node) workflow.add_node(retrieve, retrieve_node) workflow.add_node(evaluate, evaluate_node) workflow.add_node(rewrite, rewrite_node) workflow.add_node(generate, generate_node) workflow.add_node(hallucination_check, hallucination_check_node) workflow.add_node(direct_answer, direct_answer_node) # 设置入口 workflow.set_entry_point(analyze) # 添加条件边 workflow.add_conditional_edges( analyze, lambda state: retrieve if state[needs_retrieval] else direct_answer ) workflow.add_edge(direct_answer, END) workflow.add_edge(retrieve, evaluate) workflow.add_conditional_edges( evaluate, lambda state: ( generate if state[quality_score] 0.7 or state[iteration] 3 else rewrite ) ) workflow.add_edge(rewrite, retrieve) workflow.add_edge(generate, hallucination_check) workflow.add_conditional_edges( hallucination_check, lambda state: ( END if not state[has_hallucination] else generate # 可以添加一个纠正节点 ) ) # 编译图 return workflow.compile() # 使用示例 # graph create_agentic_rag_graph(llm, retriever) # result graph.invoke({query: 什么是Agentic RAG}) # print(result[answer])5.3 可视化工作流def visualize_workflow(graph): 可视化Agentic RAG工作流 try: from IPython.display import Image, display # 生成Mermaid图 mermaid_code graph TD A[查询分析] --|需要检索| B[检索文档] A --|直接回答| C[生成答案] B -- D[评估质量] D --|质量不足| E[改写查询] E -- B D --|质量足够| F[生成答案] F -- G[幻觉检测] G --|存在幻觉| H[纠正答案] H -- G G --|通过检测| I[返回答案] C -- I print(Agentic RAG工作流:) print(mermaid_code) except ImportError: print(请安装IPython以查看可视化图形) # 可视化 # visualize_workflow(graph)六、RAG vs Agentic RAG 实战对比6.1 测试场景设计# 测试用例 test_cases [ { name: 简单事实查询, query: 什么是RAG技术, expected_behavior: 传统RAG和Agentic RAG都应该能很好回答 }, { name: 复杂多跳查询, query: 比较Transformer和RNN在序列建模中的优缺点并说明为什么Transformer更适合并行计算, expected_behavior: Agentic RAG应该进行多步检索和推理 }, { name: 模糊查询, query: 那个新技术, expected_behavior: Agentic RAG应该进行查询澄清或改写 }, { name: 需要最新信息, query: 2024年最新的LLM模型有哪些, expected_behavior: Agentic RAG应该使用网络搜索工具 } ]6.2 性能对比指标传统RAGAgentic RAG说明响应时间快1-2秒较慢3-10秒Agentic需要多步推理准确率中等70-80%高85-95%Agentic能处理更复杂问题复杂问题处理能力弱强Agentic支持多步推理幻觉率中等15-20%低5-10%Agentic有自我验证机制成本低高Agentic需要更多LLM调用可解释性低高Agentic有明确的推理步骤6.3 选择建议使用传统RAG的场景- ✅ 查询简单明确不需要复杂推理- ✅ 对响应时间要求高- ✅ 预算有限需要控制成本- ✅ 知识库稳定不需要频繁更新使用Agentic RAG的场景- ✅ 查询复杂需要多步推理- ✅ 对答案准确性要求极高- ✅ 需要处理模糊或不完整的查询- ✅ 需要最新信息结合网络搜索- ✅ 需要可解释的推理过程七、最佳实践与优化建议7.1 混合架构设计class HybridRAG: 混合RAG架构 - 根据查询复杂度自动选择策略 def __init__(self, simple_rag, agentic_rag): self.simple_rag simple_rag self.agentic_rag agentic_rag self.llm ChatOpenAI(modelgpt-4o-mini) def route_query(self, query: str) - str: 路由查询到合适的RAG系统 prompt f判断以下查询的复杂度选择最适合的处理方式。 查询: {query} 选项: 1. simple - 简单查询直接回答或单次检索即可 2. agentic - 复杂查询需要多步推理、查询改写或工具使用 只返回 simple 或 agentic。 response self.llm.invoke(prompt) return response.content.strip().lower() def query(self, query: str) - dict: 执行查询 route self.route_query(query) if route simple: print(使用传统RAG处理...) return self.simple_rag.invoke({input: query}) else: print(使用Agentic RAG处理...) return self.agentic_rag.invoke({query: query})7.2 成本优化策略分层检索先使用轻量级模型进行初步筛选缓存机制缓存常见查询的结果智能路由简单查询使用传统RAG复杂查询才使用Agentic RAG批量处理合并多个查询进行批量处理7.3 评估与监控class RAGEvaluator: RAG系统评估器 def __init__(self, llm): self.llm llm def evaluate(self, query: str, answer: str, context: List[Document]) - dict: 评估RAG回答质量 # 1. 相关性评估 relevance self._evaluate_relevance(query, context) # 2. 忠实度评估是否有幻觉 faithfulness self._evaluate_faithfulness(answer, context) # 3. 完整性评估 completeness self._evaluate_completeness(query, answer) return { relevance: relevance, faithfulness: faithfulness, completeness: completeness, overall: (relevance faithfulness completeness) / 3 } def _evaluate_relevance(self, query: str, context: List[Document]) - float: 评估检索文档的相关性 # 实现相关性评估逻辑 pass def _evaluate_faithfulness(self, answer: str, context: List[Document]) - float: 评估答案的忠实度 detector HallucinationDetector(self.llm) has_hallucination, _ detector.check(answer, context) return 0.0 if has_hallucination else 1.0 def _evaluate_completeness(self, query: str, answer: str) - float: 评估答案的完整性 # 实现完整性评估逻辑 pass八、总结与展望8.1 核心要点回顾传统RAG线性管道架构适合简单查询成本低但能力有限Agentic RAG智能决策架构适合复杂查询能力强但成本高核心差异Agentic RAG引入了查询分析、自适应检索、迭代优化和自我验证LangGraph构建Agentic RAG的理想工具支持状态机和工作流编排8.2 技术演进趋势阶段特点代表技术RAG 1.0基础检索生成向量检索、简单promptRAG 2.0高级检索策略Hybrid Search、RerankingRAG 3.0Agentic RAGLangGraph、多智能体协作RAG 4.0自主智能体完全自主规划、执行、验证8.3 未来发展方向多模态RAG支持文本、图像、音频、视频的统一检索联邦RAG跨多个知识源的分布式检索个性化RAG基于用户画像的个性化检索策略实时RAG支持流式数据的实时检索与更新可解释RAG提供清晰的推理过程和决策依据8.4 学习资源LangGraph官方文档RAG Survey论文Agentic RAG最佳实践LangChain RAG教程结语从传统RAG到Agentic RAG的演进代表了AI系统从被动工具向主动智能体的转变。理解这两种架构的差异和适用场景将帮助你构建更强大、更可靠的AI应用。选择建议- 从传统RAG开始满足基本需求- 逐步引入Agentic特性解决复杂场景- 最终构建混合架构实现最佳性价比