AI代理RAG折腾手记

代理RAG AgenticRAG 智能体
作者: 醉月思 发布日期: 2025-04-23 分类: 技术实践

边界条件才会告诉你方案能不能留。

很多人一上来就讲AI代理RAG的全景图；我更想先把这次卡住的点说清楚。

背景：当传统RAG遇到天花板

最近在做企业知识库项目，老板要求做到"像人一样思考"。但实际用下来发现，传统RAG系统就像个死板的搜索引擎——你问什么，它就按固定套路检索什么，完全没有"理解力"和"判断力"。

比如用户问"最近三个月的Q3季度财务报表增长情况"，传统RAG只会机械检索包含"财务报表"“增长"的文档，然后硬拼个答案出来。但实际场景里，这个问题需要：

  1. 先理解"最近三个月"是什么时间范围
  2. 找到Q3相关的财务数据
  3. 计算（或提取）增长指标
  4. 可能需要对比历史数据

这些步骤传统RAG根本处理不了。老板说"这个系统不够智能”，我知道得升级了。

需求：我要的是会思考的RAG

看了两天论文和技术方案，发现这就是"Agentic RAG"（代理RAG）要解决的问题。核心思路很简单：让RAG不仅能检索，还能像人一样做决策和规划。

具体需求：

  • 动态检索策略：根据问题复杂度，选择不同的检索方式（向量搜索、关键词搜索、混合检索等）
  • 多步推理：复杂问题需要分解成多个子问题，依次解决
  • 工具调用：能调用外部工具（比如数据库、API、计算器等）
  • 自我反思：检索结果不满意时，能重新调整检索策略
  • 上下文管理：长对话中能记住关键信息，不重复检索

实现：一步步搭建代理RAG系统

架构设计

先画个整体架构图，免得后面把自己绕晕：

graph TD A[用户问题] --> B[意图识别] B --> C{问题复杂度?} C -->|简单| D[单次检索] C -->|复杂| E[任务分解] E --> F[多步检索链] F --> G[工具调用] D --> H[结果整合] F --> H G --> H H --> I[答案生成] I --> J{满意度评估} J -->|满意| K[返回答案] J -->|不满意| L[调整策略] L --> C

技术选型

折腾了一圈，最终选用了这些技术：

  • 核心框架：LangGraph（做agent编排比LangChain稳定）
  • 向量数据库：Chroma（轻量级，本地部署方便）
  • 大模型：GPT-4o（推理能力强，但成本高）
  • 工具库：LangChain Tools（内置各种工具）

避坑提示：一开始试过LangChain的Agent，但状态管理经常出问题。后来换成LangGraph才稳定下来，建议有类似需求的可以直接上LangGraph。

核心代码实现

1. 基础RAG组件

先搭个基础RAG系统：

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader

# 文档加载
loader = DirectoryLoader('./data/knowledge_base', glob="**/*.md")
documents = loader.load()

# 文档分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
splits = text_splitter.split_documents(documents)

# 向量化存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

2. Agent状态定义

用LangGraph需要先定义状态结构：

from typing import TypedDict, Annotated, List, Union
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[List[Union[HumanMessage, AIMessage, SystemMessage]], operator.add]
    retrieved_docs: List[str]
    current_step: int
    max_steps: int
    tool_calls: List[str]
    final_answer: str

3. 意图识别节点

判断问题复杂度，决定用哪种检索策略：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

intent_prompt = ChatPromptTemplate.from_template("""
你是一个意图识别专家。分析用户问题的复杂度：

用户问题：{question}

请判断：
1. 问题类型（简单检索/复杂推理/需要工具）
2. 是否需要多步处理
3. 建议的检索策略

返回JSON格式：
{{
    "type": "simple|complex|tool_required",
    "need_steps": true/false,
    "strategy": "vector_search|hybrid_search|multi_step"
}}
""")

def intent_recognition(state: AgentState):
    last_message = state["messages"][-1]
    response = llm.invoke(intent_prompt.format(question=last_message.content))
    intent = json.loads(response.content)

    state["current_step"] = 0
    state["max_steps"] = 5 if intent["need_steps"] else 1

    return {
        **state,
        "strategy": intent["strategy"]
    }

4. 智能检索节点

根据策略选择不同的检索方式：

