接入外部世界——工具、MCP 与知识检索

Agent 不能只在模型内部打转。要解决真实问题,它必须能够观察、检索、操作外部系统。这就是工具使用、MCP 和 RAG 的共同命题。

工具使用:把能力变成受控接口

Agent 要行动,就必须通过工具接触外部世界。工具可以是搜索、数据库查询、文件读写、浏览器控制、代码执行、消息发送、部署操作,也可以是业务系统 API。

工具设计的关键不是”模型会不会调用”,而是工具是否提供了合适的边界。

工具设计的五个关键属性

一个好工具应该具备:

明确的参数结构

参数应该有清晰的类型和约束。字符串参数需要说明格式(路径、URL、查询表达式),数字参数需要范围,枚举参数需要列出可选值。参数名称要与业务语言对齐,避免模型猜测。

清晰的成功和失败返回

工具返回应该区分成功和失败,失败时给出可操作的错误信息。不要返回 null 或空字符串让模型猜测。返回结构要稳定,避免在调用链中传递隐式状态。

可审计的调用日志

每次工具调用应该记录:调用时间、参数、调用者、返回值摘要、执行耗时。这些日志是调试、安全审计、成本分析的基础。

必要的权限和确认机制

高风险操作需要权限检查和人工确认。权限检查应该前置,在参数验证阶段就完成,避免执行到一半才发现无权限。确认机制要设计得当,既不能让每个操作都打断流程,也不能让危险操作静默执行。

足够窄的职责

一个工具应该只做一件事。bash 这样的通用工具只能作为探索手段,高频、高风险、高价值动作应该提升为专用工具。

工具分类体系

只读工具

  • 文件读取、日志查询、配置查看
  • 数据库只读查询、搜索接口
  • 系统状态检查、进程信息

只读工具可以并行调用、可以缓存结果、可以自动重试。设计时要考虑批量接口,避免 N+1 查询。

写入工具

  • 文件写入、配置修改
  • 数据库写入、消息发送
  • 部署操作、资源创建

写入工具需要幂等性设计、事务支持、回滚机制。返回值要包含操作后的状态,让模型能够验证结果。

高风险工具

  • 删除、格式化、强制重启
  • 权限变更、安全策略修改
  • 生产环境变更、数据迁移

高风险工具应该专用化,不要隐藏在通用 Shell 或 HTTP 请求里。每个高风险操作要有独立的工具、明确的参数、强制的确认机制。

工具权限模型

权限模型的核心是最小权限原则:默认拒绝,显式授权。

权限分级

  • 无限制:只读工具,可以自动执行
  • 需确认:写入工具,需要用户确认后执行
  • 需审批:高风险工具,需要单独的审批流程
  • 禁用:某些工具在特定环境下完全不可用

权限检查点

权限检查应该在三个层面进行:

  1. 工具注册时:哪些工具对哪些 Agent 可见
  2. 调用时:当前 Agent 是否有权限调用这个工具
  3. 执行时:底层系统是否允许这个操作

从通用工具到专用工具的演进路径

实践中通常先用通用工具探索,再把高频、高风险、高价值动作提升为专用工具。

第一阶段:通用工具探索

# 提供通用 bash 工具,但限制在沙箱环境
tools:
  - name: bash
    type: read_only
    constraints:
      allowed_commands: [ls, cat, grep, find]
      timeout: 5s

第二阶段:识别高频模式

从调用日志中发现哪些操作被频繁调用:

# 日志显示这些操作占 80% 调用
ls /app/config
cat /app/config/database.yml
grep -r "error" /app/logs

第三阶段:提升为专用工具

tools:
  - name: get_config
    description: "获取应用配置"
    parameters:
      - name: key
        type: string
        description: "配置键,支持点分隔路径"
 
  - name: search_logs
    description: "搜索应用日志"
    parameters:
      - name: query
        type: string
        description: "搜索关键词或正则表达式"
      - name: level
        type: enum
        values: [error, warn, info]

反模式

过宽的 bash 权限

直接给 Agent 无限制的 bash 权限是最常见的错误。这会让系统变得不可控,难以审计,难以调试。

过窄的专用工具

