AI学习-从本地 RAG 到生产级 Agent 系统

fallen_fish

356人浏览 · 2026-05-28 00:21:43

fallen_fish · 2026-05-28 00:21:43 发布

AI学习-从本地 RAG 到生产级 Agent 系统

📚 目录

什么是 Agent
ReAct 框架：推理-行动循环
模型如何知道调用工具
Memory 系统
Tool Use 工程实践
Agent 质量的差异与优化
主流框架选型
多 Agent 协作系统
生产级 Agent 工程
学习路线与推荐资源

1. 什么是 Agent

**Agent（智能体）**是指能够自主感知环境、做出决策并采取行动来达成目标的 AI 系统。它与普通 LLM 调用的根本区别在于：

LLM：单次【问题→答案】映射
Agent：多步【感知→推理→行动】循环

1.1 RAG vs Agent：

前面已经完成了本地 RAG 系统，这是进入 Agent 世界最好的起点，因为 RAG 的【检索】本质上就是 Agent 最常用的第一个工具。

维度	工作方式	核心差异
RAG	固定流程：用户问题 → 向量检索 → 拼 Prompt → LLM 回答	被动，每次必须检索
Agent	动态决策：LLM 自主判断要不要检索、检索什么、检索几次	主动，按需调用工具

核心认知：Agent 不是新技术，是把 LLM 从「被动回答者」变成「主动执行者」的工程设计模式。

1.2 Agent 的四个核心组件

一个完整的 Agent 系统由四部分构成：

组件	职责	类比
规划（Planning）	决定下一步做什么、如何分解任务	大脑/思维
记忆（Memory）	存储对话历史、检索结果、中间状态	短期/长期记忆
工具（Tools）	搜索、代码执行、数据库查询、API 调用等	手和感官
行动（Action）	执行工具调用、输出结果、与外部世界交互	执行器

2. ReAct 框架：推理-行动循环

**ReAct（Reasoning + Acting）**是目前最主流的 Agent 框架，它让模型在每一步先写出推理过程，再执行工具调用，再观察结果，循环往复直到任务完成。

2.1 三步循环

Thought（思考）：模型输出内部推理，分析当前状态、决定下一步
Action（行动）：调用具体工具，执行操作
Observation（观察）：获得工具返回结果，塞回对话历史

完整的一轮示例：

Thought: 用户问入职两年有几天年假，我需要查公司政策
Action: search_hr_policy(query="年假天数")
Observation: 入职第一年5天，第二年起每年增加1天，最多15天
Thought: 已有足够信息，入职两年=第一年5天+第二年增加1天=6天
Answer: 您入职两年，共有 6 天年假。

2.2 为什么 Thought 步骤至关重要

不加 Thought 的 Agent 容易出现「随机调用工具」的问题——模型不经思考直接调用，导致参数错误、调用时机不对。Thought 步骤的作用：

强迫模型在行动前先推理，降低盲目调用概率
推理过程留在对话历史中，后续步骤可以参考
出错时便于调试，可以看到模型的推理链

工程注意：deepseek-r1 的 <think> 标签本质上是显式的 Thought 步骤。建议在 Agent 场景下保留思考过程，不要全部过滤掉。

2.3 循环终止条件

Agent 循环需要明确的终止条件，否则会无限运行：

模型输出不包含工具调用（msg.tool_calls 为空）
达到最大步数上限（建议 5~10 步）
工具返回特殊的完成信号
检测到循环：连续两步调用相同工具+相同参数

3. 模型如何知道调用工具

这是理解 Function Calling 机制最核心的问题，答案分两层。

3.1 训练层面（根本原因）

现代 LLM 在 RLHF / 指令微调阶段，接触了大量格式化的工具调用训练数据：

用户问题 + 工具列表 → 模型输出 JSON 调用 → 工具返回结果 → 自然语言回答

模型从这些数据里学到了一个行为模式：

如果问题需要外部信息 → 输出工具调用 JSON，而不是直接编造
如果问题可以直接回答 → 输出自然语言
这不是写死的 if-else，是从数据归纳出来的概率判断

3.2 推理层面（每次调用时）

API 收到请求后，将 tools 参数转化为系统 Prompt 注入给模型，大致内容为：

You have access to the following functions:

- search_hr_policy(query: str): 在HR知识库检索政策，当需要...时调用

If you want to call a function, respond ONLY with JSON:
{"name": "search_hr_policy", "arguments": {"query": "..."}}

模型在生成 token 时评估两条路径的概率：

