Skip to content

第七章:知识库被 Agent 调用 —— 从 RAG 到 MCP ​

本章使用 MCP 作为受治理的工具接口边界,协议本身不替代鉴权、路由与责任归属。

本章在全书中的角色

读完本章你能做到:把建好的知识库暴露给 Claude Desktop、Cursor、Codex App 等任意 MCP Client 调用,设计多工具并存时的路由裁决机制。

前置条件:已完成第四章架构(有向量库/图谱/Skill 库)和第五章安全配置(权限已就绪)。

三类读者:工程师 → 直接从 MCP Server 代码段开始;架构师 → 重点看路由冲突与裁决机制;产品经理 → 重点看 MCP 是什么、与传统 API 的区别。

这是完整链路的最后一环:知识已经蒸馏、已经入库,Agent 如何高效调用它?


6.1 三种调用模式总览 ​

知识库被 Agent 调用有三种根本不同的模式,不可混淆:

text
模式1:RAG 检索(被动)
  Agent 查询 → 知识库返回相关块 → Agent 用于生成回答
  → 适合:声明性知识(是什么/为什么),频繁更新的参考资料

模式2:Skill 导航(主动)
  Agent 浏览 Skill 目录 → 找到匹配 Skill → 加载执行
  → 适合:程序性知识(怎么做),可复用的操作流程

模式3:Agentic RAG(自主迭代)
  Agent 自主决定何时检索、检索什么、是否足够、是否重试
  → 适合:复杂多跳问题,需要多步推理的任务

RAG 提供声明性证据(是什么),Skill 提供程序性指导(怎么做)。最佳实践是两者结合:

来自 Anything2Skill 论文实验数据:

  • 纯 RAG:95.41%(qsv)/ 76.50%(GitHub-CLI)
  • Anything2Skill + RAG:98.85%(qsv)/ 94.10%(GitHub-CLI)
  • 结论:Skill 和 RAG 互补,分别解决程序性和声明性两类需求

6.2 模式1:RAG 检索调用 SOP ​

基础向量 RAG 调用:

python
# LangChain 方式
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings()
)

# Agent 工具定义
from langchain.tools import Tool

knowledge_retrieval_tool = Tool(
    name="knowledge_search",
    description="""在知识库中搜索相关信息。
    输入:自然语言查询
    适用:当需要查询事实、参考资料、文档内容时""",
    func=lambda q: vectorstore.similarity_search(q, k=5)
)

GraphRAG 调用(LightRAG):

python
# 将 LightRAG 封装为 Agent 工具
from langchain.tools import BaseTool

class LightRAGTool(BaseTool):
    name = "graphrag_search"
    description = """知识图谱查询工具。
    适合:需要理解实体关系、跨文档综合的复杂查询。
    输入:自然语言查询
    模式:mix(默认,综合效果最好)"""
    
    def _run(self, query: str, mode: str = "mix") -> str:
        result = rag.query(query, param=QueryParam(mode=mode))
        return result
    
    async def _arun(self, query: str, mode: str = "mix") -> str:
        result = await rag.aquery(query, param=QueryParam(mode=mode))
        return result

查询路由(向量 RAG vs GraphRAG 混合):

python
# 简单路由规则(生产中用小模型分类器替代)
def route_query(query: str) -> str:
    multi_hop_signals = ["关系", "联系", "影响", "导致", "比较", "所有", "哪些", "趋势"]
    if any(s in query for s in multi_hop_signals):
        return "graphrag"
    return "vector_rag"

# 路由执行
def retrieve(query: str) -> str:
    if route_query(query) == "graphrag":
        return rag.query(query, param=QueryParam(mode="mix"))
    else:
        docs = vectorstore.similarity_search(query, k=5)
        return "\n\n".join([d.page_content for d in docs])

6.3 模式2:Skill 导航调用 SOP ​

Corpus2Skill 调用方式(导航式,无向量检索):

bash
# 编译阶段(一次性,离线)
python -m corpus2skill compile \
  --input ./knowledge_docs/ \
  --output ./skill_tree/ \
  --p 10 \
  --max-top 8 \
  --model claude-sonnet-4-6 \
  --embed-model Qwen/Qwen3-Embedding-0.6B

Agent 调用时(serve time):

python
from corpus2skill import serve

