AgenticRAG：智能体驱动的检索增强生成

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools.retriever import create_retriever_tool
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain import hub

# 1. 构建向量数据库检索器
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 2. 将检索器包装为 Agent 工具
retriever_tool = create_retriever_tool(
    retriever,
    name="knowledge_base_search",
    description="搜索公司知识库。当需要查询产品信息、政策文档时使用此工具。"
)

# 3. 定义工具集（可以添加多个工具）
tools = [retriever_tool]

# 4. 使用 ReAct Prompt
prompt = hub.pull("hwchase17/react")

# 5. 创建 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(llm, tools, prompt)

# 6. 创建 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,         # 显示推理过程
    max_iterations=10,    # 最大迭代次数
    return_intermediate_steps=True  # 返回中间步骤
)

# 7. 运行 Agent
result = agent_executor.invoke({
    "input": "请比较我们产品 A 和产品 B 的核心差异，并给出选购建议"
})

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI

# 1. 构建两个独立的知识库索引
docs_a = SimpleDirectoryReader("./data/product_a").load_data()
docs_b = SimpleDirectoryReader("./data/product_b").load_data()

index_a = VectorStoreIndex.from_documents(docs_a)
index_b = VectorStoreIndex.from_documents(docs_b)

# 2. 将索引转换为查询引擎工具
tools = [
    QueryEngineTool(
        query_engine=index_a.as_query_engine(),
        metadata=ToolMetadata(
            name="product_a_docs",
            description="包含产品 A 的所有文档、规格和用户手册"
        ),
    ),
    QueryEngineTool(
        query_engine=index_b.as_query_engine(),
        metadata=ToolMetadata(
            name="product_b_docs",
            description="包含产品 B 的所有文档、规格和用户手册"
        ),
    ),
]

# 3. 创建 ReAct Agent
llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    verbose=True,
    max_iterations=15
)

# 4. 查询
response = agent.chat("比较产品 A 和产品 B 的电池续航和摄像头规格，哪个更适合摄影爱好者？")

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

# 1. 定义状态
class AgenticRAGState(TypedDict):
    question: str
    retrieved_docs: List[str]
    generation: str
    retry_count: int
    is_relevant: bool

# 2. 定义节点函数
def retrieve(state: AgenticRAGState):
    """检索相关文档"""
    docs = retriever.invoke(state["question"])
    return {"retrieved_docs": [doc.page_content for doc in docs]}

def evaluate_relevance(state: AgenticRAGState):
    """评估检索结果相关性"""
    # LLM 评估文档是否与问题相关
    evaluation = relevance_checker.invoke({
        "question": state["question"],
        "docs": state["retrieved_docs"]
    })
    return {"is_relevant": evaluation.is_relevant}

def rewrite_query(state: AgenticRAGState):
    """改写查询以获取更好的检索结果"""
    new_query = query_rewriter.invoke({"question": state["question"]})
    return {"question": new_query, "retry_count": state["retry_count"] + 1}

def generate(state: AgenticRAGState):
    """基于检索内容生成答案"""
    answer = rag_chain.invoke({
        "context": "\n".join(state["retrieved_docs"]),
        "question": state["question"]
    })
    return {"generation": answer}

# 3. 定义路由逻辑
def decide_to_regenerate(state: AgenticRAGState):
    if state["is_relevant"] or state["retry_count"] >= 3:
        return "generate"
    return "rewrite"

# 4. 构建状态图
workflow = StateGraph(AgenticRAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("evaluate", evaluate_relevance)
workflow.add_node("rewrite", rewrite_query)
workflow.add_node("generate", generate)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges("evaluate", decide_to_regenerate)
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)

app = workflow.compile()
result = app.invoke({"question": "...", "retry_count": 0})

# ❌ 模糊的工具描述
bad_tool = Tool(
    name="search",
    description="搜索信息"  # 太模糊，Agent 不知道何时用
)

# ✅ 精准的工具描述
good_tool = Tool(
    name="product_manual_search",
    description=(
        "搜索产品说明书和技术文档。"
        "当用户询问产品规格、操作步骤、故障排除时使用。"
        "不适用于价格查询或库存查询。"
    )
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=8,          # 复杂任务建议 5-10 次
    max_execution_time=60,     # 超时限制（秒）
    early_stopping_method="generate"  # 超限后强制生成答案
)

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

class RelevanceScore(BaseModel):
    is_relevant: bool = Field(description="文档是否与问题相关")
    score: float = Field(description="相关性评分 0-1")
    reason: str = Field(description="判断理由")

relevance_evaluator = ChatPromptTemplate.from_messages([
    ("system", "你是一个检索质量评估专家。评估给定文档是否能帮助回答问题。"),
    ("human", "问题：{question}\n\n文档内容：{document}\n\n请评估相关性：")
]) | llm.with_structured_output(RelevanceScore)