直接回答路径：如果它已知足够信息
工具调用路径：如果需要外部信息且 description 与问题匹配

关键认知：tool_choice="auto" 不是框架在判断，是模型自己在判断。API 只做格式转换，判断权在模型。

3.3 Tool Description 决定调用准确率

description 是模型判断「该不该调这个工具」的唯一依据，写法差异对准确率影响显著：

质量	description 内容	效果
差的写法	“查询HR信息”	太模糊，随机调用
一般写法	“查询年假、报销等政策”	稍好，仍有歧义
好的写法	“当需要年假/报销/绩效政策时调用。不适用于闲聊、数学计算或政策已在上文出现的情况”	明确触发+排除条件

好的 description 模板：

[何时调用] 当用户询问...时调用本工具。
[何时不调用] 不适用于...（排除条件）。
[参数说明] query: 检索关键词，应简洁具体，例如"年假天数"而非"公司政策"。

4. Memory 系统

记忆是 Agent 与普通对话机器人最大的差异之一。设计良好的记忆系统决定了 Agent 能处理多复杂的任务。

4.1 四种记忆类型

类型	内容	特点	实现方式
工作记忆 (Working Memory)	当前对话的 messages 列表	容量有限，会话结束即消失	你代码里的 `messages` 变量
情景记忆 (Episodic Memory)	历史任务的结果、用户偏好记录	持久化存储，跨会话可用	存到数据库/文件
语义记忆 (Semantic Memory)	知识库、文档、事实	静态知识，通过 RAG 检索	你已有的向量库
程序记忆 (Procedural Memory)	常用工作流程、任务模板	固化为 Prompt 或函数	system prompt 里的规则

4.2 上下文窗口管理

工作记忆（messages 列表）是最容易出问题的地方。随着对话轮数增加，token 消耗激增，模型开始忘记早期信息。

常用压缩策略：

截断策略：工具返回超过 N 字就截断，只保留前 K 条最相关片段
摘要策略：每隔 N 轮用 LLM 对历史对话生成摘要，替换原始记录
滑动窗口：始终保留最近 N 条消息 + 初始 system prompt
分层存储：重要信息提取存入外部数据库，按需检索回来

实用代码片段：

def trim_history(history, max_tokens=4000):
    """超出限制时，保留 system + 最近若干轮"""
    system = [m for m in history if m["role"] == "system"]
    others = [m for m in history if m["role"] != "system"]
    # 从最新往前保留，直到接近 token 上限
    return system + others[-20:]  # 简单版：只保留最近20条

5. Tool Use 工程实践

5.1 工具的四个要素

每个工具必须清晰定义四个要素，缺一不可：

要素	作用	示例/格式	注意点
name	唯一标识符	`search_hr_policy`	简洁，动词+名词
description	调用时机说明	何时调/何时不调/适用场景	这是准确率的关键
parameters	输入参数的 schema	JSON Schema 格式，含 type/description	每个参数都要写 description
function	实际执行函数	Python 函数，返回字符串	错误时返回错误信息字符串

5.2 错误处理：最容易被忽视的细节

工具调用失败时，正确的做法是把错误信息返回给模型，让模型自行决定如何处理：

try:
    result = TOOL_REGISTRY[name](**args)
except Exception as e:
    result = f"工具执行失败：{str(e)}。请尝试换一个查询词，或告知用户无法获取该信息。"

# 把 result（包括错误信息）塞回 messages
messages.append({"role": "tool", "tool_call_id": ..., "content": result})

⚠️ 静默失败最危险：工具报错后不处理，让模型直接生成答案——模型会编造一个「看起来合理」的回答，用户完全无法发现错误。

5.3 工具返回结果的处理

不要把原始工具结果直接塞给模型，先做清洗：

截断过长内容（超过 500 字的 RAG 结果只取前几条最相关的）
去除噪声（HTML 标签、多余空白、重复内容）
结构化标注（“以下是检索结果：” 比直接粘贴更清晰）
标注置信度（如果检索得分很低，可以告诉模型"结果相关度较低"）

5.4 本地模型的 Function Calling 兼容性

不同模型对 Function Calling 的支持程度差异很大：

模型	FC 支持	备注
qwen2.5:7b	✅ 优秀	推荐首选，中文+工具调用都强
llama3.1:8b	✅ 良好	英文场景更稳定
deepseek-r1:7b	⚠️ 一般	有 think 标签干扰，参数解析偶尔失败
mistral:7b	❌ 较弱	需要手动 Prompt 引导才能稳定调用

