不能只知道RAG、LangChain、Agent 的三角关系，还得知道LangGraph的四角关系。

其实就是：

最好做个涉及Agent、RAG、LangGraph、Langchain的RAG项目。

等你撸完了就知道是什么关系了。

之前在做RAG优化时，我被一个问题困扰了很久：为什么简单的"法国首都是哪里"和复杂的多跳推理问题，要走同样的处理流程？

这不是浪费计算资源吗？

直到我接触到Adaptive RAG的概念，并搞过很多大佬项目，才意识到原来RAG也可以"因材施教"。

折腾了两周后，我用LangGraph实现了一个自适应的RAG系统。

他们四角关系有点复杂：

说实话，踩了不少坑，但效果确实让人惊喜——简单查询的响应时间从850ms降到了120ms，复杂查询的准确率从72%提升到94%。

今天就把这个实现过程完整分享出来。

我先说说传统RAG的痛点。

研究数据显示，现实中的查询复杂度差异巨大：

• 简单查询：“Paris is the capital of what?” - LLM直接就能回答
• 复杂查询：“When did the people who captured Malakoff come to the region where Philipsburg is located?” - 需要4步推理才能搞定

可传统RAG呢？

管你简单复杂，一律走完整的检索-生成流程。

这就像用牛刀杀鸡，或者用小刀砍树，怎么都不合适。

在实践中，我发现主要有三种处理思路：

实际测试下来的数据对比：

指标	传统RAG	Adaptive RAG	提升率
简单查询延迟	850ms	120ms	85.9%
复杂查询准确率	72%	94%	30.6%
计算资源利用	100%	31.25%	68.75%

效率提升的计算其实很简单：Adaptive RAG时间 / 传统RAG时间 ≈ 31.25%

核心设计：让系统学会"判断"

我的设计思路是这样的：既然查询有难易之分，那就让系统先判断难度，再选择合适的处理策略。整个工作流如下：

经过几次重构，我最终采用了这样的模块化结构：

building-adaptive-rag/
├── src/
│   ├── workflow/           # 核心工作流逻辑
│   │   ├── chains/         # LLM处理链
│   │   │   ├── answer_grader.py
│   │   │   ├── generation.py
│   │   │   ├── hallucination_grader.py
│   │   │   ├── retrieval_grader.py
│   │   │   └── router.py
│   │   ├── nodes/          # 工作流节点
│   │   │   ├── generate.py
│   │   │   ├── grade_documents.py
│   │   │   ├── retrieve.py
│   │   │   └── web_search.py
│   │   ├── consts.py       # 节点常量
│   │   ├── graph.py        # 主工作流编排
│   │   └── state.py        # 状态管理
│   ├── models/             # 模型配置
│   └── cli/                # 命令行接口
├── data/                   # 数据处理
│   └── ingestion.py
└── tests/                  # 测试套件

说实话，一开始我把所有逻辑都写在一个文件里，调试起来简直是噩梦。

拆分成模块后，每个组件职责明确，维护起来轻松多了。

1. 状态管理 - 系统的"记忆"

首先要解决的是状态在各个节点间的传递问题。

我用TypedDict定义了一个中央状态：

from typing import List, TypedDict

class GraphState(TypedDict):
    """State object for workflow containing query, documents, and control flags."""
    question: str           # 用户的原始查询
    generation: str         # LLM生成的响应  
    web_search: bool       # Web搜索需求的控制标志
    documents: List[str]   # 检索到的文档上下文

这个设计看起来简单，但实践中发现TypedDict既保证了类型安全，又保持了足够的灵活性，是个不错的选择。

2. 查询路由器 - 系统的"大脑"

路由器是整个系统的第一道关卡，负责判断查询该走哪条路：

from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""
    datasource: Literal["vectorstore", "websearch"] = Field(
        ...,
        description="Route to web search or vectorstore.",
    )

# 这个提示词我调了好几版，现在这个效果最稳定
system = """You are an expert at routing a user question to a vectorstore or web search.
The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
Use the vectorstore for questions on these topics. For all else, use web-search."""

route_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)

question_router = route_prompt | llm.with_structured_output(RouteQuery)

