高效代码库分析MCP服务器架构计划
概述
一种MCP服务器,提供智能代码库分析,无需一次将所有内容加载到内存中。代码库托管在GitHub上,并在本地与高效的基于git的更新同步。
核心架构
1.存储库同步层
初始克隆
def initialize_repository(repo_url, local_path):
"""
Clone repository with optimizations
"""
# Shallow clone to save bandwidth and storage
git.clone(
repo_url,
local_path,
depth=1, # Only latest commit initially
single_branch=True, # Only main branch
filter='blob:none' # Blobless clone - fetch files on demand
)
# Store metadata
store_repo_metadata({
'url': repo_url,
'local_path': local_path,
'current_commit': get_current_commit_hash(),
'last_synced': timestamp()
})增量更新策略
def sync_repository():
"""
Pull only changes since last sync
"""
current_commit = get_stored_commit_hash()
# Fetch changes
git.fetch(origin='origin', depth=1)
# Get diff between commits
changed_files = git.diff(
current_commit,
'origin/main',
name_only=True
)
# Categorize changes
changes = {
'added': [],
'modified': [],
'deleted': [],
'renamed': []
}
for file_info in git.diff_tree(current_commit, 'origin/main'):
status = file_info.status
if status == 'A':
changes['added'].append(file_info.path)
elif status == 'M':
changes['modified'].append(file_info.path)
elif status == 'D':
changes['deleted'].append(file_info.path)
elif status == 'R':
changes['renamed'].append({
'old': file_info.old_path,
'new': file_info.new_path
})
# Pull the actual changes
git.pull(origin='origin', branch='main')
# Update indexes incrementally
process_changes(changes)
# Update commit hash
update_stored_commit_hash(get_current_commit_hash())
return changes智能同步调度
- 按需:当用户进行查询时,检查是否需要同步
- 周期性的:每N分钟进行一次背景同步(可配置)
- Webhook已触发:如果配置了GitHub webhook,则同步推送事件
- 手册:公开用于显式同步请求的工具
2.索引阶段(一次性+增量更新)
Git感知文件系统爬虫
- 目的:构建代码库的轻量级索引
- 策略:
- 使用git ls树进行文件发现(比文件系统扫描更快) - 从git中提取元数据(提交哈希、作者、最后修改时间) - 尊重 .gitignore 通过git自动 - 使用git责备来跟踪文件历史记录 - 以pgvector扩展名存储在PostgreSQL数据库中
符号提取器
- 目的:提取高级代码结构而不进行完全解析
- 索引什么:
- 函数/方法签名 - 类定义 - 进出口报表 - 注释和文档字符串(仅第一行) - 文件级元数据
- 工具:使用特定语言的解析器(Tree sitter用于多语言支持)
- 存储:结构化索引,包括:
{
file_path: string,
symbols: [{name, type, line_start, line_end, signature}],
imports: [string],
exports: [string],
hash: string // for change detection
}依赖关系图生成器
- 目的:映射文件之间的关系
- 数据:
- 导入/要求关系 - 调用图(轻量级,符号级) - 继承体系
- 存储:图形数据库或邻接列表
基于pgvector的语义分块
- 目的:创建可搜索的上下文块
- 策略:
- 按逻辑单元(函数、类、模块)分组 - 包括周围环境(导入、相关定义) - 生成语义搜索的嵌入 - 使用pgvector扩展名存储在PostgreSQL中 - 存储块元数据:文件、行号、父符号、git提交哈希
矢量存储模式:
-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;
-- Code chunks table with embeddings
CREATE TABLE code_chunks (
id SERIAL PRIMARY KEY,
file_id INTEGER REFERENCES files(id),
chunk_text TEXT,
chunk_type TEXT, -- 'function', 'class', 'module', etc.
symbol_name TEXT,
line_start INTEGER,
line_end INTEGER,
embedding vector(384), -- Dimension depends on model
commit_hash TEXT,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Create HNSW index for fast vector search
CREATE INDEX ON code_chunks
USING hnsw (embedding vector_cosine_ops);
-- Create GiST index for metadata filtering
CREATE INDEX idx_chunks_file ON code_chunks(file_id);
CREATE INDEX idx_chunks_symbol ON code_chunks(symbol_name);
CREATE INDEX idx_chunks_type ON code_chunks(chunk_type);2.查询处理流水线
查询分析层
User Query → Intent Classification → Strategy Selection意图类型:
find_definition:定位定义某物的位置find_usage:查找符号的所有用法explain_function:了解特定代码trace_flow:遵循执行路径find_similar:语义代码搜索architectural:高层结构问题
战略路由器
根据意图,选择适当的检索策略:
策略1:基于符号的查找 (快速)
- For:“函数X在哪里定义?”
- 流程:直接索引查找→ 返回文件+行号
- 加载:仅加载相关文件部分
策略2:依赖遍历 (中等)
- For:“此模块依赖于什么?”
- 流程:图形查询→ 返回连接节点
- 加载:仅依赖元数据
策略3:使用pgvector进行语义搜索 (中速)
- For:“查找处理身份验证的代码”
- 流程:
1. 使用与索引相同的模型嵌入查询 1. 基于pgvector余弦相似度的向量相似度搜索 1. 使用元数据过滤器以获得更好的结果 1. 按相关性+新近度排名 1. 返回带有上下文的前k个块
- SQL查询:
SELECT
c.id,
c.chunk_text,
c.symbol_name,
c.line_start,
c.line_end,
f.path,
1 - (c.embedding $1::vector) as similarity
FROM code_chunks c
JOIN files f ON c.file_id = f.id
WHERE
c.chunk_type = ANY($2) -- Optional: filter by type
AND 1 - (c.embedding $1::vector) > 0.7 -- Similarity threshold
ORDER BY c.embedding $1::vector
LIMIT 10;- 加载:仅匹配块+周围上下文
策略4:深度分析 (慢速,按需)
- For:“解释支付流程是如何工作的”
- 流程:
1. 使用语义搜索查找入口点 1. 遍历依赖关系图 1. 仅加载相关文件 1. 为法学硕士构建专注的背景 1. 增量流分析
3.上下文组装
智能上下文生成器
def build_context(query, max_tokens=8000):
relevant_symbols = find_relevant(query)
context = []
token_count = 0
# Priority order
for symbol in rank_by_relevance(relevant_symbols):
chunk = load_chunk(symbol)
if token_count + len(chunk) 15分钟前)
- 如果是:运行 `sync_repository()` 在后台
- 继续使用当前索引数据
1. **查询分析**:意图= `architectural + explain`
1. **符号搜索**:查询PostgreSQL中的“auth\*”符号→ 15 火柴
1. **语义搜索**:
- 嵌入查询:“身份验证系统”→ 矢量(384)
- 查询pgvector:SELECT c.*, f.path, 1 - (c.embedding $1::vector) as similarity FROM code_chunks c JOIN files f ON c.file_id = f.id WHERE 1 - (c.embedding $1::vector) > 0.7 ORDER BY c.embedding $1::vector LIMIT 5;
- 获取前5个相关块
1. **依赖遍历**:从依赖关系图中获取身份验证模块依赖关系
1. **上下文组装**:
- 从git加载主认证模块(2KB)
- 加载密钥中间件(1.5KB)
- 加载与身份验证相关的配置(0.5KB)
- 总计:4KB,8KB预算
1. **响应**:提供结构化的解释,包括:
- 带有GitHub URL的文件引用
- 行号
- 提交哈希用于版本跟踪
- 定义链接
## Git特定优化
### 1.金发克隆
git clone --filter=blob:none --depth=1
- 最初只下载树和提交对象
- 按需获取文件内容
- 为大型存储库节省约70%的初始克隆时间
### 2.备件检查(可选)
对于monorepos,只签出相关目录:
def setup_sparse_checkout(paths): """ Only checkout specified paths """ git.config('core.sparseCheckout', 'true') with open('.git/info/sparse-checkout', 'w') as f: for path in paths: f.write(f"{path}\n") git.read_tree('-mu', 'HEAD')
### 3.Git对象数据库查询
不要从磁盘读取文件,直接查询git:
def get_file_at_commit(file_path, commit_hash): """ Get file content without checkout """ return git.show(f"{commit_hash}:{file_path}")
### 4.高效的差分处理
def get_changed_files_efficient(from_commit, to_commit): """ Get changed files without full checkout """ return git.diff_tree( '--no-commit-id', '--name-status', '-r', from_commit, to_commit )
## 可扩展性考虑因素
- **小代码库(\10000个文件)**:
- 分层索引(首先进行模块级摘要)
- 智能同步(仅限活动分支)
- GPU批量嵌入生成
- 考虑多个工作进程
- 按存储库/模块使用PostgreSQL分区
- 浅深度增量git获取
**PostgreSQL性能调优**:
-- Tune HNSW index parameters for your dataset CREATE INDEX idx_chunks_embedding ON code_chunks USING hnsw (embedding vector_cosine_ops) WITH ( m = 16, -- Increase for better recall (16-64) ef_construction = 64 -- Increase for better index quality (64-200) );
-- Query-time tuning SET hnsw.ef_search = 100; -- Higher = better accuracy, slower
-- Connection pooling -- Use pgBouncer or SQLAlchemy pool for concurrent requests
**嵌入生成优化**:
def batch_embed_chunks(chunks, batch_size=32): """ Generate embeddings in batches for efficiency """ model = SentenceTransformer('all-MiniLM-L6-v2') model.to('cuda' if torch.cuda.is_available() else 'cpu')
all_embeddings = [] for i in range(0, len(chunks), batch_size): batch = chunks[i:i+batch_size] embeddings = model.encode( batch, convert_to_numpy=True, show_progress_bar=False ) all_embeddings.extend(embeddings)
return all_embeddings
## 成功指标
- 查询响应时间:90%的查询\85%
- PostgreSQL存储:约2-3x代码库大小(包括嵌入)
- 同步频率:每15分钟一次(可配置)
- 内存使用率:驻留内存\<500MB(无GPU)
## 部署架构
┌─────────────────────┐ │ GitHub Repo │ │ (Remote) │ └──────────┬──────────┘ │ │ git fetch/pull │ (incremental) ▼ ┌─────────────────────┐ ┌──────────────────┐ │ Local Git Clone │ │ PostgreSQL + │ │ (Shallow/Blobless)│◄────►│ pgvector │ └──────────┬──────────┘ └────────┬─────────┘ │ │ │ │ ▼ ▼ ┌─────────────────────┐ ┌──────────────────┐ │ MCP Server │◄────►│ Embedding Model │ │ (FastAPI/Python) │ │ (sentence-trans)│ └──────────┬──────────┘ └──────────────────┘ │ │ MCP Protocol ▼ ┌─────────────────────┐ │ Claude Desktop │ │ or API Client │ └─────────────────────┘