6. Agent 质量的差异与优化

同样的基础模型，不同的工程设计可能带来质量上的巨大差距。差距主要体现在以下维度：

6.1 四个核心质量维度

维度	影响	优化方向
工具描述质量	影响工具调用的准确率和时机	description 加上负面示例
上下文管理	影响长任务的稳定性和记忆能力	截断/摘要/分层存储
错误恢复	影响任务完成率和鲁棒性	把错误塞回 messages 让模型决策
任务分解	影响复杂任务的处理质量	先让模型输出计划再逐步执行

6.2 任务分解：处理复杂任务的关键

对于需要多步骤的复杂任务，直接扔给 Agent 效果往往很差。正确做法是先让模型规划再执行：

在 system prompt 里加入规划指令：

system: |
  对于复杂问题（涉及多个政策/多个步骤），请先输出执行计划：
  1. 列出需要查询哪些信息
  2. 列出执行顺序
  再按计划逐步执行，每步完成后确认结果再继续。

人机协作：复杂任务建议加入「人工检查点」——Agent 完成规划后先展示给用户确认，再执行。这是生产系统中最常用的安全设计。

6.3 常见失败模式与应对

失败模式	表现	应对方法
无限循环	同一工具被反复调用	检测「连续相同调用」，自动跳出
幻觉工具调用	调用不存在的工具名	用 `TOOL_REGISTRY.get(name)` 做校验
参数错误	JSON 解析失败	用 try/except 包裹 json.loads，失败时让模型重试
上下文遗忘	长对话中忘记早期信息	摘要策略 + 关键信息单独存储
过度工具调用	不需要检索时也检索	description 里明确写排除条件
静默失败	工具报错模型自行编造	所有工具调用加 try/except，错误信息显式返回

7. 主流框架选型

选择框架不是越强大越好，而是与你的任务复杂度匹配。

7.1 框架对比

框架	核心设计	优势	适用场景
原生 Function Calling	完全自己写循环	极简，无依赖	学习原理最佳；维护成本高
LangGraph	状态机 + 节点图	灵活，可视化流程	复杂任务；生产推荐
LlamaIndex Workflows	事件驱动 DAG	与 RAG 集成好	RAG+Agent 融合场景
AutoGen	多 Agent 对话	多角色协作自然	需要多 Agent 讨论的任务
CrewAI	角色驱动多 Agent	上手快，配置简单	快速原型，有角色分工

7.2 你现在的代码应该如何演进

基于你已有的 RAG + 原生 Function Calling 代码，建议的演进路径：

当前阶段（已完成）：原生 Function Calling + 手写 Agent 循环
下一步：引入 LangGraph，把 Agent 循环改造成状态图，支持条件跳转和并行
再下一步：多 Agent 协作——至少两个专职 Agent（检索 Agent + 写作 Agent）
生产阶段：加入监控、日志、限流、人工检查点

8. 多 Agent 协作系统

当单个 Agent 的能力触及上限（上下文太长、任务太复杂、需要专业分工），就需要引入多 Agent 协作。

8.1 三种主要协作模式

主从模式（Orchestrator-Worker）

一个主控 Agent 负责规划和分配，多个工作 Agent 负责执行具体子任务。最常用，适合任务有明确分工的场景。

主控 Agent: 接受用户任务 → 分解 → 分配给专职 Worker
    HR Worker:  只负责查询公司政策
    Writer Worker: 只负责生成报告
主控 Agent: 汇总结果 → 返回用户

流水线模式（Pipeline）

多个 Agent 串联，每个 Agent 的输出是下一个的输入。适合有固定处理步骤的场景，如：搜集资料 → 分析 → 撰写报告。

对等协作模式（Peer Collaboration）

多个 Agent 平等对话，通过讨论达成共识。适合需要多角度验证的任务，如代码 Review、方案评估。

8.2 多 Agent 的核心挑战

通信开销：Agent 之间传递信息需要 token，多轮协作后成本急剧上升
一致性：多个 Agent 可能基于不同信息得出矛盾结论
调试难度：单 Agent 失败容易定位，多 Agent 失败很难追踪
死锁：Agent A 等 Agent B 的结果，Agent B 等 Agent A

建议：先把单 Agent 做稳，再引入多 Agent。过早引入多 Agent 会放大所有工程问题，而不是解决它们。

9. 生产级 Agent 工程

9.1 评估体系

没有评估体系的 Agent 是无法改进的。建立评估体系需要三个层次：

