接入外部世界——工具、MCP 与知识检索
Agent 不能只在模型内部打转。要解决真实问题,它必须能够观察、检索、操作外部系统。这就是工具使用、MCP 和 RAG 的共同命题。
工具使用:把能力变成受控接口
Agent 要行动,就必须通过工具接触外部世界。工具可以是搜索、数据库查询、文件读写、浏览器控制、代码执行、消息发送、部署操作,也可以是业务系统 API。
工具设计的关键不是”模型会不会调用”,而是工具是否提供了合适的边界。
工具设计的五个关键属性
一个好工具应该具备:
明确的参数结构
参数应该有清晰的类型和约束。字符串参数需要说明格式(路径、URL、查询表达式),数字参数需要范围,枚举参数需要列出可选值。参数名称要与业务语言对齐,避免模型猜测。
清晰的成功和失败返回
工具返回应该区分成功和失败,失败时给出可操作的错误信息。不要返回 null 或空字符串让模型猜测。返回结构要稳定,避免在调用链中传递隐式状态。
可审计的调用日志
每次工具调用应该记录:调用时间、参数、调用者、返回值摘要、执行耗时。这些日志是调试、安全审计、成本分析的基础。
必要的权限和确认机制
高风险操作需要权限检查和人工确认。权限检查应该前置,在参数验证阶段就完成,避免执行到一半才发现无权限。确认机制要设计得当,既不能让每个操作都打断流程,也不能让危险操作静默执行。
足够窄的职责
一个工具应该只做一件事。bash 这样的通用工具只能作为探索手段,高频、高风险、高价值动作应该提升为专用工具。
工具分类体系
只读工具
- 文件读取、日志查询、配置查看
- 数据库只读查询、搜索接口
- 系统状态检查、进程信息
只读工具可以并行调用、可以缓存结果、可以自动重试。设计时要考虑批量接口,避免 N+1 查询。
写入工具
- 文件写入、配置修改
- 数据库写入、消息发送
- 部署操作、资源创建
写入工具需要幂等性设计、事务支持、回滚机制。返回值要包含操作后的状态,让模型能够验证结果。
高风险工具
- 删除、格式化、强制重启
- 权限变更、安全策略修改
- 生产环境变更、数据迁移
高风险工具应该专用化,不要隐藏在通用 Shell 或 HTTP 请求里。每个高风险操作要有独立的工具、明确的参数、强制的确认机制。
工具权限模型
权限模型的核心是最小权限原则:默认拒绝,显式授权。
权限分级
- 无限制:只读工具,可以自动执行
- 需确认:写入工具,需要用户确认后执行
- 需审批:高风险工具,需要单独的审批流程
- 禁用:某些工具在特定环境下完全不可用
权限检查点
权限检查应该在三个层面进行:
- 工具注册时:哪些工具对哪些 Agent 可见
- 调用时:当前 Agent 是否有权限调用这个工具
- 执行时:底层系统是否允许这个操作
从通用工具到专用工具的演进路径
实践中通常先用通用工具探索,再把高频、高风险、高价值动作提升为专用工具。
第一阶段:通用工具探索
# 提供通用 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: disabledMCP:工具接入的标准化协议
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