def smart_retrieval(state: AgentState):
    question = state["messages"][-1].content
    strategy = state.get("strategy", "vector_search")

    if strategy == "vector_search":
        docs = retriever.invoke(question)
    elif strategy == "hybrid_search":
        # 混合检索：向量+关键词
        docs = retriever.invoke(question)
        # 这里可以加入BM25等其他检索方式
    elif strategy == "multi_step":
        # 多步检索
        docs = []
        sub_questions = decompose_question(question)
        for sub_q in sub_questions:
            docs.extend(retriever.invoke(sub_q))
    else:
        docs = retriever.invoke(question)

    # 排重和排序
    unique_docs = list({doc.page_content: doc for doc in docs}.values())

    return {
        **state,
        "retrieved_docs": [doc.page_content for doc in unique_docs[:5]]
    }

5. 工具调用节点

支持调用外部工具：

from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> str:
    """简单的计算器工具，用于数学计算"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "计算错误"

@tool
def search_web(query: str) -> str:
    """网络搜索工具，用于查找最新信息"""
    search = DuckDuckGoSearchRun()
    return search.run(query)

tools = [calculator, search_web]

def tool_execution(state: AgentState):
    last_message = state["messages"][-1]
    tool_calls = []

    # 这里简化处理，实际需要根据LLM的tool_calls来执行
    for tool in tools:
        if tool.name in last_message.content:
            # 模拟工具调用
            result = tool.invoke({"expression": last_message.content.split(":")[-1]})
            tool_calls.append(f"{tool.name}: {result}")

    return {
        **state,
        "tool_calls": tool_calls
    }

6. 答案生成节点

整合检索结果和工具调用结果，生成最终答案：

from langchain.chains import RetrievalQA

answer_prompt = ChatPromptTemplate.from_template("""
你是一个专业助手。基于以下信息回答用户问题：

用户问题：{question}

检索到的文档：
{retrieved_docs}

工具调用结果：
{tool_results}

请提供准确、详细的回答。如果信息不足，请诚实说明。
""")

def answer_generation(state: AgentState):
    question = state["messages"][-1].content
    retrieved_docs = "\n".join(state["retrieved_docs"])
    tool_results = "\n".join(state.get("tool_calls", []))

    response = llm.invoke(answer_prompt.format(
        question=question,
        retrieved_docs=retrieved_docs,
        tool_results=tool_results
    ))

    return {
        **state,
        "final_answer": response.content,
        "messages": state["messages"] + [response]
    }

7. 满意度评估节点

检查答案是否满意，不满意则调整策略重新检索：

def satisfaction_check(state: AgentState):
    answer = state["final_answer"]

    evaluate_prompt = ChatPromptTemplate.from_template("""
评估以下回答的质量：

问题：{question}
回答：{answer}

评估标准：
1. 是否直接回答了问题
2. 信息是否准确完整
3. 逻辑是否清晰