为每个小操作都创建一个专用工具,会导致工具爆炸、维护成本高、模型难以选择合适的工具。

缺少错误类型

所有错误都返回统一的错误码,让模型无法区分”参数错误”、“权限不足”、“资源不存在”等情况。

实践案例:一个代码 Agent 的工具箱设计

# 只读层
read_tools:
  - read_file:          # 读取文件内容
  - search_code:        # 搜索代码
  - get_git_status:     # 获取 git 状态
  - list_tests:         # 列出测试用例
 
# 写入层
write_tools:
  - create_file:        # 创建文件(需要确认)
  - update_file:        # 更新文件(需要确认)
  - run_tests:          # 运行测试
  - git_commit:         # 提交代码(需要确认)
 
# 高风险层(需审批)
critical_tools:
  - delete_branch:      # 删除分支
  - force_push:         # 强制推送
  - modify_production:  # 修改生产环境
 
# 工具权限矩阵
permissions:
  developer:
    read_tools: auto
    write_tools: confirm
    critical_tools: disabled
 
  senior_developer:
    read_tools: auto
    write_tools: auto
    critical_tools: confirm
 
  bot:
    read_tools: auto
    write_tools: disabled
    critical_tools: disabled

MCP:工具接入的标准化协议

MCP(Model Context Protocol)的价值在于把模型上下文和外部能力连接方式标准化。它让 Agent 不必为每个工具单独写一套私有协议,而是通过统一接口发现工具、读取资源、调用能力。

MCP 解决的三类问题

连接问题

不同工具如何以统一方式暴露给 Agent。每个工具不再需要实现自己的协议,只需要实现 MCP 接口。

上下文问题

外部资源如何被模型按需读取。MCP 提供了资源的标准化描述和访问方式。

治理问题

工具能力如何被列出、授权、审计和禁用。MCP 提供了工具发现和元数据的标准格式。

MCP 核心概念

Tools(工具)

工具是可调用的函数,有明确的输入输出:

{
  "name": "search_database",
  "description": "在数据库中搜索记录",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "搜索查询"
      },
      "limit": {
        "type": "integer",
        "default": 10
      }
    },
    "required": ["query"]
  }
}

Resources(资源)

资源是可读取的数据,可以是文件、数据库记录、API 响应:

{
  "uri": "file:///app/config.json",
  "name": "应用配置",
  "description": "当前应用的配置文件",
  "mimeType": "application/json"
}

Prompts(提示模板)

提示模板是预定义的提示模式,可以包含参数:

{
  "name": "code_review",
  "description": "代码审查提示模板",
  "arguments": [
    {
      "name": "file_path",
      "description": "要审查的文件路径",
      "required": true
    }
  ]
}

MCP Server 配置示例

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/mac08/workspace"],
      "disabled": false
    },
    "database": {
      "command": "python",
      "args": ["-m", "mcp_server.database", "--config", "db_config.json"],
      "env": {
        "DB_HOST": "localhost",
        "DB_PORT": "5432"
      }
    },
    "github": {
      "command": "node",
      "args": ["dist/index.js"],
      "cwd": "/path/to/github-mcp-server"
    }
  }
}

MCP vs 直接函数调用

维度MCP直接函数调用
复杂度较高,需要协议实现较低,直接调用
可扩展性高,易于添加新工具低,需要修改代码
标准化高,统一接口低,每个工具不同
适用场景多工具、多语言、长期维护单一应用、原型阶段

何时用 MCP

  • 需要集成多个外部工具
  • 工具由不同团队或语言实现
  • 需要工具发现和动态加载
  • 需要跨进程或跨机器调用

何时用直接函数调用

  • 单一应用内部的几个函数
  • 原型阶段快速验证
  • 工具数量少且固定

MCP 的安全考量

MCP 只是连接层,真正的治理仍然要靠权限模型、工具描述、调用审计和人为边界。

协议层面的安全

  • 传输加密(TLS)
  • 认证机制(API Key、OAuth)
  • 速率限制

工具层面的安全

  • 工具描述要准确,不能误导模型
  • 参数验证要严格
  • 权限检查要前置

使用层面的安全

  • 调用审计要完整
  • 异常检测要及时
  • 人工干预要明确