工具调用准确率：工具被正确调用的比例（调用时机对 + 参数正确）
任务完成率：Agent 成功完成用户意图的比例
用户满意度：最终答案的质量评分

最小可行的评估方案：

准备 20~50 个典型问题 + 期望答案（Golden Set）
跑 Agent，对比实际输出 vs 期望输出
记录每次工具调用日志，分析失败原因
每次优化后重新跑 Golden Set，看指标变化

9.2 可观测性：日志与追踪

生产 Agent 必须记录每一步的输入输出，否则出问题无从排查：

import logging, time

def traced_tool_call(name, args, fn):
    t0 = time.time()
    logging.info(f"[TOOL] {name} | args={args}")
    try:
        result = fn(**args)
        logging.info(f"[TOOL] {name} | ok | {time.time()-t0:.2f}s")
        return result
    except Exception as e:
        logging.error(f"[TOOL] {name} | error: {e}")
        return f"执行失败：{e}"

9.3 安全与控制

Agent 的自主性越高，安全风险越大。生产系统必须设置明确的护栏：

安全措施	说明
工具权限最小化	只暴露任务必要的工具，不给 Agent 文件系统写权限除非必须
操作确认机制	破坏性操作（删除、发送邮件、转账）必须人工确认
步数硬限制	设置 max_steps 上限（建议 10 步），超出自动终止并报警
输出过滤	Agent 输出在返回用户前，检查是否包含敏感信息
速率限制	限制单用户单位时间的 Agent 调用次数，防止滥用

9.4 MCP 协议：工具生态的标准化

**Model Context Protocol（MCP）**是 Anthropic 发布的开放协议，旨在标准化 LLM 与外部工具/数据源的连接方式。

MCP 的意义：之前每个 Agent 框架都有自己的工具接口格式，导致工具无法复用。MCP 提供统一接口规范，工具开发者写一次，所有支持 MCP 的 Agent 框架都能用。

本地工具（文件系统、Shell）：直接用 MCP 本地服务器
云服务工具（GitHub、Slack、数据库）：通过 MCP 远程服务器
Claude Desktop / Claude Code：已原生支持 MCP

10. 学习路线与推荐资源

10.1 四阶段学习路线

阶段	主题	核心内容	里程碑
阶段一 (2~3 周)	理解机制	ReAct 框架、Function Calling 原理、Memory 类型	能用原生 API 写出基础 Agent 循环
阶段二 (3~4 周)	动手实践	LangGraph / LlamaIndex、Agentic RAG、Tool Use 实战	能独立搭建研究助手 Agent
阶段三 (3~4 周)	多 Agent	AutoGen / CrewAI、协作模式、任务分解规划	能构建两个 Agent 协作完成复杂任务
阶段四 (持续)	工程化	评估体系、可观测性、安全、MCP 协议、部署	能把 Agent 系统上线并稳定运行

10.2 你的下一步行动

基于你已有的 RAG 代码，最高效的下一步是：

改造现有 RAG：把 search_hr_policy 包装成 Tool，加上 run_agent 循环（已完成）
加第二个工具：比如 get_employee_info(name) 查员工信息，体验多工具选择
引入 LangGraph：把 Agent 循环改写为状态图，理解状态管理的优势
做 Agentic RAG：让 Agent 自主决定「是否检索」「检索几次」「用哪个关键词」
构建评估集：准备 30 个问答对，建立自动评估管道，量化优化效果

10.3 推荐参考资料

论文

ReAct: Synergizing Reasoning and Acting in Language Models（2022）
Toolformer: Language Models Can Teach Themselves to Use Tools（2023）
AgentBench: Evaluating LLMs as Agents（2023）

文档

课程

DeepLearning.AI: AI Agents in LangGraph（推荐）
DeepLearning.AI: Multi AI Agent Systems with crewAI
Hugging Face Agents Course（免费，中文社区有译版）

🎯 最后一句话

Agent 工程 90% 的差距来自工程质量（工具描述、错误处理、上下文管理），而不是换更大的模型。先把基础做扎实，再谈复杂架构。

📝 附录：完整代码清单

主要依赖

pip install openai langchain-community chromadb sentence-transformers rank-bm25

启动本地 LLM（Ollama 示例）

ollama run qwen3.5:9b

运行测试

python agent_demo.py

核心代码结构速查

# ============================================================
# 从你的 RAG 升级到 Agent —— 最小改动版
# 核心思路：把 rag_answer 变成 Tool，让 LLM 自己决定调不调
# ============================================================

import json
import re
from openai import OpenAI

