代码语义搜索与AI助手系统

MCP是一个本地化语义代码搜索和AI助手系统,能够智能索引Git仓库到向量嵌入中,并提供基于LLM的自然语言代码探索功能。适用于开发者在大型代码库中快速定位和理解代码。

  • by NirakhRastogi
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 NirakhRastogi/codebase_mcp

使用说明

MCP——最小码基处理器

MCP(Minimal Codebase Processor)是一个功能强大的本地语义代码搜索和AI辅助系统,可以智能地将Git存储库索引到向量嵌入中,并为自然语言代码探索提供LLM驱动的代理。

🚀 什么是MCP?

MCP由两个协同工作的核心组件组成:

🖥️ 服务器.py --MCP服务器

基于FastAPI的语义搜索引擎:

  • 克隆和曲目 Git仓库自动
  • 智能解析 使用AST编写Python代码,使用正则表达式模式编写Java/Kotlin/TypeScript/JavaScript代码
  • 创建向量嵌入 使用句子转换器(all-MiniLM-L6-v2)
  • 商店嵌入 在持久的Chroma矢量数据库中
  • 提供REST API 搜索和同步的端点
  • 支持增量索引 高效地跟踪存储库更改

🤖 agent.py --AI代码助理

一种智能代理,能够:

  • 翻译自然语言 将问题转化为精确的MCP搜索查询
  • 迭代搜索 使用多轮查询的代码库
  • 综合有根据的答案 仅基于检索到的代码片段
  • 避免幻觉 严格遵守MCP搜索结果
  • 提供交互式CLI 用于无缝代码探索

✨ 主要特点

  • 🧠 结构感知解析:Python(AST)和其他语言的高级代码解析(正则表达式模式)
  • ⚡ 快速语义搜索:支持CUDA的基于向量的相似性搜索
  • 🔄 增量同步:仅重新索引更改的文件以节省时间
  • 🛡️ 安全API:API密钥认证和速率限制(30 req/min)
  • 🎯 接地AI:LLM响应严格基于实际代码,没有幻觉
  • 🌐 多语言支持:Python、Java、Kotlin、TypeScript、JavaScript等
  • 📦 易于集成:用于嵌入工具和工作流的简单REST API

📋 目录


🏗️ 组件体系结构

Server.py-深度学习

这 MCP服务器 是一个生产就绪的FastAPI应用程序,具有以下功能:

🔍 智能代码解析

  • python:使用AST(抽象语法树)提取具有精确行号的函数和类
  • 多语言:基于正则表达式的Java、Kotlin、TypeScript、JavaScript解析
  • 回退堵塞:对不支持的文件类型进行基于文本的分块
  • 文件类型检测:智能MIME类型检测和二进制文件过滤

🧠 高级索引引擎

  • 异步处理:使用ThreadPoolExecutor进行非阻塞索引
  • 批量嵌入:以可配置的批处理文档(默认值:256)
  • GPU加速:可用时自动使用CUDA
  • 增量更新:Git提交跟踪,实现高效的重新索引

🗄️ 矢量存储器

  • Chroma DB:具有自动收集管理的持久矢量存储
  • 句子转换器: all-MiniLM-L6-v2 高质量嵌入模型
  • 元数据保存:存储文件路径、符号和代码结构信息

🛡️ 安全与性能

  • API密钥验证:可通过环境变量进行配置
  • 速率限制:基于IP的请求限制(默认30个请求/分钟)
  • CORS支持:用于网络集成的跨源资源共享
  • 性能监控:用于性能分析的内置时序装饰器

Agent.py-深潜

这 AI 代码助手 利用先进的快速工程和迭代搜索:

🎯 多级查询处理

  1. 查询生成:将自然语言转换为精确的搜索词
  2. 迭代搜索:执行最多3轮精确搜索
  3. 上下文聚合:合并多轮搜索的结果
  4. 接地合成:严格根据检索到的代码生成答案