用Pydantic的好处是能确保输出格式，避免解析错误。我之前用普通的字符串匹配，经常出现识别失败的情况。

3. 文档处理 - 知识的"仓库"

向量存储的构建是基础中的基础：

def create_vectorstore():
    """Create or load vector store for document retrieval."""
    urls = [
        "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/",
    ]
    
    # 250 token这个值是测试出来的最优值
    # 太大会丢失细节，太小又会破坏语义完整性
    text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
        chunk_size=250,     
        chunk_overlap=0     # 零重叠避免冗余
    )
    
    # 使用Chroma进行向量存储
    vectorstore = Chroma.from_documents(
        documents=doc_splits,
        embedding=GoogleGenerativeAIEmbeddings(model="models/text-embedding-004"),
        persist_directory="./chroma_langchain_db",
    )
    
    return vectorstore.as_retriever()

关于chunk_size，我也测试了100、250、500几个值，250在语义完整性和检索精度之间达到了最好的平衡。

4. 多层质量控制 - 系统的"免疫系统"

这部分是Adaptive RAG的精髓。可以设计了三层质量门控：

def grade_documents(state: GraphState) -> Dict[str, Any]:
    """
    Determines whether retrieved documents are relevant to the question
    If any document is not relevant, set flag to run web search
    """
    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]
    
    filtered_docs = []
    web_search = False
    
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        
        if grade.lower() == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            web_search = True  # 这里是关键：任何不相关都触发补充搜索
            
    return {
        "documents": filtered_docs, 
        "question": question, 
        "web_search": web_search
    }

我的策略比较保守：只要有一个文档不相关，就触发Web搜索。宁可多搜一次，也不能给用户不准确的答案。

质量保证的层级设计：

层级	检查内容	失败后动作
L1	文档相关性	触发Web搜索
L2	幻觉检测	重新生成
L3	答案完整性	补充搜索

5. 工作流编排 - 把所有组件串起来

这是整个系统最复杂的部分，需要处理各种条件分支：

def grade_generation_grounded_in_documents_and_question(state):
    """Grade answer quality and groundedness."""
    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]
    
    # 先检查是否有幻觉
    score = hallucination_grader.invoke({
        "documents": documents, 
        "generation": generation
    })
    
    if score.binary_score:
        # 再检查答案是否有用
        score = answer_grader.invoke({
            "question": question, 
            "generation": generation
        })
        return "useful" if score.binary_score else "not useful"
    else:
        return "not supported"

# 构建工作流 - 这里的顺序很重要
workflow = StateGraph(GraphState)

# 添加所有节点
workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, web_search)

# 设置条件入口点
workflow.set_conditional_entry_point(
    route_question,
    {WEBSEARCH: WEBSEARCH, RETRIEVE: RETRIEVE},
)

# 添加边缘关系
workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)

# 条件边缘：文档评分后的决策
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {WEBSEARCH: WEBSEARCH, GENERATE: GENERATE},
)

# 自校正循环 - 这是系统智能的关键
workflow.add_conditional_edges(
    GENERATE,
    grade_generation_grounded_in_documents_and_question,
    {
        "not supported": GENERATE,  # 有幻觉就重新生成
        "useful": END,              # 答案好就结束
        "not useful": WEBSEARCH     # 不够好就补充搜索
    },
)

workflow.add_edge(WEBSEARCH, GENERATE)

app = workflow.compile()

这个自校正机制我调试了很久。最开始没有"not useful"这个分支，导致一些答案虽然没有幻觉，但并没有真正回答用户的问题。

让我展示一个实际运行的例子，查询是：“What is agent memory?”

执行过程追踪：

---ROUTE QUESTION---     # 路由判断：走向量库
---RETRIEVE---           # 开始检索
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT RELEVANT---    # 文档1：相关
---GRADE: DOCUMENT RELEVANT---    # 文档2：相关  
---GRADE: DOCUMENT NOT RELEVANT--- # 文档3：不相关（触发器）
---GRADE: DOCUMENT RELEVANT---    # 文档4：相关
---ASSESS DOCUMENTS---
---WEB SEARCH---         # 自动补充搜索
---GENERATE---           # 融合生成
---CHECK HALLUCINATIONS--- # 质量检查通过