# Agent 有两个工具:
# 1. code_execution:浏览 SKILL.md 和 INDEX.md(导航)
# 2. get_document(doc_id):获取完整文档内容

# 典型查询流程(2-3 轮):
# 轮1:Agent 读 SKILL.md → 了解知识库全局结构(鸟瞰图)
# 轮2:Agent 读相关 INDEX.md → 缩小到具体文档列表
# 轮3:Agent 调用 get_document → 获取完整证据

Corpus2Skill 的核心优势:Agent 知道"还有多少没看",可以回溯,可以跨分支综合——这是向量检索永远无法提供的结构性可见性。

Agent Skill 原子性约束与降级协议(防执行崩溃):

[!] The 3-7-1 Rule:一个合格的 SKILL.md 必须满足:最多 3 层逻辑嵌套、最多 7 个主要执行步骤、单一的输入输出结构。

  • 降级拆解:若蒸馏出的长视频或复杂流程超过 7 步,必须拆分为一个 MASTER_SKILL.md 引用多个子 Skill。
  • 硬编码降级:若 Skill 中包含大量复杂的 IF-THEN 判断,不要用自然语言让 LLM 去猜,必须将其提取为 Python 脚本并封装为 MCP 工具,SKILL 仅负责调用该工具。

成本优化(官方数据):

  • 默认配置:每次查询 $0.172
  • 启用 prompt cache(默认开启):$0.089(节省 48%)
  • cache 命中率约 70%,命中部分仅需原价的 1/10

6.4 模式3:Agentic RAG 调用 SOP ​

Agentic RAG 的核心架构(AgenticRAG 论文,2026):

Agent 不再是 RAG 的被动消费者,而是主动控制检索过程:

流程图
图表接近视口时加载…
查看 Mermaid 源码
flowchart TD
    Start[Agentic Query] --> Reason{证据充足?}
    Reason -->|Yes| End[生成终局回答]
    Reason -->|No: 分析缺口| Act
    
    subgraph Action_Space [Act 工具箱]
        direction LR
        A1[search: 跨库搜索]
        A2[find: 文档内精确定位]
        A3[open: 读指定片段]
        A4[summarize: 降维释放Token]
    end
    
    Act --> Action_Space
    Action_Space --> Observe[Observe: 更新已知信息]
    Observe --> Check{Stop 条件触发?}
    
    Check -->|迭代达上限/超时| End
    Check -->|继续| Reason
    
    classDef default fill:#fafafa,stroke:#334155,stroke-width:1px;
    classDef decision fill:#fef08a,stroke:#ca8a04,stroke-width:1px;
    class Reason,Check decision;

五种 Agentic RAG 模式(按复杂度排序):

text
模式 1 — 迭代检索(Iterative Retrieval)
  流程:检索 → 阅读 → 评估缺口 → 精化查询 → 重复
  适用:单一查询,但第一次检索不够完整
  成本:低(1-3轮额外检索)

模式 2 — 查询分解(Query Decomposition)
  流程:复杂查询 → 拆分为 2-6 个子查询 → 并行检索 → 合并
  适用:"A 相比 B 在 C 方面表现如何,考虑到 D 条件" 类查询
  成本:中(N个子查询 × 检索成本)

模式 3 — 假设驱动检索(Hypothesis-Driven)
  流程:提出假设 → 专门检索证实/证伪 → 更新假设
  适用:"X 是否真的是 Y 的原因?"调查类查询
  成本:中(每假设 1-2 轮检索)

模式 4 — 跨源三角验证(Cross-Corpus Triangulation)
  流程:同一查询 → 多个知识源并行检索 → 融合/标注分歧
  适用:需要从内部文档 + 知识图谱 + 向量库多源印证
  成本:高(M个源 × 检索成本)

模式 5 — 证据加权合成(Evidence-Weighted Synthesis)
  流程:检索多份可能矛盾的证据 → 按来源可信度加权 → 合成
  适用:不同文档对同一问题有不同说法
  成本:中(额外一次推理用于合成)

A-RAG 实现(推荐起点):

python
# A-RAG:层次化检索接口,Agent 自主选择检索策略
# 三个工具:keyword_search / semantic_search / chunk_read

from arag import ARAG

agent = ARAG(
    corpus_dir="./knowledge_docs/",
    llm="claude-opus-4-7",  # 推理能力强的模型效果更好
    max_iterations=7,
    token_budget=40000
)

