自适应 RAG 来了！用本地大模型打造更聪明的智能检索系统，看到就是赚到，建议收藏！！

前言大模型越来越强大但它们依旧有一个致命短板知识更新慢。如果直接问 ChatGPT 之类的模型一个近期事件的问题它很可能答不上来。这就是为什么RAG检索增强生成变得重要 —— 在回答问题之前先去找相关资料再让模型结合这些资料生成答案。不过RAG 并不是“一刀切”的方案有些问题根本不需要检索比如定义类问题有些问题需要一次检索就能解决而另一些则需要多次尝试比如先改写问题再检索。这就是自适应RAG的核心根据问题的不同动态选择最合适的策略。本文我们将用LangGraph 本地 LLMOllama Mistral搭建一个 Adaptive RAG 系统能在Web 搜索和向量库检索之间灵活切换还能自我纠错。注意我们的 Adaptive RAG 系统有两个主要分支**Web Search**处理最近事件相关的问题因为向量库的数据是历史快照不会包含最新信息。借助Tavily 搜索 API获取网页结果再交给 LLM 组织答案。Self-Corrective RAG针对我们自己构建的知识库这里我们抓取了 Lilian Weng 的几篇经典博客Agent、Prompt Engineering、Adversarial Attack。向量库用Chroma搭建文本向量用Nomic 本地 Embedding生成。如果第一次检索结果不相关会尝试改写问题再次检索。同时会过滤掉“答非所问”的文档避免垃圾结果。1. 环境准备%capture--no-stderr%pip install-U langchain-nomic langchain_community tiktoken langchainhub chromadb langchain langgraph tavily-python nomic[local]设置 API KeyTavily 搜索 Nomic embedding。importgetpass,osdef_set_env(var:str):ifnotos.environ.get(var):os.environ[var]getpass.getpass(f{var}: )_set_env(TAVILY_API_KEY)_set_env(NOMIC_API_KEY)2. 本地模型和向量库我们将要构建了一个向量数据库内容是 Lilian Weng 的三篇博客。以后凡是涉及Agent/Prompt Engineering/Adversarial Attack的问题就走这里。# Ollama 模型local_llm mistral# 文本切分 向量化from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import WebBaseLoaderfrom langchain_community.vectorstores import Chromafrom langchain_nomic.embeddings import NomicEmbeddingsurls[https://lilianweng.github.io/posts/2023-06-23-agent/,https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/,https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/,]docs[WebBaseLoader(url).load()forurlinurls]docs_list[itemforsublistindocsforiteminsublist]text_splitterRecursiveCharacterTextSplitter.from_tiktoken_encoder(chunk_size250,chunk_overlap0)doc_splitstext_splitter.split_documents(docs_list)vectorstoreChroma.from_documents(documentsdoc_splits,collection_namerag-chroma,embeddingNomicEmbeddings(modelnomic-embed-text-v1.5,inference_modelocal),)retrievervectorstore.as_retriever()3. 问题路由器Router假如这个问题和 Agent 相关所以走向量库。fromlangchain.promptsimportPromptTemplatefrom langchain_community.chat_modelsimportChatOllamafrom langchain_core.output_parsersimportJsonOutputParser llmChatOllama(modellocal_llm,formatjson,temperature0)promptPromptTemplate(templateYou are an expert at routing a user question to a vectorstore or web search... Question to route: {question},input_variables[question],)question_routerprompt|llm|JsonOutputParser()questionllm agent memoryprint(question_router.invoke({question:question}))执行结果{datasource:vectorstore}4. 检索质量评估Retrieval Grader如果检索到的文档与问题相关。retrieval_graderprompt|llm|JsonOutputParser()questionagent memorydocsretriever.get_relevant_documents(question)doc_txtdocs[1].page_contentprint(retrieval_grader.invoke({question:question,document:doc_txt}))执行结果{score:yes}5. 答案生成RAG Generate成了一段关于 “Agent Memory” 的解释。fromlangchainimporthubfrom langchain_core.output_parsersimportStrOutputParser prompthub.pull(rlm/rag-prompt)llmChatOllama(modellocal_llm,temperature0)rag_chainprompt|llm|StrOutputParser()questionagent memorygenerationrag_chain.invoke({context:docs,question:question})print(generation)执行结果In an LLM-powered autonomous agent system,the Large Language Model(LLM)functionsasthe agents brain...6. 幻觉检测Hallucination Grader如果答案确实是基于文档生成的没有瞎编。如果答案不靠谱就让系统重新检索或改写问题。hallucination_graderprompt|llm|JsonOutputParser()hallucination_grader.invoke({documents:docs,generation:generation})执行结果{score:yes}7. 答案有用性评估Answer Grader如果这个答案对用户有用。answer_grader.invoke({question:question,generation:generation})执行结果{score:yes}8. 问题重写器Question Rewriter改成了更适合检索的问法。question_rewriter.invoke({question:question})执行结果What is agent memory and how can it be effectively utilized in vector database retrieval?9. Web 搜索工具当问题和近期事件有关时就会走 Tavily 搜索而不是本地库。fromlangchain_community.tools.tavily_searchimportTavilySearchResultsweb_search_toolTavilySearchResults(k3)执行结果日志---ROUTE QUESTION---Whatisthe AlphaCodium paper about?{datasource:web_search}---ROUTE QUESTION TO WEB SEARCH------WEB SEARCH---Node web_search:------GENERATE------CHECK HALLUCINATIONS------DECISION:GENERATION IS GROUNDED IN DOCUMENTS------GRADE GENERATION vs QUESTION------DECISION:GENERATION ADDRESSES QUESTION---Node generate:---(The AlphaCodium paper introduces a new approach for code generation...)10.工作流LangGraph 具体实现我们用LangGraph把这些步骤连起来形成一个有条件分支的工作流开始→ 判断走 Web Search 还是 Vectorstore如果走 Vectorstore检索 → 文档过滤 →如果靠谱 → 返回结果如果不靠谱 → 改写问题 → 再检索如果没文档改写问题 → 再检索如果有文档生成答案 → 检查是否靠谱如果走 Web Search直接搜 → 生成答案 → 检查 → 返回结果最终系统能在不同类型的问题上灵活切换而不是死板地“一问一搜”。fromtypingimportListfrom typing_extensionsimportTypedDictclass GraphState(TypedDict): Represents the state of our graph. Attributes: question: question generation: LLM generation documents: list of documents question:strgeneration:strdocuments:List[str]### Nodesfrom langchain.schema import Documentdef retrieve(state): Retrieve documents Args: state (dict): The current graph state Returns: state (dict): New key added to state, documents, that contains retrieved documents print(---RETRIEVE---) question state[question] # Retrieval documents retriever.get_relevant_documents(question) return {documents: documents, question: question}def generate(state): Generate answer Args: state (dict): The current graph state Returns: state (dict): New key added to state, generation, that contains LLM generation print(---GENERATE---) question state[question] documents state[documents] # RAG generation generation rag_chain.invoke({context: documents, question: question}) return {documents: documents, question: question, generation: generation}def grade_documents(state): Determines whether the retrieved documents are relevant to the question. Args: state (dict): The current graph state Returns: state (dict): Updates documents key with only filtered relevant documents print(---CHECK DOCUMENT RELEVANCE TO QUESTION---) question state[question] documents state[documents] # Score each doc filtered_docs [] for d in documents: score retrieval_grader.invoke( {question: question, document: d.page_content} ) grade score[score] if grade yes: print(---GRADE: DOCUMENT RELEVANT---) filtered_docs.append(d) else: print(---GRADE: DOCUMENT NOT RELEVANT---) continue return {documents: filtered_docs, question: question}def transform_query(state): Transform the query to produce a better question. Args: state (dict): The current graph state Returns: state (dict): Updates question key with a re-phrased question print(---TRANSFORM QUERY---) question state[question] documents state[documents] # Re-write question better_question question_rewriter.invoke({question: question}) return {documents: documents, question: better_question}def web_search(state): Web search based on the re-phrased question. Args: state (dict): The current graph state Returns: state (dict): Updates documents key with appended web results print(---WEB SEARCH---) question state[question] # Web search docs web_search_tool.invoke({query: question}) web_results \n.join([d[content] for d in docs]) web_results Document(page_contentweb_results) return {documents: web_results, question: question}### Edges ###def route_question(state): Route question to web search or RAG. Args: state (dict): The current graph state Returns: str: Next node to call print(---ROUTE QUESTION---) question state[question] print(question) source question_router.invoke({question: question}) print(source) print(source[datasource]) if source[datasource] web_search: print(---ROUTE QUESTION TO WEB SEARCH---) return web_search elif source[datasource] vectorstore: print(---ROUTE QUESTION TO RAG---) return vectorstoredef decide_to_generate(state): Determines whether to generate an answer, or re-generate a question. Args: state (dict): The current graph state Returns: str: Binary decision for next node to call print(---ASSESS GRADED DOCUMENTS---) state[question] filtered_documents state[documents] if not filtered_documents: # All documents have been filtered check_relevance # We will re-generate a new query print( ---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY--- ) return transform_query else: # We have relevant documents, so generate answer print(---DECISION: GENERATE---) return generatedef grade_generation_v_documents_and_question(state): Determines whether the generation is grounded in the document and answers question. Args: state (dict): The current graph state Returns: str: Decision for next node to call print(---CHECK HALLUCINATIONS---) question state[question] documents state[documents] generation state[generation] score hallucination_grader.invoke( {documents: documents, generation: generation} ) grade score[score] # Check hallucination if grade yes: print(---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---) # Check question-answering print(---GRADE GENERATION vs QUESTION---) score answer_grader.invoke({question: question, generation: generation}) grade score[score] if grade yes: print(---DECISION: GENERATION ADDRESSES QUESTION---) return useful else: print(---DECISION: GENERATION DOES NOT ADDRESS QUESTION---) return not useful else: pprint(---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---)fromlanggraph.graphimportEND,StateGraph,STARTworkflowStateGraph(GraphState)# Define the nodesworkflow.add_node(web_search, web_search) # web searchworkflow.add_node(retrieve, retrieve) # retrieveworkflow.add_node(grade_documents, grade_documents) # grade documentsworkflow.add_node(generate, generate) # generateworkflow.add_node(transform_query, transform_query) # transform_query# Build graphworkflow.add_conditional_edges( START, route_question, { web_search: web_search, vectorstore: retrieve, },)workflow.add_edge(web_search, generate)workflow.add_edge(retrieve, grade_documents)workflow.add_conditional_edges( grade_documents, decide_to_generate, { transform_query: transform_query, generate: generate, },)workflow.add_edge(transform_query, retrieve)workflow.add_conditional_edges( generate, grade_generation_v_documents_and_question, { not supported: generate, useful: END, not useful: transform_query, },)# Compileapp workflow.compile()inputs{question:What is the AlphaCodium paper about?}foroutputinapp.stream(inputs):forkey,valueinoutput.items():pprint(fNode {key}:)pprint(\n---\n)pprint(value[generation])执行结果---ROUTE QUESTION---Whatisthe AlphaCodium paper about?{datasource:web_search}---ROUTE QUESTION TO WEB SEARCH------WEB SEARCH---Node web_search:------GENERATE------CHECK HALLUCINATIONS------DECISION:GENERATION IS GROUNDED IN DOCUMENTS------GRADE GENERATION vs QUESTION------DECISION:GENERATION ADDRESSES QUESTION---Node generate:---(The AlphaCodium paper introduces a new approach for code generation...)我们写的这套 自适应 RAG 系统展示了几个关键点灵活路由不同问题走不同管道Web / Vectorstore。自我纠错检索结果不相关时自动改写问题再试。质量把控通过“幻觉检测 答案有用性判断”尽量避免胡编乱造。本地化Embedding 和 LLM 都可以跑在本地隐私友好节省成本。未来可以扩展的方向包括增加“多步推理”路线先子问题分解再检索。更细的路由分类比如结构化查询 vs 自然语言查询。融合图数据库或知识图谱增强事实性。最后为什么要学AI大模型当下⼈⼯智能市场迎来了爆发期并逐渐进⼊以⼈⼯通⽤智能AGI为主导的新时代。企业纷纷官宣“ AI ”战略为新兴技术⼈才创造丰富的就业机会⼈才缺⼝将达 400 万DeepSeek问世以来生成式AI和大模型技术爆发式增长让很多岗位重新成了炙手可热的新星岗位薪资远超很多后端岗位在程序员中稳居前列。与此同时AI与各行各业深度融合飞速发展成为炙手可热的新风口企业非常需要了解AI、懂AI、会用AI的员工纷纷开出高薪招聘AI大模型相关岗位。最近很多程序员朋友都已经学习或者准备学习 AI 大模型后台也经常会有小伙伴咨询学习路线和学习资料我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频全系列的学习资料这些学习资料不仅深入浅出而且非常实用让大家系统而高效地掌握AI大模型的各个知识点。这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】AI大模型系统学习路线在面对AI大模型开发领域的复杂与深入精准学习显得尤为重要。一份系统的技术路线图不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点还能提供一条高效、有序的学习路径。但知道是一回事做又是另一回事初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性在这基础上找到高质量的学习资源不浪费时间、不走弯路又是重中之重。AI大模型入门到实战的视频教程项目包看视频学习是一种高效、直观、灵活且富有吸引力的学习方式可以更直观地展示过程能有效提升学习兴趣和理解力是现在获取知识的重要途径光学理论是没用的要学会跟着一起敲要动手实操才能将自己的所学运用到实际当中去这时候可以搞点实战案例来学习。海量AI大模型必读的经典书籍PDF阅读AI大模型经典书籍可以帮助读者提高技术水平开拓视野掌握核心技术提高解决问题的能力同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说阅读经典书籍是非常有必要的。600AI大模型报告实时更新这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示。AI大模型面试真题答案解析我们学习AI大模型必然是想找到高薪的工作下面这些面试题都是总结当前最新、最热、最高频的面试题并且每道题都有详细的答案面试前刷完这套面试题资料小小offer不在话下这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】