反模式

为单一应用引入 MCP

如果只有一个应用、几个函数,引入 MCP 是过度设计。直接函数调用更简单。

忽略权限模型

MCP 不自动带来安全,只是标准化了连接方式。权限模型仍然需要自己设计。

工具描述不准确

工具描述与实际功能不符,会让模型做出错误决策。

RAG:不是把资料塞给模型

RAG 的目标是让 Agent 在回答或行动前检索外部知识。它解决的是模型内置知识过期、不完整、不可追溯的问题。

但 RAG 不是简单的向量搜索。一个可用的 RAG 流程至少包含:

问题理解
  -> 查询改写
  -> 多路召回
  -> 相关性排序
  -> 证据压缩
  -> 带引用生成
  -> 答案校验

RAG 的完整 Pipeline

查询改写

用户查询可能模糊、有歧义、不完整。查询改写要:

  • 纠正拼写错误
  • 补充上下文
  • 拆分复杂查询
  • 扩展同义词
def rewrite_query(query: str, context: str) -> List[str]:
    """将原始查询改写为多个检索查询"""
    queries = []
    # 1. 原始查询
    queries.append(query)
    # 2. 去除停用词
    queries.append(remove_stopwords(query))
    # 3. 同义词扩展
    queries.append(expand_synonyms(query))
    # 4. 基于上下文的改写
    if context:
        queries.append(rewrite_with_context(query, context))
    return queries

多路召回

单一召回策略容易漏掉相关信息。多路召回要结合:

  • 向量检索(语义相似)
  • 关键词检索(精确匹配)
  • 混合检索(两者结合)
def multi_recall(query: str, top_k: int = 10) -> List[Document]:
    """多路召回相关文档"""
    results = []
    # 1. 向量检索
    vector_results = vector_search(query, top_k=top_k)
    results.extend(vector_results)
    # 2. 关键词检索
    keyword_results = keyword_search(query, top_k=top_k)
    results.extend(keyword_results)
    # 3. 去重
    results = deduplicate(results)
    return results

相关性排序

召回的文档需要重新排序,最相关的排在前面。排序要考虑:

  • 查询相关性
  • 文档质量
  • 时效性
  • 引用关系
def rerank(documents: List[Document], query: str) -> List[Document]:
    """重新排序文档"""
    scores = []
    for doc in documents:
        score = (
            relevance_score(doc, query) * 0.5 +
            quality_score(doc) * 0.3 +
            freshness_score(doc) * 0.2
        )
        scores.append((doc, score))
    return [doc for doc, _ in sorted(scores, key=lambda x: -x[1])]

证据压缩

检索到的文档可能很长,需要压缩成模型能处理的长度。压缩要保留关键信息:

  • 关键实体
  • 关键关系
  • 关键数据
  • 引用来源
def compress_evidence(documents: List[Document], max_length: int) -> str:
    """压缩证据"""
    # 1. 提取关键句子
    key_sentences = extract_key_sentences(documents)
    # 2. 按重要性排序
    key_sentences = rank_by_importance(key_sentences)
    # 3. 截断到最大长度
    compressed = ""
    for sentence in key_sentences:
        if len(compressed) + len(sentence) > max_length:
            break
        compressed += sentence + " "
    return compressed

带引用生成

生成答案时要包含引用,让用户能够追溯来源:

def generate_with_citation(question: str, evidence: str) -> Tuple[str, List[str]]:
    """生成带引用的答案"""
    answer = model.generate(
        question=question,
        evidence=evidence,
        format="answer_with_citation"
    )
    # 提取引用
    citations = extract_citations(answer)
    return answer, citations

答案校验

生成的答案需要校验:

  • 事实一致性
  • 逻辑一致性
  • 完整性
def validate_answer(answer: str, evidence: str) -> bool:
    """校验答案"""
    # 1. 事实一致性检查
    fact_check = check_factual_consistency(answer, evidence)
    # 2. 逻辑一致性检查
    logic_check = check_logical_consistency(answer)
    # 3. 完整性检查
    completeness_check = check_completeness(answer)
    return fact_check and logic_check and completeness_check

检索策略