result = agent.query("哪些项目同时涉及风险A和风险B,且负责人都是同一团队?")
# Agent 自主选择:先 keyword_search 找关键实体,再 semantic_search 补充,
# 最后 chunk_read 读完整上下文——无需预设工作流

6.5 Skill 检索与路由(大规模 Skill 库) ​

当 Skill 数量超过 50 个时,需要 Skill 检索机制(否则全量加载超出上下文):

Anthropic Tool Search Tool 模式(官方推荐):

python
# 启动时只加载 search_tool,其余 Skill 按需检索
# 效果:Opus 4 准确率从 49% → 74%(Tool Search),token 减少 85%

# 系统提示配置:
system_prompt = """
你可以访问以下技能库。使用 search_skills 工具查找适合当前任务的技能,
然后使用 load_skill 工具加载并执行它。

可用工具:
- search_skills(query): 在技能库中搜索相关技能
- load_skill(skill_name): 加载指定技能
"""

# Agent 调用流程:
# 1. 接收任务
# 2. search_skills("需要做什么") → 找到候选 Skill
# 3. load_skill("匹配的Skill名") → 加载 SKILL.md
# 4. 按 Skill 指导执行

组合 Skill 路由(Gao 2026 论文,Compositional Skill Routing):

text
任务分解器(Decomposer)
    ↓ 分解为原子子任务
双编码器检索(Bi-encoder FAISS)
    ↓ 为每个子任务检索最匹配的 Skill
Skill-Aware 反馈循环(SAD)
    ↓ 将检索结果反馈给分解器重新调整(准确率 51% → 67.7%)
DAG 规划器(Dependency-Aware Planner)
    ↓ 组合为可执行计划

适用阈值:

  • < 50 Skills → 直接 prompt caching 全量加载,不需要检索
  • 50-200 Skills → 单步 retrieve-and-rerank(SkillRouter 方案)
  • 200+ Skills → 完整 Compositional Skill Routing

6.6 Agent 记忆层(跨会话知识持久化) ​

TencentDB Agent Memory 架构(Tencent,2026,集成 OpenClaw):

