EYES 乱1. MCP协议与LLM交互概述
MCP（Model Context Protocol）是一种用于连接大语言模型（LLM）与外部工具的标准协议。它采用客户端-服务器架构，基于JSON-RPC 2.0进行通信，为LLM与外部系统的交互提供了标准化接口。
1.1 MCP核心组件
MCP协议定义了三种核心能力：
- 工具（Tools）：用于执行具体操作的功能模块
- 资源（Resources）：提供只读数据的接口
- 提示（Prompts）：预设的交互模板
这种设计使得LLM能够以结构化的方式访问外部功能，而无需了解底层实现细节。
1.2 LLM与MCP交互流程
LLM与MCP服务的交互主要通过MCP客户端作为中间桥梁完成，基本流程如下：
- 工具发现：客户端从MCP服务器获取可用工具列表
- 指令转换：将LLM指令转换为对工具的标准调用
- 结果返回：将执行结果反馈给LLM模型
这种架构设计使得LLM能够动态发现和使用外部能力，极大地扩展了模型的功能边界。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 交互过程详解
2.1 工具发现阶段
MCP客户端首先向服务器发送tools/list请求，获取服务器提供的工具列表。响应信息包括：
- 工具名称
- 功能描述
- 参数模式（JSON Schema）
客户端随后将这些工具定义转换为LLM可理解的"函数调用"（Function Calling）格式。这种转换确保了LLM能够正确理解和使用这些工具。
2.2 LLM交互阶段
当用户向LLM发送查询时，客户端会附带转换后的工具定义。LLM根据查询内容决定是否使用工具，以及使用哪个工具。如果需要使用工具，LLM会返回一个包含以下信息的"函数调用"请求：
- 工具名称
- 调用参数
这种设计使得LLM能够灵活地决定何时以及如何使用外部工具。
2.3 工具执行阶段
MCP客户端接收到LLM的调用请求后，会向对应的MCP服务器发送调用请求。服务器执行相应操作后，将结果返回给客户端，客户端再将结果传递给LLM。LLM结合这些结果生成最终回复。
整个过程实现了LLM与外部系统的无缝集成，使得LLM能够利用外部工具完成更复杂的任务。
3. 智能知识库助手实现
3.1 系统架构
我们实现了一个基于MCP协议的智能知识库助手，系统主要包含以下组件：
- SQLite存储：管理文档元数据和问答记录
- FAISS索引：实现高效的语义检索
- 嵌入服务：使用Sentence-Transformer生成文本向量
- MCP服务器：提供标准化的工具接口
这种架构结合了结构化存储和向量检索的优势，能够高效地处理知识库相关操作。
3.2 核心功能实现
3.2.1 文档管理
系统提供了完整的文档管理功能，包括：
- 文档添加
- 文档检索
- 文档更新
每个文档都会生成对应的嵌入向量，用于后续的语义检索。
3.2.2 语义检索
基于FAISS实现的语义检索功能，能够根据查询内容的语义相似度返回最相关的文档。检索过程包括：
- 生成查询文本的嵌入向量
- 在FAISS索引中搜索相似向量
- 返回对应的文档内容
这种检索方式比传统的关键词匹配更能理解用户的查询意图。
3.2.3 问答记录
系统会自动记录所有的问答交互，包括：
- 用户问题
- 系统回答
- 检索到的文档
- 响应时间
这些记录可用于后续的分析和优化。
4. 代码实现详解
4.1 环境配置
系统需要Python 3.11及以上版本，主要依赖包包括：
- mcp_use：MCP协议实现
- pysqlite3：SQLite数据库接口
- faiss-cpu：向量检索库
- sentence-transformers：文本嵌入模型
安装命令如下：
bash复制pip install mcp_use pysqlite3 faiss-cpu numpy sentence-transformers
4.2 MCP服务器实现
服务器核心代码如下：
python复制import os
import json
import sqlite3
import hashlib
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
from sentence_transformers import SentenceTransformer
import numpy as np
import faiss
from mcp.server.fastmcp import FastMCP
# 数据模型定义
@dataclass
class Document:
id: str
title: str
content: str
source: str
created_at: str
embedding: Optional[List[float]] = None
# SQLite存储实现
class SQLiteStore:
def __init__(self, db_path: str = "knowledge_base.db"):
self.db_path = db_path
self._init_db()
# 初始化数据库表结构
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS documents (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
source TEXT,
created_at TEXT NOT NULL,
embedding_hash TEXT
)
""")
# 其他表结构初始化...
conn.commit()
# FAISS索引实现
class FAISSIndex:
def __init__(self, dimension: int = 768, index_path: str = "kb_index.faiss"):
self.dimension = dimension
self.index_path = index_path
self.index: Optional[faiss.IndexFlatIP] = None
self.doc_ids: List[str] = []
self._load_or_create()
# 嵌入服务实现
class EmbeddingService:
def __init__(self, model: str = "maidalun1020/bce-embedding-base_v1"):
self.model = model
self.embedder = SentenceTransformer(self.model)
def generate(self, text: str) -> List[float]:
return self.embedder.encode(text[:8000])
# MCP工具定义
mcp = FastMCP("KnowledgeBaseAssistant", json_response=True)
@mcp.tool()
def search_knowledge_base(query: str, top_k: int = 3) -> str:
# 实现语义检索功能
pass
# 其他工具定义...
if __name__ == "__main__":
mcp.run(transport="stdio")
4.3 MCP客户端实现
客户端核心代码如下：
python复制import asyncio
import json
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def fetch_tools(session: ClientSession):
await session.initialize()
tool_result = await session.list_tools()
return [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
} for tool in tool_result.tools]
async def execute_tool_call(session, tool_name, tool_args):
try:
result = await session.call_tool(tool_name, arguments=tool_args)
return f"[Tool Result] {result.content}"
except Exception as e:
return f"[Error calling tool {tool_name}: {e}]"
async def main():
server_params = StdioServerParameters(command="python", args=["zhishi_assistant.py"])
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
available_tools = await fetch_tools(session)
query = "MCP 是什么？"
messages = [{"role": "user", "content": query}]
# 调用LLM并处理响应
response = client.chat.completions.create(
model=model_name,
messages=messages,
tools=available_tools,
tool_choice="auto"
)
# 处理工具调用和生成最终响应...
if __name__ == "__main__":
asyncio.run(main())
5. 实际应用示例
5.1 查询示例
当用户查询"MCP是什么？"时，系统会执行以下步骤：
- LLM决定使用
search_knowledge_base工具 - 客户端向MCP服务器发送检索请求
- 服务器返回相关文档
- LLM基于检索结果生成最终回答
5.2 回答格式
系统生成的回答通常包含以下要素：
- 概念定义
- 关键特性表格
- 核心作用说明
- 相关资源提示
例如：
code复制MCP是Model Context Protocol的缩写，由Anthropic于2024年11月发布...
| 特性 | 说明 |
|------|------|
| 架构 | 客户端-服务器 |
| 协议 | JSON-RPC 2.0 |
主要作用是为LLM与外部工具提供标准化接口...
6. 性能优化建议
6.1 缓存策略
- 实现查询结果缓存
- 对常用文档建立内存缓存
- 使用LRU算法管理缓存大小
6.2 索引优化
- 定期重建FAISS索引
- 实现增量索引更新
- 优化索引参数设置
6.3 异步处理
- 使用异步IO处理网络请求
- 实现批量操作支持
- 优化数据库访问模式
7. 常见问题排查
7.1 工具调用失败
可能原因：
- 工具名称不匹配
- 参数格式错误
- 服务器未正确实现工具
解决方案：
- 检查工具定义一致性
- 验证参数JSON Schema
- 查看服务器日志
7.2 检索结果不相关
可能原因：
- 嵌入模型不匹配
- 文档预处理不足
- 检索参数不合理
解决方案：
- 统一嵌入模型版本
- 优化文档预处理流程
- 调整top_k参数
7.3 性能瓶颈
可能原因：
- 向量索引过大
- 数据库查询未优化
- 网络延迟过高
解决方案：
- 分片向量索引
- 添加数据库索引
- 优化网络配置
8. 扩展应用场景
8.1 企业知识管理
- 集成内部文档系统
- 支持多模态内容
- 实现权限控制
8.2 智能客服系统
- 结合对话管理
- 支持多轮交互
- 集成业务系统
8.3 数据分析助手
- 连接数据库
- 执行复杂查询
- 可视化结果
这种基于MCP的架构可以灵活扩展到各种需要LLM与外部系统集成的场景，为构建智能应用提供了坚实基础。