返回评分（1-5分）和简要说明。
""")

    last_message = state["messages"][-2]  # 用户的问题
    evaluation = llm.invoke(evaluate_prompt.format(
        question=last_message.content,
        answer=answer
    ))

    score = int(evaluation.content.split("评分：")[1].split("分")[0])

    if score < 3 and state["current_step"] < state["max_steps"]:
        # 不满意，重新检索
        state["current_step"] += 1
        return {
            **state,
            "strategy": "hybrid_search"  # 换个检索策略
        }
    else:
        # 满意或达到最大步数，结束
        return {
            **state,
            "status": "completed"
        }

8. 构建Graph

把所有节点连起来：

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("intent_recognition", intent_recognition)
workflow.add_node("smart_retrieval", smart_retrieval)
workflow.add_node("tool_execution", tool_execution)
workflow.add_node("answer_generation", answer_generation)
workflow.add_node("satisfaction_check", satisfaction_check)

# 设置入口
workflow.set_entry_point("intent_recognition")

# 添加边
workflow.add_edge("intent_recognition", "smart_retrieval")
workflow.add_edge("smart_retrieval", "tool_execution")
workflow.add_edge("tool_execution", "answer_generation")
workflow.add_edge("answer_generation", "satisfaction_check")

# 条件边：根据满意度决定是否重新检索
workflow.add_conditional_edges(
    "satisfaction_check",
    lambda x: "smart_retrieval" if x.get("status") != "completed" else END,
    {
        "smart_retrieval": "smart_retrieval",
        END: END
    }
)

# 编译图
app = workflow.compile()

使用示例

# 简单问题
result = app.invoke({
    "messages": [HumanMessage(content="公司的年假政策是什么？")],
    "retrieved_docs": [],
    "current_step": 0,
    "max_steps": 1,
    "tool_calls": [],
    "final_answer": ""
})

# 复杂问题
result = app.invoke({
    "messages": [HumanMessage(content="比较Q1和Q2的销售额增长情况，并预测Q3趋势")],
    "retrieved_docs": [],
    "current_step": 0,
    "max_steps": 5,
    "tool_calls": [],
    "final_answer": ""
})

print(result["final_answer"])

踩坑：那些让我通宵的坑

坑1：状态管理混乱

问题：LangChain的Agent状态经常丢失，导致对话不连贯。

解决方案：改用LangGraph的StateGraph，明确每个节点的状态传递。另外，给状态加版本号，方便调试：

state = {
    "version": "v1.2",
    "messages": [...],
    "retrieved_docs": [...],
    # ...
}

坑2：检索策略切换失败

问题：有时候切换到混合检索后，结果反而更差。

解决方案：给每种检索策略设置置信度阈值，只有低于阈值才切换：

def smart_retrieval(state: AgentState):
    # ... 检索逻辑 ...

    confidence = calculate_confidence(docs, question)
    if confidence < 0.7 and state["current_step"] > 0:
        # 切换策略
        state["strategy"] = "hybrid_search"
        # 重新检索...

    return state

坑3：成本爆炸

问题：GPT-4o调用太多，账单吓人。

解决方案：

  1. 简单问题用GPT-4o-mini，复杂问题才用GPT-4o
  2. 设置最大调用次数，避免无限循环
  3. 缓存常见问题的答案
from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_answer(question_hash: str) -> str:
    # 从缓存获取答案
    pass

# 在意图识别前先查缓存
question_hash = hashlib.md5(question.encode()).hexdigest()
cached = get_cached_answer(question_hash)
if cached:
    return cached

坑4：幻觉问题

问题：有时候检索结果不足，LLM还是会"编造"答案。

解决方案：

  1. 强制要求引用来源
  2. 设定"不知道就不知道"的prompt
  3. 限制只能基于检索结果回答
answer_prompt = ChatPromptTemplate.from_template("""
你只能基于以下文档回答问题。如果文档中没有相关信息，请回答"我不知道"。

文档：{docs}

问题：{question}

回答时必须引用来源。
""")

结果：终于达到了老板的要求

上线两个月，效果出乎意料：

性能指标：

  • 准确率从65%提升到82%
  • 复杂问题解决率从40%提升到75%
  • 用户满意度从3.2分提升到4.1分（满分5分）

典型案例： 用户问"公司去年研发投入占比多少，今年有什么变化？"，系统能够：

  1. 识别需要对比两年数据
  2. 分别检索2024和2025的研发报告
  3. 提取并计算占比
  4. 总结变化趋势

成本控制：

  • 通过分级模型策略，成本控制在传统方案的1.3倍
  • 答案缓存减少了30%的重复调用

结语

折腾代理RAG这两个月，最大的感受是：AI系统要像人一样思考，首先要理解人是怎么思考的。

传统RAG就像个只会做选择题的学生，而代理RAG更像会做综合题的高手——它知道什么时候该查资料，什么时候该用计算器，什么时候该换个思路。

当然，这套方案还有优化空间：

  • 更好的意图识别（降低误判率）
  • 更智能的检索策略（根据问题类型动态调整）
  • 更强的工具生态（支持更多外部系统）

如果你也在做类似的系统，希望这些经验能帮到你。踩过的坑不白踩，下次就不用再通宵了。

P.S. 老板现在对系统很满意，已经打算推广到其他部门了。不过新需求又来了——要支持多轮对话的上下文记忆。这又是另一个故事了…

版权声明: 本文首发于 指尖魔法屋-AI代理RAG折腾手记（https://blog.thinkmoon.cn/post/374-ai-agentic-rag-retrieval-decision-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

Harness Engineering 的介绍与编程实践
这篇文章从 AI 编程的真实工程场景出发，梳理 Harness Engineering 如何通过上下文、工具、沙盒、验证和人工闸门，让 Agent 从“会写代码”走向“能稳定交付”。
📁 技术实践
📅 2026-04-23
AI长度惩罚折腾手记
文档生成项目里有个反差：用户问「Python 快速排序怎么实现」，模型回一行「平均复杂度 O(n log n)」就收工；换到 API 文档场景，…
📁 技术实践
📅 2025-12-31