有意思的是，系统检索了4个文档，发现有1个不太相关，自动触发了Web搜索补充。最终答案是50%本地内容+50%Web搜索结果的组合。

这种自适应行为正是我想要的效果。

Web搜索扩展了系统的知识边界：

def web_search(state: GraphState) -> Dict[str, Any]:
    print("---WEB SEARCH---")
    question = state["question"]
    documents = state.get("documents", [])
    
    # Tavily API只取3个结果，多了反而影响质量
    tavily_results = web_search_tool.invoke({"query": question})["results"]
    
    # 结果合并
    joined_tavily_result = "\n".join(
        [tavily_result["content"] for tavily_result in tavily_results[:3]]
    )
    web_results = Document(page_content=joined_tavily_result)
    
    # 混合文档源
    if documents:
        documents.append(web_results)
    else:
        documents = [web_results]
    
    return {"documents": documents, "question": question}

为什么限制3个结果？我测试过1-10个不同数量，发现3个是信息覆盖和处理效率的最佳平衡点。

写了这么多代码，测试必不可少：

def test_retrival_grader_answer_yes() -> None:
    """测试相关文档的识别"""
    question = "What is retrieval augmented generation?"
    docs = retriever.invoke(question)
    doc_txt = docs[1].page_content
    
    res: GradeDocuments = retrieval_grader.invoke(
        {"question": question, "document": doc_txt}
    )
    assert res.binary_score == "yes"

def test_hallucination_grader_answer_no() -> None:
    """测试幻觉检测 - 这个case很有意思"""
    question = "What are the benefits of vector databases?"
    docs = retriever.invoke(question)
    
    # 故意生成一个完全不相关的答案
    res: GradeHallucinations = hallucination_grader.invoke({
        "documents": docs,
        "generation": "To bake a perfect chocolate cake, preheat to 350°"
    })
    assert not res.binary_score

测试不仅帮我发现了bug，也让我对系统的行为更有信心。

基于实践经验，我总结了几个优化方向：

优化项	实施方法	预期收益
缓存策略	LRU缓存频繁查询	减少40%重复计算
并发处理	异步文档评分	提升2.5x吞吐量
向量索引	HNSW算法优化	检索速度提升3x
批处理	批量嵌入生成	降低60%API调用

不过要注意，这些优化要根据实际场景来。比如缓存，如果你的查询重复率不高，反而会浪费内存。

搞完这个项目，我有几点深刻体会：

1. 自适应真的很重要。不同查询用不同策略处理，这个思路可以应用到很多场景。就像教学要因材施教，系统也要"因查询施策"。
2. 质量控制不能省。我最初版本没有幻觉检测，结果经常生成看似合理但实际错误的答案。多层质量门控虽然增加了复杂度，但确实提高了可靠性。
3. 模块化设计救了我。开始时代码都堆在一起，改个小功能要翻半天。拆分后，每个模块职责清晰，调试和扩展都方便多了。

时间复杂度分析也挺有意思：

• 简单查询：O(n)，n是文档数量
• 复杂查询：O(n·m)，m是推理步数
• 混合查询：O(n+k)，k是Web搜索结果数

我觉得闲了，会继续优化的地方：

