RAG 进阶：Agentic RAG 与多智能体协作（1）：从“问答机器人“到“AI 员工“

晨xi的光

260人浏览 · 2026-10-13 13:29:14

晨xi的光 · 2026-10-13 13:29:14 发布

RAG 的流水线搭完了：解析 → 切片 → 索引 → 检索 → 重排 → 生成 → 服务化。但这套系统有个本质局限——它是被动的。

用户问一句，它答一句。用户不问，它不动。它不会主动查天气、不会调 API、不会写数据库、不会在回答前先确认"你要的是 2024 年还是 2025 年的数据？"

Agent 要解决的就是这个问题：让 AI 从"回答者"变成"行动者"。

一、RAG vs Agentic RAG：本质区别在哪？

维度	传统 RAG	Agentic RAG
流程	线性：检索 → 生成	循环：感知 → 规划 → 行动 → 观察 → 再规划
工具	无，只查向量库	可调用搜索、API、数据库、代码解释器
决策	无，固定流水线	有，根据上下文决定下一步做什么
状态	无状态，每轮独立	有状态，记住历史动作和中间结果
适用场景	知识问答	复杂任务、多步推理、需要外部数据

一句话总结： 传统 RAG 是"查资料写答案"，Agentic RAG 是"接到任务后自己想办法完成"。

二、Agent 的核心组件

一个 Agent 由四个基本零件组成，缺一不可：

1. 大脑（LLM + Prompt）

负责理解任务、做决策、规划步骤。这里的关键不是模型多大，而是 Prompt 里是否清晰定义了"你可以做什么"和"你该怎么做" 。

2. 工具集（Tools）

Agent 的"手"。每个工具是一个函数，有明确的输入输出签名。常见工具：

search_knowledge_base(query) — 查知识库
web_search(query) — 联网搜索
query_database(sql) — 查数据库
send_email(to, subject, body) — 发邮件
calculator(expression) — 计算器

3. 记忆（Memory）

Agent 需要记住：

短期记忆：当前任务的历史对话和已执行的动作
长期记忆：跨会话的用户偏好、历史经验（可选）

4. 规划器（Planner）

决定"下一步做什么"。有两种主流模式：

ReAct：Reason + Act，交替进行推理和行动
Plan-and-Execute：先列完整计划，再逐步执行

三、ReAct 模式详解：最经典的 Agent 循环

ReAct 的核心思想是：先想清楚再动手，每动一步再想想。

一个典型的 ReAct 循环长这样：

Thought: 用户想知道上季度销售额，我需要先查数据库。
Action: query_database("SELECT SUM(amount) FROM sales WHERE quarter='Q3'")
Observation: 1250000
Thought: 数据拿到了，但用户可能还想知道环比增长。
Action: query_database("SELECT SUM(amount) FROM sales WHERE quarter='Q2'")
Observation: 1100000
Thought: 环比增长 = (1250000-1100000)/1100000 = 13.6%
Final Answer: 上季度销售额为 125 万元，环比增长 13.6%。

关键点：LLM 每次输出都包含 Thought（思考）、Action（行动）、Observation（观察结果）三个部分，循环直到产出 Final Answer。

四、动手：用 LangGraph 搭建第一个 Agent

LangGraph 是目前最流行的 Agent 框架，它把 Agent 建模成一个状态图，每个节点是一个动作，边是决策逻辑。

下面是一个最小可用的 ReAct Agent 实现：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
import operator

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息历史自动累加
    next_step: str  # 下一步动作

# 2. 定义工具
def search_knowledge_base(query: str) -> str:
    """搜索内部知识库"""
    # 这里替换成你实际的 RAG 检索逻辑
    return f"知识库中关于 '{query}' 的信息..."

def web_search(query: str) -> str:
    """联网搜索"""
    return f"网络搜索结果: {query}"

tools = {
    "search_knowledge_base": search_knowledge_base,
    "web_search": web_search,
}

# 3. 定义 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools([
    {"name": "search_knowledge_base", "description": "搜索内部知识库", "parameters": {"query": {"type": "string"}}},
    {"name": "web_search", "description": "联网搜索最新信息", "parameters": {"query": {"type": "string"}}},
])

# 4. 定义节点
def agent_node(state: AgentState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    results = []
    for tool_call in tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        result = tools[tool_name](**tool_args)
        results.append(ToolMessage(content=result, tool_call_id=tool_call["id"]))
    return {"messages": results}

# 5. 定义路由逻辑
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"  # 还有工具要调用，继续
    return END  # 没有工具调用了，结束

# 6. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")  # 工具执行完回到 agent

app = workflow.compile()

# 7. 运行
result = app.invoke({
    "messages": [HumanMessage(content="帮我查一下公司最新的报销政策，顺便搜一下同行是怎么做的")]
})
for msg in result["messages"]:
    print(msg.type, ":", msg.content[:100])

这段代码做了什么：