向量检索

向量检索适合语义相似的场景:

  • 同义词查询
  • 概念查询
  • 跨语言查询

优点是能理解语义,缺点是可能漏掉精确匹配。

关键词检索

关键词检索适合精确匹配的场景:

  • 专有名词
  • 代码标识符
  • 技术术语

优点是精确匹配,缺点是不能理解语义。

混合检索

混合检索结合两者优点:

def hybrid_search(query: str, alpha: float = 0.5) -> List[Document]:
    """混合检索"""
    vector_results = vector_search(query)
    keyword_results = keyword_search(query)
    # 合并分数
    scores = {}
    for doc, score in vector_results:
        scores[doc.id] = scores.get(doc.id, 0) + alpha * score
    for doc, score in keyword_results:
        scores[doc.id] = scores.get(doc.id, 0) + (1 - alpha) * score
    return sorted(scores.items(), key=lambda x: -x[1])

Chunking 策略

固定大小

最简单的方式,按固定大小切分:

def chunk_fixed_size(text: str, size: int = 512) -> List[str]:
    """固定大小切分"""
    chunks = []
    for i in range(0, len(text), size):
        chunks.append(text[i:i+size])
    return chunks

优点是实现简单,缺点是可能切断语义。

语义分割

按语义单元切分(段落、章节):

def chunk_semantic(text: str) -> List[str]:
    """语义分割"""
    # 按段落分割
    paragraphs = text.split('\n\n')
    # 按句子合并
    chunks = []
    current = ""
    for para in paragraphs:
        if len(current) + len(para) > 512:
            chunks.append(current)
            current = para
        else:
            current += para
    if current:
        chunks.append(current)
    return chunks

优点是保持语义完整,缺点是实现复杂。

层级切分

文档有层级结构(章节、小节、段落):

def chunk_hierarchical(doc: Document) -> List[Chunk]:
    """层级切分"""
    chunks = []
    # 第一层:章节
    for chapter in doc.chapters:
        # 第二层:小节
        for section in chapter.sections:
            # 第三层:段落
            for para in section.paragraphs:
                chunks.append(Chunk(
                    content=para,
                    path=f"{chapter.title}/{section.title}",
                    level=3
                ))
    return chunks

优点是保持结构,缺点是需要结构化数据。

动态检索

Agent 在行动过程中需要按需检索:

def agent_with_dynamic_rag(task: str):
    """带动态检索的 Agent"""
    # 1. 理解任务
    plan = understand_task(task)
    # 2. 识别信息需求
    info_needs = identify_info_needs(plan)
    # 3. 按需检索
    context = {}
    for need in info_needs:
        context[need] = retrieve(need)
    # 4. 执行任务
    result = execute(plan, context)
    return result

代码 Agent 在修改前检索相关文件、接口文档和历史决策;客服 Agent 在回复前检索产品政策、用户订单和历史对话。

RAG 质量评估

检索质量

  • 召回率:是否找到所有相关文档
  • 精确率:找到的文档是否相关
  • 排序质量:相关文档是否排在前面

生成质量

  • 事实一致性:答案是否与检索内容一致
  • 完整性:答案是否完整回答问题
  • 可追溯性:是否有引用

端到端质量

  • 用户满意度
  • 问题解决率
  • 平均交互轮次

反模式

直接向量搜索不做后处理

只做向量搜索,不做查询改写、重排序、证据压缩,效果会很差。

Chunk 太大或太小

太大:一个 chunk 包含太多信息,检索不精确 太小:一个 chunk 信息不完整,模型难以理解

缺少引用追溯

生成的答案没有引用,用户无法验证,也无法深入了解。

三者协作架构

工具、MCP、RAG 的关系可以这样理解:

  • 工具是 Agent 的行动入口
  • MCP 是工具和资源的标准接入层
  • RAG 是知识检索和证据供给机制
                    +------------------+
                    |     Agent        |
                    +------------------+
                             |
        +--------------------+--------------------+
        |                    |                    |
+-------v-------+    +-------v-------+    +-------v-------+
|      MCP      |    |      RAG      |    |   直接工具    |
|  (发现层)     |    |  (知识层)     |    |   (行动层)    |
+-------+-------+    +-------+-------+    +-------+-------+
        |                    |                    |