1. 不能只知道RAG、LangChain、Agent 的三角关系，还得知道LangGraph的四角关系。

   其实就是：

    RAG、LangChain、Agent 到底有什么关系？ - 知乎_files/v2-080a63fc858a44c925d4231c4551f982_720w.webp)

   最好做个涉及Agent、RAG、LangGraph、Langchain的RAG项目。

   等你撸完了就知道是什么关系了。

   之前在做RAG优化时，我被一个问题困扰了很久：为什么简单的"法国首都是哪里"和复杂的多跳推理问题，要走同样的处理流程？

   这不是浪费计算资源吗？

   直到我接触到Adaptive RAG的概念，并搞过很多大佬项目，才意识到原来RAG也可以"因材施教"。

   折腾了两周后，我用LangGraph实现了一个自适应的RAG系统。

   他们四角关系有点复杂：

    RAG、LangChain、Agent 到底有什么关系？ - 知乎_files/v2-0f5ef71d3a51effc407f609d2bbb4063_720w.webp)

   说实话，踩了不少坑，但效果确实让人惊喜——简单查询的响应时间从850ms降到了120ms，复杂查询的准确率从72%提升到94%。

   今天就把这个实现过程完整分享出来。

   我先说说传统RAG的痛点。

   研究数据显示，现实中的查询复杂度差异巨大：

   • 简单查询：“Paris is the capital of what?” - LLM直接就能回答
   • 复杂查询：“When did the people who captured Malakoff come to the region where Philipsburg is located?” - 需要4步推理才能搞定

   可传统RAG呢？

   管你简单复杂，一律走完整的检索-生成流程。

   这就像用牛刀杀鸡，或者用小刀砍树，怎么都不合适。

   在实践中，我发现主要有三种处理思路：

    RAG、LangChain、Agent 到底有什么关系？ - 知乎_files/v2-28213203be6caa4a92fa648671c0f6f7_720w.webp)

   实际测试下来的数据对比：

   指标	传统RAG	Adaptive RAG	提升率
   简单查询延迟	850ms	120ms	85.9%
   复杂查询准确率	72%	94%	30.6%
   计算资源利用	100%	31.25%	68.75%

   效率提升的计算其实很简单：Adaptive RAG时间 / 传统RAG时间 ≈ 31.25%

   核心设计：让系统学会"判断"

   我的设计思路是这样的：既然查询有难易之分，那就让系统先判断难度，再选择合适的处理策略。整个工作流如下：

    RAG、LangChain、Agent 到底有什么关系？ - 知乎_files/v2-c3577d1ceb69e1306499d5056f9e69b2_720w.webp)

   经过几次重构，我最终采用了这样的模块化结构：

   building-adaptive-rag/
   ├── src/
   │   ├── workflow/           # 核心工作流逻辑
   │   │   ├── chains/         # LLM处理链
   │   │   │   ├── answer_grader.py
   │   │   │   ├── generation.py
   │   │   │   ├── hallucination_grader.py
   │   │   │   ├── retrieval_grader.py
   │   │   │   └── router.py
   │   │   ├── nodes/          # 工作流节点
   │   │   │   ├── generate.py
   │   │   │   ├── grade_documents.py
   │   │   │   ├── retrieve.py
   │   │   │   └── web_search.py
   │   │   ├── consts.py       # 节点常量
   │   │   ├── graph.py        # 主工作流编排
   │   │   └── state.py        # 状态管理
   │   ├── models/             # 模型配置
   │   └── cli/                # 命令行接口
   ├── data/                   # 数据处理
   │   └── ingestion.py
   └── tests/                  # 测试套件
   

   说实话，一开始我把所有逻辑都写在一个文件里，调试起来简直是噩梦。

   拆分成模块后，每个组件职责明确，维护起来轻松多了。

   1. 状态管理 - 系统的"记忆"

   首先要解决的是状态在各个节点间的传递问题。

   我用TypedDict定义了一个中央状态：

   from typing import List, TypedDict
   
   class GraphState(TypedDict):
       """State object for workflow containing query, documents, and control flags."""
       question: str           # 用户的原始查询
       generation: str         # LLM生成的响应  
       web_search: bool       # Web搜索需求的控制标志
       documents: List[str]   # 检索到的文档上下文
   

   这个设计看起来简单，但实践中发现TypedDict既保证了类型安全，又保持了足够的灵活性，是个不错的选择。

   2. 查询路由器 - 系统的"大脑"

   路由器是整个系统的第一道关卡，负责判断查询该走哪条路：

   from typing import Literal
   from langchain_core.prompts import ChatPromptTemplate
   from pydantic import BaseModel, Field
   
   class RouteQuery(BaseModel):
       """Route a user query to the most relevant datasource."""
       datasource: Literal["vectorstore", "websearch"] = Field(
           ...,
           description="Route to web search or vectorstore.",
       )
   
   # 这个提示词我调了好几版，现在这个效果最稳定
   system = """You are an expert at routing a user question to a vectorstore or web search.
   The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
   Use the vectorstore for questions on these topics. For all else, use web-search."""
   
   route_prompt = ChatPromptTemplate.from_messages(
       [
           ("system", system),
           ("human", "{question}"),
       ]
   )
   
   question_router = route_prompt | llm.with_structured_output(RouteQuery)
   

   用Pydantic的好处是能确保输出格式，避免解析错误。我之前用普通的字符串匹配，经常出现识别失败的情况。

   3. 文档处理 - 知识的"仓库"

   向量存储的构建是基础中的基础：

   def create_vectorstore():
       """Create or load vector store for document retrieval."""
       urls = [
           "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/",
       ]
       
       # 250 token这个值是测试出来的最优值
       # 太大会丢失细节，太小又会破坏语义完整性
       text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
           chunk_size=250,     
           chunk_overlap=0     # 零重叠避免冗余
       )
       
       # 使用Chroma进行向量存储
       vectorstore = Chroma.from_documents(
           documents=doc_splits,
           embedding=GoogleGenerativeAIEmbeddings(model="models/text-embedding-004"),
           persist_directory="./chroma_langchain_db",
       )
       
       return vectorstore.as_retriever()
   

   关于chunk_size，我也测试了100、250、500几个值，250在语义完整性和检索精度之间达到了最好的平衡。

   4. 多层质量控制 - 系统的"免疫系统"

   这部分是Adaptive RAG的精髓。可以设计了三层质量门控：

   def grade_documents(state: GraphState) -> Dict[str, Any]:
       """
       Determines whether retrieved documents are relevant to the question
       If any document is not relevant, set flag to run web search
       """
       print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
       question = state["question"]
       documents = state["documents"]
       
       filtered_docs = []
       web_search = False
       
       for d in documents:
           score = retrieval_grader.invoke(
               {"question": question, "document": d.page_content}
           )
           grade = score.binary_score
           
           if grade.lower() == "yes":
               print("---GRADE: DOCUMENT RELEVANT---")
               filtered_docs.append(d)
           else:
               print("---GRADE: DOCUMENT NOT RELEVANT---")
               web_search = True  # 这里是关键：任何不相关都触发补充搜索
               
       return {
           "documents": filtered_docs, 
           "question": question, 
           "web_search": web_search
       }
   

   我的策略比较保守：只要有一个文档不相关，就触发Web搜索。宁可多搜一次，也不能给用户不准确的答案。

   质量保证的层级设计：

   层级	检查内容	失败后动作
   L1	文档相关性	触发Web搜索
   L2	幻觉检测	重新生成
   L3	答案完整性	补充搜索

   5. 工作流编排 - 把所有组件串起来

   这是整个系统最复杂的部分，需要处理各种条件分支：

   def grade_generation_grounded_in_documents_and_question(state):
       """Grade answer quality and groundedness."""
       print("---CHECK HALLUCINATIONS---")
       question = state["question"]
       documents = state["documents"]
       generation = state["generation"]
       
       # 先检查是否有幻觉
       score = hallucination_grader.invoke({
           "documents": documents, 
           "generation": generation
       })
       
       if score.binary_score:
           # 再检查答案是否有用
           score = answer_grader.invoke({
               "question": question, 
               "generation": generation
           })
           return "useful" if score.binary_score else "not useful"
       else:
           return "not supported"
   
   # 构建工作流 - 这里的顺序很重要
   workflow = StateGraph(GraphState)
   
   # 添加所有节点
   workflow.add_node(RETRIEVE, retrieve)
   workflow.add_node(GRADE_DOCUMENTS, grade_documents)
   workflow.add_node(GENERATE, generate)
   workflow.add_node(WEBSEARCH, web_search)
   
   # 设置条件入口点
   workflow.set_conditional_entry_point(
       route_question,
       {WEBSEARCH: WEBSEARCH, RETRIEVE: RETRIEVE},
   )
   
   # 添加边缘关系
   workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
   
   # 条件边缘：文档评分后的决策
   workflow.add_conditional_edges(
       GRADE_DOCUMENTS,
       decide_to_generate,
       {WEBSEARCH: WEBSEARCH, GENERATE: GENERATE},
   )
   
   # 自校正循环 - 这是系统智能的关键
   workflow.add_conditional_edges(
       GENERATE,
       grade_generation_grounded_in_documents_and_question,
       {
           "not supported": GENERATE,  # 有幻觉就重新生成
           "useful": END,              # 答案好就结束
           "not useful": WEBSEARCH     # 不够好就补充搜索
       },
   )
   
   workflow.add_edge(WEBSEARCH, GENERATE)
   
   app = workflow.compile()
   

   这个自校正机制我调试了很久。最开始没有"not useful"这个分支，导致一些答案虽然没有幻觉，但并没有真正回答用户的问题。

   让我展示一个实际运行的例子，查询是：“What is agent memory?”

   执行过程追踪：

   ---ROUTE QUESTION---     # 路由判断：走向量库
   ---RETRIEVE---           # 开始检索
   ---CHECK DOCUMENT RELEVANCE TO QUESTION---
   ---GRADE: DOCUMENT RELEVANT---    # 文档1：相关
   ---GRADE: DOCUMENT RELEVANT---    # 文档2：相关  
   ---GRADE: DOCUMENT NOT RELEVANT--- # 文档3：不相关（触发器）
   ---GRADE: DOCUMENT RELEVANT---    # 文档4：相关
   ---ASSESS DOCUMENTS---
   ---WEB SEARCH---         # 自动补充搜索
   ---GENERATE---           # 融合生成
   ---CHECK HALLUCINATIONS--- # 质量检查通过
   

   有意思的是，系统检索了4个文档，发现有1个不太相关，自动触发了Web搜索补充。最终答案是50%本地内容+50%Web搜索结果的组合。

   这种自适应行为正是我想要的效果。

   Web搜索扩展了系统的知识边界：

   def web_search(state: GraphState) -> Dict[str, Any]:
       print("---WEB SEARCH---")
       question = state["question"]
       documents = state.get("documents", [])
       
       # Tavily API只取3个结果，多了反而影响质量
       tavily_results = web_search_tool.invoke({"query": question})["results"]
       
       # 结果合并
       joined_tavily_result = "\n".join(
           [tavily_result["content"] for tavily_result in tavily_results[:3]]
       )
       web_results = Document(page_content=joined_tavily_result)
       
       # 混合文档源
       if documents:
           documents.append(web_results)
       else:
           documents = [web_results]
       
       return {"documents": documents, "question": question}
   

   为什么限制3个结果？我测试过1-10个不同数量，发现3个是信息覆盖和处理效率的最佳平衡点。

   写了这么多代码，测试必不可少：

   def test_retrival_grader_answer_yes() -> None:
       """测试相关文档的识别"""
       question = "What is retrieval augmented generation?"
       docs = retriever.invoke(question)
       doc_txt = docs[1].page_content
       
       res: GradeDocuments = retrieval_grader.invoke(
           {"question": question, "document": doc_txt}
       )
       assert res.binary_score == "yes"
   
   def test_hallucination_grader_answer_no() -> None:
       """测试幻觉检测 - 这个case很有意思"""
       question = "What are the benefits of vector databases?"
       docs = retriever.invoke(question)
       
       # 故意生成一个完全不相关的答案
       res: GradeHallucinations = hallucination_grader.invoke({
           "documents": docs,
           "generation": "To bake a perfect chocolate cake, preheat to 350°"
       })
       assert not res.binary_score
   

   测试不仅帮我发现了bug，也让我对系统的行为更有信心。

   基于实践经验，我总结了几个优化方向：

   优化项	实施方法	预期收益
   缓存策略	LRU缓存频繁查询	减少40%重复计算
   并发处理	异步文档评分	提升2.5x吞吐量
   向量索引	HNSW算法优化	检索速度提升3x
   批处理	批量嵌入生成	降低60%API调用

   不过要注意，这些优化要根据实际场景来。比如缓存，如果你的查询重复率不高，反而会浪费内存。

   搞完这个项目，我有几点深刻体会：

   1. 自适应真的很重要。不同查询用不同策略处理，这个思路可以应用到很多场景。就像教学要因材施教，系统也要"因查询施策"。
   2. 质量控制不能省。我最初版本没有幻觉检测，结果经常生成看似合理但实际错误的答案。多层质量门控虽然增加了复杂度，但确实提高了可靠性。
   3. 模块化设计救了我。开始时代码都堆在一起，改个小功能要翻半天。拆分后，每个模块职责清晰，调试和扩展都方便多了。

   时间复杂度分析也挺有意思：

   • 简单查询：O(n)，n是文档数量
   • 复杂查询：O(n·m)，m是推理步数
   • 混合查询：O(n+k)，k是Web搜索结果数

   我觉得闲了，会继续优化的地方：

   1. 增加LLM Fallback：有些查询既不需要检索也不需要搜索，直接用LLM回答就行，比如"你好"这种。
   2. 个性化路由策略：不同用户可能有不同偏好，可以根据历史交互学习最优路由策略。
   3. 更智能的chunk策略：现在是固定250 token，理想情况应该根据内容类型动态调整。

   Adaptive RAG让我看到了RAG系统的新可能。

   它不再是一个死板的管道，而是一个会"思考"、会"判断"的智能系统。

   系统能够智能地选择策略、主动补充信息、自我纠正错误。

   这时你就清楚四角关系，其实不过如此⋯⋯脱离项目的理解，没有什么意义。不如为擼边理解。