# ── 1. 你原有的知识库和检索器（保持不变）──────────────────────
documents = [
    "公司年假政策：入职第一年享有5天年假，第二年起每年增加1天，最多15天。",
    "报销流程：员工需在费用发生后30天内提交报销申请，超过30天不予报销。",
    "远程办公规定：每周最多可申请3天远程办公，需提前一天向直属上级报备。",
    "绩效考核：每年12月进行年度考核，分为A/B/C/D四个等级，A级员工获得20%奖金。",
    "培训补贴：员工参加与工作相关的培训课程，公司报销最高3000元/年的培训费用。",
]

from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.retrievers import EnsembleRetriever

embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese")
vectorstore = Chroma.from_texts(texts=documents, embedding=embeddings)

bm25_retriever = BM25Retriever.from_texts(documents)
bm25_retriever.k = 3
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)


# ── 2. 把检索逻辑封装成 Tool 函数 ──────────────────────────────
def search_hr_policy(query: str) -> str:
    """在公司HR知识库中检索相关政策"""
    docs = retriever.invoke(query)
    if not docs:
        return "未找到相关政策。"
    return "\n".join([f"- {d.page_content}" for d in docs])


def get_current_date() -> str:
    """获取今天的日期（用于计算入职年限等）"""
    from datetime import date
    return str(date.today())


# ── 3. 定义 Tool Schema（告诉 LLM 有哪些工具可用）──────────────
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_hr_policy",
            "description": "在公司HR知识库中检索政策文件，当需要回答关于年假、报销、绩效、培训等公司制度问题时调用",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "检索关键词，例如：'年假天数'、'报销时限'",
                    }
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_date",
            "description": "获取今天的日期，当问题涉及时间计算时使用",
            "parameters": {"type": "object", "properties": {}},
        },
    },
]

# Tool 名称 → 实际函数的映射
TOOL_REGISTRY = {
    "search_hr_policy": search_hr_policy,
    "get_current_date": get_current_date,
}


# ── 4. 连接本地 LLM（你原有的配置）───────────────────────────────
client = OpenAI(
    api_key="ollama",
    base_url="http://localhost:11500/v1"
)
MODEL = "qwen3.5:9b"


def strip_think(text: str) -> str:
    """去掉 deepseek-r1 的 <think> 标签"""
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()


# ── 5. Agent 循环：核心升级点 ──────────────────────────────────
def run_agent(user_question: str, max_steps: int = 5) -> str:
    """
    ReAct 风格的 Agent 主循环：
      1. LLM 看到问题，决定调用哪个 Tool
      2. 执行 Tool，把结果返回给 LLM
      3. LLM 再决定：继续调 Tool，还是直接回答
      4. 重复直到 LLM 给出最终答案（或达到步数上限）
    """
    messages = [
        {
            "role": "system",
            "content": (
                "你是公司的HR助手。回答员工问题时，"
                "必须先调用 search_hr_policy 检索相关政策，"
                "只根据检索到的内容回答，不要编造信息。"
            ),
        },
        {"role": "user", "content": user_question},
    ]

    print(f"\n{'='*55}")
    print(f"问题：{user_question}")

    for step in range(max_steps):
        # 请求 LLM（携带 tools 参数）
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",   # 让 LLM 自己决定用不用工具
        )
        msg = response.choices[0].message

        # ── 情况A：LLM 决定调用工具 ──────────────────────────
        if msg.tool_calls:
            # 把 LLM 的决策加入对话历史
            messages.append(msg)

            for tool_call in msg.tool_calls:
                name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)

                print(f"\n[步骤{step+1}] 🔧 调用工具：{name}({args})")

                # 执行实际函数
                result = TOOL_REGISTRY[name](**args)
                print(f"         📄 工具返回：{result[:100]}...")  # 截断显示

                # 把工具结果放回对话，LLM 下一轮能看到
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result,
                })

        # ── 情况B：LLM 直接给出最终回答 ──────────────────────
        else:
            answer = strip_think(msg.content or "")
            print(f"\n[最终回答] 💬 {answer}")
            print(f"{'='*55}")
            return answer

    return "超过最大步数，未能完成回答。"


# ── 6. 测试运行 ────────────────────────────────────────────────
if __name__ == "__main__":
    questions = [
        "我入职两年了，有几天年假？",
        "报销需要多少天内提交？",
        "绩效A级有什么奖励？",
        # 下面这个问题会让 Agent 组合调用两个工具
        "今天是几号？年底考核还有多久？",
    ]

    for q in questions:
        run_agent(q)