网络内容爬取与向量存储服务

HoarderMCP是一个用于网络内容爬取、处理和向量存储的模型上下文协议服务器,提供网页内容提取、信息处理和向量相似性搜索功能。

  • by illogical
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 illogical/hoardermcp

使用说明

HoarderMCP

HoarderMCP是一个模型上下文协议(MCP)服务器,专为web内容抓取、处理和矢量存储而设计。它提供了用于摄取网络内容、提取相关信息并通过向量相似性搜索使其可搜索的工具。

特性

  • 网络爬虫:抓取网站和站点地图以提取内容
  • 高级内容处理:
    • 基于头部分割的Markdown语义分块
    • Python和C#的代码感知分块,语法保持不变
    • 可配置的块大小和重叠可实现最佳上下文
    • 基于令牌的大小优化
  • 矢量存储器:使用向量嵌入存储和搜索内容
  • API-第一个:RESTful API,便于与其他服务集成
  • 异步:内置async/await以实现高性能
  • 可扩展:支持多种矢量存储(Milvus、FAISS、Chroma等)
  • 可观测性:与Langfuse集成,用于跟踪和监测

安装

  1. 克隆存储库:

    git clone https://github.com/yourusername/hoardermcp.git
    cd hoardermcp
    
  2. 创建并激活虚拟环境:

    python -m venv venv
    source venv/bin/activate  # On Windows: venv\Scripts\activate
    
  3. 在开发模式下安装软件包:

    pip install -e .
    
  4. 安装开发依赖项:

    pip install -r requirements-dev.txt
    

先决条件

运行服务器

  1. 使用Docker启动Milvus:

    docker-compose up -d
    
  2. 运行开发服务器:

    python -m hoardermcp.main --reload
    

    API将于 http://localhost:8000

API文档

服务器运行后,您可以访问:

  • OpenAPI文档: http://localhost:8000/docs
  • ReDoc: http://localhost:8000/redoc

配置

配置可以通过环境变量或 .env 文件。看 .env.example 查看可用选项。

用法

摄入内容物

import httpx

# Ingest a webpage
response = httpx.post(
    "http://localhost:8000/ingest",
    json={
        "sources": [
            {
                "url": "https://example.com",
                "content_type": "text/html"
            }
        ]
    }
)
print(response.json())

高级分块示例

from hoardermcp.core.chunking import ChunkingFactory, ChunkingConfig, ChunkingStrategy
from hoardermcp.models.document import Document, DocumentMetadata, DocumentType

# Create a markdown document
markdown_content = """# Title\n\n## Section 1\nContent for section 1\n\n## Section 2\nContent for section 2"""
doc = Document(
    id="test",
    content=markdown_content,
    metadata=DocumentMetadata(
        source="example.md",
        content_type=DocumentType.MARKDOWN
    )
)

# Configure chunking
config = ChunkingConfig(
    strategy=ChunkingStrategy.SEMANTIC,
    chunk_size=1000,
    chunk_overlap=200
)

# Get appropriate chunker and process document
chunker = ChunkingFactory.get_chunker(
    doc_type=DocumentType.MARKDOWN,
    config=config
)
chunks = chunker.chunk_document(doc)

print(f"Document split into {len(chunks)} chunks")
for i, chunk in enumerate(chunks):
    print(f"Chunk {i + 1} (length: {len(chunk.content)}): {chunk.content[:50]}...")

搜索内容

import httpx

# Search for similar content
response = httpx.post(
    "http://localhost:8000/search",
    json={
        "query": "What is HoarderMCP?",
        "k": 5
    }
)
print(response.json())

发展

代码的风格

本项目使用:

  • 黑色 用于代码格式化
  • isort 用于进口分拣
  • 米皮 用于静态类型检查

提交前运行以下命令:

black .
isort .
mypy .

测试

使用pytest运行测试:

pytest