2. 增加LLM Fallback：有些查询既不需要检索也不需要搜索，直接用LLM回答就行，比如"你好"这种。

3. 个性化路由策略：不同用户可能有不同偏好，可以根据历史交互学习最优路由策略。

4. 更智能的chunk策略：现在是固定250 token，理想情况应该根据内容类型动态调整。

Adaptive RAG让我看到了RAG系统的新可能。

它不再是一个死板的管道，而是一个会"思考"、会"判断"的智能系统。

系统能够智能地选择策略、主动补充信息、自我纠正错误。

理但实际错误的答案。多层质量门控虽然增加了复杂度，但确实提高了可靠性。
3. 模块化设计救了我。开始时代码都堆在一起，改个小功能要翻半天。拆分后，每个模块职责清晰，调试和扩展都方便多了。

时间复杂度分析也挺有意思：

• 简单查询：O(n)，n是文档数量
• 复杂查询：O(n·m)，m是推理步数
• 混合查询：O(n+k)，k是Web搜索结果数

我觉得闲了，会继续优化的地方：

1. 增加LLM Fallback：有些查询既不需要检索也不需要搜索，直接用LLM回答就行，比如"你好"这种。
2. 个性化路由策略：不同用户可能有不同偏好，可以根据历史交互学习最优路由策略。
3. 更智能的chunk策略：现在是固定250 token，理想情况应该根据内容类型动态调整。

