RAG + Agent 融合

为什么需要 RAG + Agent

传统 RAG 是固定管道：查询 → 检索 → 生成。这种方式对复杂问题效果不佳——它无法判断检索结果是否足够、无法在回答不满意时重新检索、也无法根据问题类型选择不同的检索策略。

将 RAG 融入 Agent 后，检索成为 Agent 可以主动调用的工具，Agent 可以决定何时检索、检索什么、是否需要多轮检索，以及如何综合多个知识源的结果。

基础：将向量检索封装为 Agent 工具

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
from langchain_core.documents import Document
from pydantic import BaseModel, Field
from typing import List, Optional

# ── 初始化向量数据库 ──────────────────────────────────────
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="company_knowledge",
    embedding_function=embeddings,
    persist_directory="./knowledge_db"
)
retriever = vectorstore.as_retriever(
    search_type="mmr",          # MMR：最大边际相关性，避免冗余
    search_kwargs={"k": 6, "fetch_k": 20}
)

# ── 封装为工具 ────────────────────────────────────────────
class KnowledgeSearchInput(BaseModel):
    query: str = Field(description="搜索问题，用自然语言描述需要查找的内容")
    filter_source: Optional[str] = Field(
        default=None,
        description="按来源过滤，如 'product_manual' 或 'faq'"
    )

@tool("knowledge_base_search", args_schema=KnowledgeSearchInput)
def search_knowledge_base(query: str, filter_source: str = None) -> str:
    """搜索公司内部知识库。适用于：产品文档、FAQ、政策规定、技术规格等。
    不适用于：实时数据、外部新闻、用户个人信息。"""
    search_kwargs = {"k": 5}
    if filter_source:
        search_kwargs["filter"] = {"source": filter_source}

    docs = retriever.invoke(query, config={"search_kwargs": search_kwargs})

    if not docs:
        return "知识库中未找到相关内容，建议使用网络搜索。"

    results = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源")
        results.append(f"[{i}] 来源：{source}\n{doc.page_content[:500]}")

    return "\n\n".join(results)

Self-RAG 模式：自我评估检索质量

Self-RAG 是 2023 年 Asai et al. 提出的框架，让 LLM 在检索和生成过程中发出"反思标记"，决定是否需要检索以及如何评估答案质量：

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class SelfRAGState(TypedDict):
    question: str
    documents: List[str]          # 检索到的文档
    generation: str               # 生成的答案
    retrieval_score: str          # "yes"/"no" 是否需要检索
    relevance_score: str          # "yes"/"no" 文档是否相关
    hallucination_score: str      # "yes"/"no" 是否有幻觉
    answer_score: str             # "yes"/"no" 答案是否满足问题

# ── 评估器 LLM ───────────────────────────────────────────
class RetrievalDecision(BaseModel):
    need_retrieval: bool
    reasoning: str

class RelevanceScore(BaseModel):
    is_relevant: bool
    reasoning: str

class HallucinationScore(BaseModel):
    has_hallucination: bool
    reasoning: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retrieval_grader = llm.with_structured_output(RetrievalDecision)
relevance_grader = llm.with_structured_output(RelevanceScore)
hallucination_grader = llm.with_structured_output(HallucinationScore)

# ── 节点函数 ──────────────────────────────────────────────
def decide_retrieval(state: SelfRAGState):
    """决定是否需要检索外部知识。"""
    decision = retrieval_grader.invoke([
        SystemMessage(content="判断回答这个问题是否需要检索外部知识库。"),
        HumanMessage(content=state["question"])
    ])
    return {"retrieval_score": "yes" if decision.need_retrieval else "no"}

def retrieve_docs(state: SelfRAGState):
    """执行检索。"""
    docs = retriever.invoke(state["question"])
    return {"documents": [d.page_content for d in docs]}

def grade_documents(state: SelfRAGState):
    """评估检索到的文档是否与问题相关。"""
    relevant_docs = []
    for doc in state["documents"]:
        score = relevance_grader.invoke([
            SystemMessage(content="评估文档是否与用户问题相关。"),
            HumanMessage(content=f"问题：{state['question']}\n\n文档：{doc}")
        ])
        if score.is_relevant:
            relevant_docs.append(doc)
    return {
        "documents": relevant_docs,
        "relevance_score": "yes" if relevant_docs else "no"
    }

def generate_answer(state: SelfRAGState):
    """生成答案。"""
    context = "\n\n".join(state.get("documents", []))
    response = llm.invoke([
        SystemMessage(content=f"基于以下上下文回答问题：\n\n{context}"),
        HumanMessage(content=state["question"])
    ])
    return {"generation": response.content}

def check_hallucination(state: SelfRAGState):
    """检查生成的答案是否有幻觉（内容超出文档范围）。"""
    score = hallucination_grader.invoke([
        SystemMessage(content="判断答案是否完全基于文档内容，未添加文档中不存在的信息。"),
        HumanMessage(content=
            f"文档：{state['documents']}\n\n答案：{state['generation']}"
        )
    ])
    return {
        "hallucination_score": "yes" if score.has_hallucination else "no"
    }

# ── 路由函数 ──────────────────────────────────────────────
def route_after_decision(state): return state["retrieval_score"]
def route_after_grading(state):  return state["relevance_score"]
def route_after_hallucination(state): return state["hallucination_score"]

# ── 构建 Self-RAG 图 ──────────────────────────────────────
sg = StateGraph(SelfRAGState)
sg.add_node("decide",     decide_retrieval)
sg.add_node("retrieve",   retrieve_docs)
sg.add_node("grade",      grade_documents)
sg.add_node("generate",   generate_answer)
sg.add_node("check_hall", check_hallucination)

sg.add_edge(START, "decide")
sg.add_conditional_edges("decide", route_after_decision,
                         {"yes": "retrieve", "no": "generate"})
sg.add_edge("retrieve", "grade")
sg.add_conditional_edges("grade", route_after_grading,
                         {"yes": "generate", "no": "retrieve"})  # 重新检索
sg.add_edge("generate", "check_hall")
sg.add_conditional_edges("check_hall", route_after_hallucination,
                         {"yes": "generate", "no": END})  # 重新生成

self_rag_graph = sg.compile()

Adaptive RAG：动态选择检索策略

Adaptive RAG 根据问题的类型和复杂度，动态选择最合适的检索路由策略：

from pydantic import BaseModel
from typing import Literal

class QuestionRoute(BaseModel):
    route: Literal["direct", "knowledge_base", "web_search"]
    reasoning: str

router_llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(QuestionRoute)

def adaptive_router(state):
    question = state["question"]
    decision = router_llm.invoke([
        SystemMessage(content="""根据问题类型选择最合适的处理策略：
        - direct：通用知识问题，不需要额外检索
        - knowledge_base：需要查询公司内部文档/产品手册/FAQ
        - web_search：需要实时信息/最新新闻/当前状态
        """),
        HumanMessage(content=question)
    ])
    return {"route": decision.route}

# 在图中根据路由分发到不同处理节点
# graph.add_conditional_edges("route_question", ...
# {"direct": "generate", "knowledge_base": "retrieve_kb", "web_search": "search"})