+-------v-------+    +-------v-------+    +-------v-------+
|  MCP Servers  |    | 向量数据库    |    |  本地函数     |
|  - 文件系统   |    |  - 文档库     |    |  - HTTP API   |
|  - 数据库     |    |  - 代码库     |    |  - Shell      |
|  - GitHub     |    |  - 知识库     |    |  - 数据库     |
+---------------+    +---------------+    +---------------+

一个成熟 Agent 系统的协作流程:

1. Agent 接收任务
2. MCP 发现可用资源和工具
3. RAG 检索相关知识
4. Agent 规划行动步骤
5. 专用工具执行受控动作
6. 结果进入反馈循环
7. Agent 更新记忆和状态

工具层安全设计

工具层安全设计要考虑输入验证、权限检查、调用审计、敏感信息过滤。

输入验证

def validate_input(tool_name: str, params: Dict) -> bool:
    """验证工具输入"""
    # 1. 参数类型检查
    schema = get_tool_schema(tool_name)
    if not validate_schema(params, schema):
        return False
    # 2. 参数范围检查
    if not validate_range(params):
        return False
    # 3. 参数格式检查
    if not validate_format(params):
        return False
    # 4. 危险参数检查
    if has_dangerous_params(params):
        return False
    return True

权限检查

def check_permission(agent_id: str, tool_name: str, params: Dict) -> bool:
    """检查权限"""
    # 1. 工具级权限
    if not has_tool_permission(agent_id, tool_name):
        return False
    # 2. 资源级权限
    resource = extract_resource(params)
    if not has_resource_permission(agent_id, resource):
        return False
    # 3. 操作级权限
    action = extract_action(params)
    if not has_action_permission(agent_id, action):
        return False
    return True

调用审计

def audit_call(agent_id: str, tool_name: str, params: Dict, result: Any):
    """审计工具调用"""
    log = {
        "timestamp": time.time(),
        "agent_id": agent_id,
        "tool_name": tool_name,
        "params_hash": hash_params(params),
        "result_summary": summarize_result(result),
        "duration": result.duration,
        "success": result.success,
    }
    audit_log.append(log)

敏感信息过滤

def filter_sensitive_info(text: str) -> str:
    """过滤敏感信息"""
    # 1. 识别敏感信息
    patterns = [
        r'\b\d{16}\b',  # 信用卡号
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # 邮箱
        r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
    ]
    # 2. 替换为占位符
    for pattern in patterns:
        text = re.sub(pattern, '[REDACTED]', text)
    return text

不可逆动作的防护策略

def execute_with_protection(tool_name: str, params: Dict) -> Any:
    """执行不可逆动作"""
    # 1. 检查是否为不可逆动作
    if is_irreversible(tool_name):
        # 2. 预览影响
        impact = preview_impact(tool_name, params)
        # 3. 请求确认
        if not request_confirmation(impact):
            return None
        # 4. 创建回滚点
        rollback_point = create_rollback_point()
        try:
            # 5. 执行动作
            result = execute(tool_name, params)
            return result
        except Exception as e:
            # 6. 回滚
            rollback(rollback_point)
            raise e
    else:
        return execute(tool_name, params)

设计建议

工具层最重要的设计原则是:把不可逆动作做窄,把只读动作做快,把高频动作做稳。

只读工具可以并行、缓存、自动执行;写入工具要有权限、确认、回滚和日志;高风险工具要尽量专用化,不要隐藏在通用 Shell 或 HTTP 请求里。

MCP 让工具生态更容易扩展,但不自动带来安全。协议只是连接层,真正的治理仍然要靠权限模型、工具描述、调用审计和人为边界。

RAG 不是简单的向量搜索,而是一个完整的 pipeline:查询改写、多路召回、重排序、证据压缩、带引用生成、答案校验。每一步都不能少。

实践中,三者要协同工作:MCP 负责发现和连接,RAG 负责知识和证据,工具负责执行和反馈。一个成熟 Agent 系统往往三者都有。


原书相关章节

作者

Jimmy Song

智能体设计模式:智能系统构建实战指南