自适应RAG：用本地 LLM 构建更聪明的检索增强生成系统

摘要：大模型越来越强大，但它们依旧有一个致命短板：知识更新慢。如果直接问 ChatGPT 之类的模型一个近期事件的问题，它很可能答不上来。这就是为什么RAG（检索增强生成）变得重要 —— 在回答问题之前，先去找相关资料，再让模型结合这些资料生成答案。

大模型越来越强大，但它们依旧有一个致命短板：知识更新慢。如果直接问 ChatGPT 之类的模型一个近期事件的问题，它很可能答不上来。这就是为什么 RAG（检索增强生成） 变得重要 —— 在回答问题之前，先去找相关资料，再让模型结合这些资料生成答案。

不过，RAG 并不是“一刀切”的方案：有些问题根本不需要检索（比如定义类问题），有些问题需要一次检索就能解决，而另一些则需要多次尝试（比如先改写问题，再检索）。这就是 自适应RAG 的核心：根据问题的不同，动态选择最合适的策略。

本文我们将用 LangGraph + 本地 LLM（Ollama + Mistral） 搭建一个 Adaptive RAG 系统，能在 Web 搜索 和 向量库检索 之间灵活切换，还能自我纠错。

注意：我们的 Adaptive RAG 系统有两个主要分支：

Web Search：处理最近事件相关的问题（因为向量库的数据是历史快照，不会包含最新信息）。借助 Tavily 搜索 API 获取网页结果，再交给 LLM 组织答案。

Self-Corrective RAG：针对我们自己构建的知识库（这里我们抓取了 Lilian Weng 的几篇经典博客：Agent、Prompt Engineering、Adversarial Attack）。向量库用 Chroma 搭建，文本向量用 Nomic 本地 Embedding 生成。如果第一次检索结果不相关，会尝试改写问题，再次检索。同时会过滤掉“答非所问”的文档，避免垃圾结果。

�pture --no-stderr%pip install -U langchain-nomic langchain_community tiktoken langchainhub chromadb langchain langgraph tavily-python nomic[local]import getpass, osdef _set_env(var: str): if not os.environ.get(var): os.environ[var] = getpass.getpass(f"{var}: ")_set_env("TAVILY_API_KEY")_set_env("NOMIC_API_KEY")

我们将要构建了一个 向量数据库，内容是 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 for url in urls]docs_list = [item for sublist in docs for item in sublist]text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size=250, chunk_overlap=0)doc_splits = text_splitter.split_documents(docs_list)vectorstore = Chroma.from_documents( documents=doc_splits, collection_name="rag-chroma", embedding=NomicEmbeddings(model="nomic-embed-text-v1.5", inference_mode="local"),)retriever = vectorstore.as_retriever

假如这个问题和 Agent 相关，所以走向量库。

from langchain.prompts import PromptTemplatefrom langchain_community.chat_models import ChatOllamafrom langchain_core.output_parsers import JsonOutputParserllm = ChatOllama(model=local_llm, format="json", temperature=0)prompt = PromptTemplate( template="""You are an expert at routing a user question to a vectorstore or web search... Question to route: {question}""", input_variables=["question"],)question_router = prompt | llm | JsonOutputParserquestion = "llm agent memory"print(question_router.invoke({"question": question}))

执行结果

如果检索到的文档与问题相关。

retrieval_grader = prompt | llm | JsonOutputParserquestion = "agent memory"docs = retriever.get_relevant_documents(question)doc_txt = docs[1].page_contentprint(retrieval_grader.invoke({"question": question, "document": doc_txt}))

执行结果

成了一段关于 “Agent Memory” 的解释。

from langchain import hubfrom langchain_core.output_parsers import StrOutputParserprompt = hub.pull("rlm/rag-prompt")llm = ChatOllama(model=local_llm, temperature=0)rag_chain = prompt | llm | StrOutputParserquestion = "agent memory"generation = rag_chain.invoke({"context": docs, "question": question})print(generation)

执行结果

In an LLM-powered autonomous agent system, the Large Language Model (LLM) functions as the agent's brain...

如果答案确实是基于文档生成的，没有瞎编。如果答案不靠谱，就让系统重新检索或改写问题。

hallucination_grader = prompt | llm | JsonOutputParserhallucination_grader.invoke({"documents": docs, "generation": generation})

执行结果

如果这个答案对用户有用。

answer_grader.invoke({"question": question, "generation": generation})

执行结果

改成了更适合检索的问法。

question_rewriter.invoke({"question": question})

执行结果

'What is agent memory and how can it be effectively utilized in vector database retrieval?'

当问题和近期事件有关时，就会走 Tavily 搜索，而不是本地库。

from langchain_community.tools.tavily_search import TavilySearchResultsweb_search_tool = TavilySearchResults(k=3)

执行结果日志

---ROUTE QUESTION---What is the 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...')

我们用 LangGraph 把这些步骤连起来，形成一个有条件分支的工作流：

开始 → 判断走 Web Search 还是 Vectorstore如果走 Vectorstore：检索 → 文档过滤 →如果靠谱 → 返回结果如果不靠谱 → 改写问题 → 再检索如果没文档：改写问题 → 再检索如果有文档：生成答案 → 检查是否靠谱如果走 Web Search：直接搜 → 生成答案 → 检查 → 返回结果

最终，系统能在不同类型的问题上灵活切换，而不是死板地“一问一搜”。

from typing import Listfrom typing_extensions import TypedDictclass GraphState(TypedDict): """ Represents the state of our graph. Attributes: question: question generation: LLM generation documents: list of documents """ question: str generation: str documents: 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_content=web_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 "vectorstore"def 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 "generate"def 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---")from langgraph.graph import END, StateGraph, STARTworkflow = StateGraph(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.compileinputs = {"question": "What is the AlphaCodium paper about?"}for output in app.stream(inputs): for key, value in output.items: pprint(f"Node '{key}':") pprint("\n---\n")pprint(value["generation"])

执行结果

我们写的这套自适应 RAG 系统展示了几个关键点：

灵活路由：不同问题走不同管道（Web / Vectorstore）。

自我纠错：检索结果不相关时，自动改写问题再试。

质量把控：通过“幻觉检测 + 答案有用性判断”，尽量避免胡编乱造。

本地化：Embedding 和 LLM 都可以跑在本地（隐私友好，节省成本）。

未来可以扩展的方向包括：增加“多步推理”路线（先子问题分解，再检索）。更细的路由分类（比如结构化查询 vs 自然语言查询）。融合图数据库或知识图谱，增强事实性。

更多免费AI功能云片AI：https://y-p.cc

&云片