🧩 LLM集成

  • Olama集成:用途 langchain_ollama 使用Llama3型号
  • 温度控制:低温(0.2)用于一致、真实的反应
  • 提示模板:针对不同推理阶段的专门提示
  • 幻觉预防:严格限制MCP搜索结果

🔄 智能搜索策略

  • 自适应查询:根据以前的结果调整搜索词
  • 停止状态检测:自动确定何时找到足够的上下文
  • 错误处理:MCP服务器不可用时性能下降
  • 结果验证:在处理之前验证和清除MCP响应

关键文件

  • server.py --实现索引、搜索和同步端点的FastAPI应用程序。
  • agent.py --命令行代理使用Ollama+LangChain风格的提示。
  • config.py -本地配置(API密钥、存储库URL、路径)。使用您的设置更新此文件。
  • requirements.txt --Python依赖关系。
  • embeddings_store/ --Chroma保存嵌入的目录。

🛠️ 需求

系统要求

  • python:3.10+(强烈推荐虚拟环境)
  • Git:用于存储库克隆和更新
  • 记忆:最低4GB RAM(建议大型存储库使用8GB+)
  • 存储:~ 1GB用于依赖关系+存储库大小+向量嵌入

硬件优化

  • 图形处理器:CUDA兼容GPU,可实现更快的嵌入生成(可选)
  • 中央处理器:建议使用多核处理器(使用 cpu_count() // 2 工人)
  • 固态硬盘:更快的I/O提高了索引性能

外部依赖

  • 奥拉玛:agent.py LLM功能需要
    # Install Ollama (visit https://ollama.ai for latest instructions)
    # Pull Llama3 model
    ollama pull llama3
    

Python依赖关系

安装所有必需的软件包:

pip install -r requirements.txt

关键依赖关系包括:

  • fastapi & uvicorn -Web服务器框架
  • langchain_chroma -矢量数据库集成
  • langchain_ollama -代理的LLM集成
  • sentence-transformers -文本嵌入模型
  • GitPython -Git存储库操作
  • torch -用于GPU加速的PyTorch
  • pydantic -数据验证和序列化

📦 安装和设置

1.克隆存储库

git clone <your-mcp-repository-url>
cd gitcode_mcp

2.创建虚拟环境(推荐)

# Create virtual environment
python -m venv mcp_env

# Activate virtual environment
# Windows:
mcp_env\Scripts\activate
# Linux/Mac:
source mcp_env/bin/activate

3.安装依赖项

pip install -r requirements.txt

4.安装和设置Olama

# Install Ollama (visit https://ollama.ai)
# After installation, pull the required model:
ollama pull llama3

5.配置系统

编辑 config.py 使用您的存储库详细信息:

REPO_NAME = 'your-project-name'
REPO_URL = "git@github.com:username/your-project.git"  # or HTTPS URL
REPO_DIR = "./repo/your-project"
API_KEY = "your-secure-api-key"

6.环境变量(可选但推荐)

# Set API key via environment variable for better security
export MCP_API_KEY="your-secure-api-key"

⚙️ 配置

更新 config.py 根据您的具体设置:

# Repository Configuration
REPO_NAME = 'your-project-name'           # Name of your project
REPO_URL = "git@github.com:user/repo.git"  # Git repository URL
REPO_DIR = "./repo/your-project"          # Local directory for cloned repo

# Security
API_KEY = "supersecret"                   # API key for server authentication

环境变量

服务器尊重以下环境变量:

  • MCP_API_KEY -覆盖范围 config.API_KEY 为了生产安全

高级配置

您可以在中修改这些常量 server.py 对于自定义:

MODEL_NAME = "all-MiniLM-L6-v2"    # Embedding model
BATCH_EMBED = 256                   # Batch size for embedding
INDEX_WORKERS = max(4, os.cpu_count() // 2)  # Indexing workers
MAX_REQ_PER_MIN = 30               # Rate limiting

🚀 运行系统

步骤1:启动MCP服务器

# Ensure you're in the project directory and virtual environment is activated
python server.py

第一次运行时会发生什么:

  1. 服务器检查存储库是否存在于 REPO_DIR
  2. 如果未找到,则克隆自 REPO_URL
  3. 在以下位置初始化Chroma矢量数据库 ./embeddings_store
  4. 执行完整的存储库索引(这可能需要几分钟)
  5. 在上启动FastAPI服务器 http://0.0.0.0:8000

服务器启动日志:

[INFO] Repository already exists.
[INFO] 🧠 No embeddings found — running full initial indexing...
[INFO] Embedding 1250 docs...
[INFO] ✅ Indexed 1250 docs.
[INFO] ✅ MCP Server ready on http://0.0.0.0:8000

MCP服务器启动并执行初始存储库索引

步骤2:运行AI代理(交互模式)

在新终端中(保持服务器运行):

# Activate virtual environment if needed
python agent.py

交互式会话示例:

🤖 Connected to MCP Code Assistant.

Ask a question (or 'exit'): How is user authentication implemented?

[LLM - Initial MCP Query]
user authentication login session management

================ ROUND 1 =================
[Executing MCP Query] user authentication login session management

✅ Retrieved 8 results from MCP.

🤖 LLM follow-up response: authentication middleware configuration

================ ROUND 2 =================
[Executing MCP Query] authentication middleware configuration

✅ Retrieved 5 results from MCP.

🤖 LLM follow-up response: STOP

🛑 LLM indicated to STOP further searches.

[Final Answer]
Based on the code analysis, user authentication is implemented using...

步骤3:直接使用API(可选)

您也可以通过HTTP直接查询服务器:

# Query the codebase
curl -X POST "http://localhost:8000/query" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: supersecret" \
  -d '{"query": "database connection setup", "top_k": 5}'

# Trigger repository sync
curl -X GET "http://localhost:8000/sync" \
  -H "X-API-Key: supersecret"

后台操作

对于生产部署,您可以将服务器作为后台服务运行:

# Using nohup (Linux/Mac)
nohup python server.py > mcp_server.log 2>&1 &

# Using screen (Linux/Mac)
screen -S mcp-server python server.py

# Using Windows Service or Task Scheduler (Windows)

📚 API 参考

MCP服务器公开具有以下端点的RESTful API:

认证

所有终结点都需要在请求头中包含一个API密钥:

X-API-Key: your-api-key

🔍 发布 /query

对索引代码库执行语义搜索。

请求

{
  "query": "authentication middleware setup",
  "top_k": 10
}

参数

  • query (字符串,必填):自然语言搜索查询
  • top_k (整数,可选):要返回的结果数(默认值:10)

响应

{
  "query": "authentication middleware setup",
  "results": [
    {
      "path": "src/auth/middleware.py",
      "snippet": "class AuthenticationMiddleware:\n    def __init__(self, secret_key):\n        self.secret_key = secret_key\n    \n    def authenticate_request(self, request):\n        token = request.headers.get('Authorization')\n        if not token:\n            raise UnauthorizedError('Missing token')",
      "score": 0.8234
    }
  ]
}

行为

  • 搜索前自动触发增量存储库同步
  • 返回截断为500个字符的代码片段
  • 结果按相似性得分排序(越高=相关性越高)

MCP服务器响应语义搜索查询的示例

🔄 获取 /sync

手动触发存储库同步和重新索引。

请求

GET /sync?full=false

参数

  • full (布尔值,可选):强制完全重新索引而不是增量索引(默认值:false)

响应

{
  "message": "Indexed 45 files."
}

行为

  • full=false:仅索引自上次提交以来更改的文件
  • full=true:重新索引整个存储库(速度较慢但很彻底)
  • 为未来的增量同步更新内部提交跟踪

📊 API响应代码

代码 描述
200 成功
401 API密钥无效或丢失
429 超过速率限制(30个要求/分钟)
500 内部服务器错误

🔧 示例用法

Python(请求)

import requests

headers = {
    'Content-Type': 'application/json',
    'X-API-Key': 'your-api-key'
}

# Search query
response = requests.post(
    'http://localhost:8000/query',
    json={'query': 'database connection pooling', 'top_k': 5},
    headers=headers
)
results = response.json()

# Manual sync
sync_response = requests.get(
    'http://localhost:8000/sync?full=true',
    headers=headers
)

JavaScript(获取)

const headers = {
    'Content-Type': 'application/json',
    'X-API-Key': 'your-api-key'
};

// Search query
const response = await fetch('http://localhost:8000/query', {
    method: 'POST',
    headers: headers,
    body: JSON.stringify({
        query: 'error handling patterns',
        top_k: 8
    })
});
const results = await response.json();

cURL

# Search for code
curl -X POST "http://localhost:8000/query" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-api-key" \
  -d '{"query": "API rate limiting implementation", "top_k": 3}'

# Force full reindex
curl -X GET "http://localhost:8000/sync?full=true" \
  -H "X-API-Key: your-api-key"

💡 使用示例

Agent.py互动会议

示例1:了解身份验证

Ask a question (or 'exit'): How does user authentication work in this codebase?

[LLM - Initial MCP Query]
user authentication login verification session management

================ ROUND 1 =================
[Executing MCP Query] user authentication login verification session management

✅ Retrieved 6 results from MCP.

🤖 LLM follow-up response: JWT token validation middleware

================ ROUND 2 =================
[Executing MCP Query] JWT token validation middleware

✅ Retrieved 4 results from MCP.

🤖 LLM follow-up response: STOP

[Final Answer]
Based on the code analysis, user authentication in this codebase works through a multi-layered approach:

1. **JWT Token-Based Authentication**: The system uses JSON Web Tokens stored in the `auth/jwt_handler.py` file...
2. **Middleware Layer**: Authentication middleware in `middleware/auth.py` intercepts requests...
3. **Session Management**: User sessions are managed through...

MCP代理执行迭代搜索并提供可靠的答案

示例2:查找配置

Ask a question (or 'exit'): Where are database connection settings configured?

[LLM - Initial MCP Query]
database connection configuration settings

================ ROUND 1 =================
[Executing MCP Query] database connection configuration settings

✅ Retrieved 5 results from MCP.

[Final Answer]
Database connection settings are configured in several locations:

1. **Main Configuration**: `config/database.py` contains the primary database connection parameters...
2. **Environment Variables**: The `.env` file includes database credentials...
3. **Connection Pool**: Database pool settings are defined in `utils/db_pool.py`...

API直接集成

示例1:构建VS代码扩展

import requests

class MCPCodeSearch:
    def __init__(self, api_key, base_url="http://localhost:8000"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            'Content-Type': 'application/json',
            'X-API-Key': api_key
        }
    
    def search_code(self, query, max_results=5):
        """Search for code snippets related to query"""
        response = requests.post(
            f"{self.base_url}/query",
            json={"query": query, "top_k": max_results},
            headers=self.headers
        )
        return response.json()["results"]
    
    def find_function_usage(self, function_name):
        """Find usages of a specific function"""
        return self.search_code(f"{function_name} function call usage")

# Usage in VS Code extension
mcp = MCPCodeSearch("your-api-key")
results = mcp.find_function_usage("authenticate_user")

示例2:代码审查助手

def analyze_pull_request(changed_files, mcp_client):
    """Analyze changed files and provide context"""
    analysis = []
    
    for file_path in changed_files:
        # Search for related code patterns
        results = mcp_client.search_code(f"similar patterns {file_path}")
        
        # Find potential dependencies
        deps = mcp_client.search_code(f"imports dependencies {file_path}")
        
        analysis.append({
            "file": file_path,
            "similar_patterns": results[:3],
            "dependencies": deps[:2]
        })
    
    return analysis

示例3:文档生成器

def generate_api_docs(mcp_client):
    """Generate API documentation from codebase"""
    
    # Find all API endpoints
    endpoints = mcp_client.search_code("API endpoint route decorator", 20)
    
    # Find authentication methods
    auth = mcp_client.search_code("authentication authorization middleware")
    
    # Find error handling
    errors = mcp_client.search_code("error handling exception status codes")
    
    return {
        "endpoints": endpoints,
        "authentication": auth,
        "error_handling": errors
    }

与开发工作流集成

示例1:预提交钩子

#!/bin/bash
# .git/hooks/pre-commit

# Trigger MCP reindex before commit
curl -X GET "http://localhost:8000/sync" \
  -H "X-API-Key: ${MCP_API_KEY}" \
  -s > /dev/null

echo "✅ MCP index updated"

示例2:CI/CD管道集成

# .github/workflows/code-analysis.yml
name: Code Analysis with MCP

on: [push, pull_request]

jobs:
  analyze:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Setup MCP Server
        run: |
          python -m pip install -r requirements.txt
          python server.py &
          sleep 30  # Wait for server to start
      
      - name: Analyze Code Patterns
        run: |
          python scripts/analyze_patterns.py
        env:
          MCP_API_KEY: ${{ secrets.MCP_API_KEY }}

高级查询技术

运行良好的自然语言查询:

  • ✅ “错误处理是如何实现的?”
  • ✅ “数据库迁移的定义在哪里?”
  • ✅ “使用了哪些身份验证方法?”
  • ✅ “日志记录是如何配置的?”
  • ✅ “API利率限制在哪里设置?”

技术问题:

  • ✅ “异步/等待模式用法”
  • ✅ “装饰器实现”
  • ✅ “类继承层次结构”
  • ✅ “配置文件解析”
  • ✅ “单元测试示例”

要避免的查询:

  • ❌ “我接下来应该实施什么?”(太开放了)
  • ❌ “这段代码好吗?”(主观)
  • ❌ “如何修复这个bug?”(没有上下文)

🔒 安全

API密钥验证

  • 基于标头:所有请求都需要 X-API-Key 头球
  • 环境变量:设置 MCP_API_KEY 生产环境变量
  • 密钥验证:服务器在每个请求上验证API密钥
# Secure API key setup
export MCP_API_KEY="your-very-secure-random-key"

速率限制

  • 基于IP的限制:每个IP地址每分钟30个请求
  • 内存存储:存储在服务器内存中的速率限制计数器
  • 自动清理:旧的费率限制条目已自动删除

CORS配置

  • 跨来源支持:已启用web应用程序集成
  • 可配置的起源:可以限制为特定域
  • 凭证支持:支持身份验证凭据

生产安全建议

1.API密钥管理


故障排除和提示

  • 确保仓库可访问(网络/凭据),以及 REPO_URL 是正确的。
  • 如果未创建嵌入或搜索返回空,请检查 ./embeddings_store 内容和服务器日志。
  • 如果您在大型存储库上看到索引速度较慢,请增加CPU工作时间或在后台作业中运行索引。
  • 要重置嵌入,请停止服务器,删除 ./embeddings_store,然后重新启动--将运行完整索引。
  • 如果LLM(Ollama)不可用,代理将无法生成查询;您可以手动创建查询并POST到 /query.

限制和下一步

  • 在这个简单的实现中,索引在主线程上是同步的——对于非常大的存储库,您应该将嵌入移动到后台工作器并提供状态端点。
  • Python之外的语言解析是启发式的;添加特定语言的解析器以获得更好的块。
  • 为大型结果集添加身份验证、TLS、持久速率限制存储和分页。

学分和许可证

这个项目是一个小型实用程序,可以使用向量搜索+LLM来帮助探索和推理本地代码库。将其作为起点,并根据您的需求进行扩展。

根据MIT许可证获得许可。