Adaptive RAG让我看到了RAG系统的新可能。

它不再是一个死板的管道，而是一个会"思考"、会"判断"的智能系统。

系统能够智能地选择策略、主动补充信息、自我纠正错误。

这时你就清楚四角关系，其实不过如此⋯⋯脱离项目的理解，没有什么意义。不如为擼边理解。

2. 增加LLM Fallback：有些查询既不需要检索也不需要搜索，直接用LLM回答就行，比如"你好"这种。

3. 个性化路由策略：不同用户可能有不同偏好，可以根据历史交互学习最优路由策略。

4. 更智能的chunk策略：现在是固定250 token，理想情况应该根据内容类型动态调整。

Adaptive RAG让我看到了RAG系统的新可能。

它不再是一个死板的管道，而是一个会"思考"、会"判断"的智能系统。

系统能够智能地选择策略、主动补充信息、自我纠正错误。

这时你就清楚四角关系，其实不过如此⋯⋯脱离项目的理解，没有什么意义。不如为擼边理解。

可能大家都想学习AI大模型技术，也_想通过这项技能真正达到升职加薪，就业或是副业的目的，但是不知道该如何开始学习_，因为网上的资料太多太杂乱了，如果不能系统的学习就相当于是白学。
为了帮助大家打破壁垒，快速了解大模型核心技术原理，学习相关大模型技术。从原理出发真正入局大模型。在这里我和MoPaaS魔泊云联合梳理打造了系统大模型学习脉络，这份 LLM大模型资料 分享出来：包括LLM大模型书籍、640套大模型行业报告、LLM大模型学习视频、LLM大模型学习路线、开源大模型学习教程等, 😝有需要的小伙伴，可以 扫描下方二维码免费领取🆓**⬇️⬇️⬇️

