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

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])

这段代码做了什么：