text
短期记忆(任务内):
  底层:完整工具输出(refs/*.md,只读存档)
  中层:步骤摘要(JSONL)
  顶层:Mermaid 任务状态图(极致压缩,Agent 只读顶层)
  → token 减少 61.38%,任务成功率提升 51.52%

长期记忆(跨会话):
  L0 Conversation → L1 Atom(原子事实)
           ↓
  L2 Scenario(场景块)→ L3 Persona(用户档案)
  → Agent 日常只查 L3 Persona,细节时下钻到 L1 Atom

记忆层写入时机:

text
任务完成时 → 写入长期记忆
  · 蒸馏本次任务中涌现的新知识 → wiki/
  · 更新用户偏好 → Persona
  · 记录本次使用了哪些 Skill、效果如何 → Skill 生命周期

任务失败时 → 写入经验教训
  · 记录失败原因和解决方案 → bdistill 规则提取
  · 更新 Skill 的 contraindications(禁忌)字段

6.7 完整端到端链路(蒸馏 → 入库 → Agent 调用) ​

text
┌─────────────────────────────────────────────────────────────────────┐
│                          原始多模态内容                               │
│                  (PDF/视频/音频/网页/代码/图像...)                    │
└────────────────────────────┬────────────────────────────────────────┘
                             │  第1-4部分(蒸馏)
                             ▼
┌─────────────────────────────────────────────────────────────────────┐
│                          内容解析层                                   │
│  MinerU / Docling / LlamaParse / faster-whisper / VLM Caption       │
│                  ↓ 统一 Markdown/JSON                                │
│                          知识蒸馏层                                   │
│  cangjie-skill / Resource2Skill / COLLEAGUE.SKILL / bdistill        │
│                          质量验证层                                   │
│  对抗一致性检测 + 置信度评分 + acceptance_predicate                  │
└────────────────────────────┬────────────────────────────────────────┘
                             │  分流入库
          ┌──────────────────┼──────────────────┐
          ▼                  ▼                  ▼
┌──────────────┐   ┌──────────────────┐  ┌───────────────┐
│  Skill 库    │   │  知识图谱         │  │  向量库        │
│  SKILL.md   │   │  (LightRAG/      │  │  (pgvector/   │
│ ~/.agents/  │   │   GraphRAG/      │  │   Qdrant/     │
│  skills/    │   │   Graphiti)      │  │   Milvus)     │
└──────┬───────┘   └────────┬─────────┘  └──────┬────────┘
       │  第6部分(Agent 调用)                   │
       └──────────────────┬──────────────────────┘
                          │
                    ┌─────▼──────┐
                    │  查询路由   │
                    │  分类器    │
                    └─────┬──────┘
              ┌────────────┼────────────┐
              ▼            ▼            ▼
      ┌──────────┐  ┌──────────┐  ┌──────────┐
      │简单事实  │  │多跳/综合  │  │程序性任务│
      │向量RAG  │  │GraphRAG  │  │Skill导航 │
      │精确/快速│  │LightRAG  │  │Corpus2S. │
      └────┬─────┘  └────┬─────┘  └────┬─────┘
           └─────────────┼─────────────┘
                         ▼
                 ┌──────────────┐
                 │ Agentic Loop │
                 │ (A-RAG 范式) │
                 │ ReAct 迭代   │
                 │ 工具调用     │
                 └──────┬───────┘
                        ▼
                   最终回答 + 来源引用
                        │
                   ┌────▼─────┐
                   │ 记忆回写  │
                   │ (长期记忆 │
                   │  更新)   │
                   └──────────┘

6.8 调用层工具总览 ​

工具/框架类型用途推荐场景
LightRAG Query APIGraphRAG 检索5种模式图谱查询多跳/综合查询
Corpus2Skill serveSkill 导航Agent 浏览层次技能树单域文档问答
Anything2Skill SkillBankSkill 检索RAG + Skill 双轨程序性知识调用
A-RAG(arag)Agentic 检索三层次工具自主检索复杂多步推理
AgenticRAG(search/find/open)Agentic 检索企业文档问答企业知识库
Anthropic Tool SearchSkill 路由大规模 Skill 库按需加载200+ Skill 场景
TencentDB Agent Memory记忆层短期+长期分层记忆长运行 Agent
Graphiti(Zep)时态记忆Agent 时态知识图谱知识随时间变化的场景
xMemoryAgent 记忆分层记忆(L0→L3)去冗余多会话连续推理

6.9 2026 新范式:MCP 协议(Model Context Protocol) ​

核心认知

MCP 不是"插件",是 2026 年 Agent 与知识库连接的标准协议。封装一次 MCP Server,Claude Desktop、Cursor、Codex App 都可以直接调用,无需写任何适配代码。

传统调用 vs MCP 调用 ​

维度传统 FastAPI 封装MCP Server 封装
每个 Agent 都需要写适配代码调用 REST API(OK) 无需,MCP 统一
新增知识库每个 Agent 改代码(OK) 只需注册新 Server
Claude Desktop 直接用(X) 不支持(OK) 原生支持
Cursor 直接用(X) 需要 Extension(OK) 原生支持
工程量每端各写一套一次封装,多端复用

MCP 协议三类能力 ​

text
Tools(工具)   — Agent 可以调用的函数,如 search_products()
Resources(资源)— Agent 可以读取的静态数据,如知识库 schema
Prompts(提示词)— 预置的对话模板,引导用户更好地描述需求

完整 MCP Server 示例 ​

参考 第 4 章 Stage 6 的 mcp_server.py 最小实现,以及 第 15 章 Prompt 05 的完整生产版封装(含 get_market_overview + 错误处理)。

LangGraph Agent 通过 MCP 调用知识库 ​

python
# langgraph_mcp_agent.py
# 展示 LangGraph Agent 如何通过 MCP 工具调用知识库
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
import asyncio

async def run_agent():
    # 加载 MCP Server 上的工具(自动发现 search_products 等)
    tools = await load_mcp_tools(server_url="http://localhost:8000/mcp")

    model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
    agent = create_react_agent(model, tools)

    result = await agent.ainvoke({
        "messages": [{"role": "user",
                      "content": "美国市场性价比最高的便携充电器有哪些?"}]
    })
    print(result["messages"][-1].content)

asyncio.run(run_agent())

知识库调用工具全景(2026 更新版) ​

工具/框架类型用途推荐场景
MCP Python SDKMCP 封装知识库→通用工具2026 首选,跨 Client 复用
LightRAG Query APIGraphRAG 检索5种模式图谱查询多跳/综合查询
Corpus2Skill serveSkill 导航Agent 浏览层次技能树单域文档问答
Anything2Skill SkillBankSkill 检索RAG + Skill 双轨程序性知识调用
A-RAGAgentic 检索三层次工具自主检索复杂多步推理
Graphiti(Zep)时态记忆Agent 时态知识图谱知识随时间变化

MCP 路由冲突与裁决机制 ​

当知识库被封装为多个 MCP Server,且多个 Server 提供功能重叠的工具时,路由器面临裁决问题。这不是协议问题,而是知识治理问题。

三类典型冲突场景 ​

冲突一:同名工具,不同数据源search_product_info 在 kb-server-public(公开竞品数据)和 kb-server-internal(内部产品数据)都存在。Agent 调用时选哪个?

冲突二:重叠覆盖,质量不同get_pricing_data 同时存在于"历史价格库"(周度更新)和"实时报价库"(分钟更新),两者对同一产品给出不同价格。

冲突三:权限边界交叉 Agent 有 Server A 的完整权限和 Server B 的只读权限,但完成当前任务需要 Server B 的最新数据。

裁决框架:PTDS 四维优先级 ​

在发生路由冲突时,按以下优先级顺序裁决:

优先级维度规则示例
P1精确性(Precision)更精确匹配任务意图的工具优先任务要"内部数据"则走内部Server
P2时效性(Timeliness)时效要求高时,新鲜度优先于深度实时价格 > 月度报告
P3数据级别(Data Level)默认优先使用最低敏感级别能满足需求的来源能用L1回答就不调L3
P4成本(Cost)同等质量下,低成本路由优先本地缓存 > API调用
python
class MCPRouter:
    def resolve_conflict(self, candidates: list[dict], task_context: dict) -> dict:
        """
        candidates: [{"server": "...", "tool": "...", "freshness": ..., 
                       "data_level": ..., "cost": ..., "precision_score": ...}]
        返回唯一胜出的工具调用方案
        """
        # P1: 精确性过滤
        best_precision = max(c["precision_score"] for c in candidates)
        candidates = [c for c in candidates 
                      if c["precision_score"] >= best_precision - 0.05]
        
        # P2: 若任务有时效要求,选最新鲜的
        if task_context.get("requires_realtime"):
            candidates = sorted(candidates, key=lambda c: c["freshness"])[:1]
        
        # P3: 最低敏感数据级别优先
        min_level = min(c["data_level"] for c in candidates)
        candidates = [c for c in candidates if c["data_level"] == min_level]
        
        # P4: 最低成本
        return min(candidates, key=lambda c: c["cost"])

冲突日志与审计 ​

所有路由冲突必须记录,否则无法发现系统性偏差:

python
def log_routing_conflict(candidates, winner, task_id):
    record = {
        "timestamp": datetime.now().isoformat(),
        "task_id": task_id,
        "candidates": len(candidates),
        "winner": winner["server"] + "." + winner["tool"],
        "reason": winner.get("selection_reason", "PTDS"),
        "losers": [c["server"] + "." + c["tool"] for c in candidates if c != winner]
    }
    # 每日汇总:哪些工具总是输?可能说明某个Server需要升级或下线
    append_to_audit_log("mcp_routing_conflicts.jsonl", record)

路由治理的核心原则

协议标准化(MCP)解决的是接口问题,不解决语义冲突和责任归属。每个 MCP Server 应该有明确的"服务边界声明":它承诺回答什么、不回答什么、精度保证范围是什么。没有边界声明的Server接入越多,路由器的不确定性越高。


→ 下一章

Agent 调用架构到位后,深入进阶选型——如何在 LoD 阶梯上选对蒸馏深度、何时用图谱 → 07-advanced-theory

来源与复核 ​

  • 复核状态:待复核。任何易漂移的版本、价格、法律或性能结论,采用前都必须回到一手来源再次确认。
  • 代码状态:示意代码。未被本地 smoke test 覆盖的片段不得解释为生产可运行。
  • 证据边界:本页成熟度只描述内容形态,不代表部署、上线或生产验收已经完成。
  • 下一验收动作:按仓库根目录 content-audit.md 中本模块的证据缺口补齐来源、fixture 与验收回执。