【大模型全套视频教程】

教程从当下的市场现状和趋势出发，分析各个岗位人才需求，带你充分了解自身情况，get 到适合自己的 AI 大模型入门学习路线。

从基础的 prompt 工程入手，逐步深入到 Agents，其中更是详细介绍了 LLM 最重要的编程框架 LangChain。最后把微调与预训练进行了对比介绍与分析。

同时课程详细介绍了AI大模型技能图谱知识树，规划属于你自己的大模型学习路线，并且专门提前收集了大家对大模型常见的疑问，集中解答所有疑惑！

深耕 AI 领域技术专家带你快速入门大模型

跟着行业技术专家免费学习的机会非常难得，相信跟着学习下来能够对大模型有更加深刻的认知和理解，也能真正利用起大模型，从而“弯道超车”，实现职业跃迁！

【精选AI大模型权威PDF书籍/教程】

精心筛选的经典与前沿并重的电子书和教程合集，包含《深度学习》等一百多本书籍和讲义精要等材料。绝对是深入理解理论、夯实基础的不二之选。

【AI 大模型面试题 】

除了 AI 入门课程，我还给大家准备了非常全面的**「AI 大模型面试题」，**包括字节、腾讯等一线大厂的 AI 岗面经分享、LLMs、Transformer、RAG 面试真题等，帮你在面试大模型工作中更快一步。

【大厂 AI 岗位面经分享（92份）】

【AI 大模型面试真题（102 道）】

【LLMs 面试真题（97 道）】

【640套 AI 大模型行业研究报告】

【AI大模型完整版学习路线图（2025版）】

明确学习方向，2025年 AI 要学什么，这一张图就够了！

👇👇点击下方卡片链接免费领取全部内容👇👇

抓住AI浪潮，重塑职业未来！

科技行业正处于深刻变革之中。英特尔等巨头近期进行结构性调整，缩减部分传统岗位，同时AI相关技术岗位（尤其是大模型方向）需求激增，已成为不争的事实。具备相关技能的人才在就业市场上正变得炙手可热。

行业趋势洞察：

• 转型加速： 传统IT岗位面临转型压力，拥抱AI技术成为关键。
• 人才争夺战： 拥有3-5年经验、扎实AI技术功底和真实项目经验的工程师，在头部大厂及明星AI企业中的薪资竞争力显著提升（部分核心岗位可达较高水平）。
• 门槛提高： “具备AI项目实操经验”正迅速成为简历筛选的重要标准，预计未来1-2年将成为普遍门槛。

与其观望，不如行动！

面对变革，主动学习、提升技能才是应对之道。掌握AI大模型核心原理、主流应用技术与项目实战经验，是抓住时代机遇、实现职业跃迁的关键一步。

01 为什么分享这份学习资料？

当前，我国在AI大模型领域的高质量人才供给仍显不足，行业亟需更多有志于此的专业力量加入。

因此，我们决定将这份精心整理的AI大模型学习资料，无偿分享给每一位真心渴望进入这个领域、愿意投入学习的伙伴！

我们希望能为你的学习之路提供一份助力。如果在学习过程中遇到技术问题，也欢迎交流探讨，我们乐于分享所知。

*02 这份资料的价值在哪里？*

专业背书，系统构建：

• 本资料由我与MoPaaS魔泊云的鲁为民博士共同整理。鲁博士拥有清华大学学士和美国加州理工学院博士学位，在人工智能领域造诣深厚：

• • 在IEEE Transactions等顶级学术期刊及国际会议发表论文超过50篇。
  • 拥有多项中美发明专利。
  • 荣获吴文俊人工智能科学技术奖（中国人工智能领域重要奖项）。

• 目前，我有幸与鲁博士共同进行人工智能相关研究。

内容实用，循序渐进：

• 资料体系化覆盖了从基础概念入门到核心技术进阶的知识点。

• 包含丰富的视频教程与实战项目案例，强调动手实践能力。

• 无论你是初探AI领域的新手，还是已有一定技术基础希望深入大模型的学习者，这份资料都能为你提供系统性的学习路径和宝贵的实践参考，助力你提升技术能力，向大模型相关岗位转型发展。

  

抓住机遇，开启你的AI学习之旅！