Kuzu Memory Graph MCP

Kuzu 内存图数据库 MCP 服务器

一个高性能的大型语言模型（LLM）内存服务器，采用具备语义搜索能力的Kuzu图数据库构建，并通过模型上下文协议（MCP）实现与AI助手和代理的无缝集成。

🌟 特点

• 基于图的记忆存储使用KuzuDB进行高效的关系遍历和复杂查询
• 多数据库支持同时操作多个Kuzu数据库，实现动态切换
• 数据库发现通过MCP资源自动发现可用数据库
• 动态主用切换在不重启服务器的情况下切换可写数据库
• 语义搜索使用句子变换器和MLX嵌入进行基于向量的相似性搜索
• 混合搜索结合基于文本和语义的搜索，以获取全面结果
• 快速且可靠针对AI/代理内存使用场景进行了优化，支持Apple Silicon加速
• 灵活实体模型支持自定义实体类型和关系
• 自动嵌入生成针对Apple Silicon优化的MLX嵌入式模型，支持回退方案

🚀 快速入门

先决条件

• Python 3.11或更高版本
• 紫外线 包管理器（推荐）
• KuzuDB 0.11.2+

安装

选项1：使用uv（推荐）

# Clone the repository
git clone https://github.com/jkear/kuzu-memory-graph-mcp.git
cd kuzu-memory-graph-mcp

# Install dependencies
uv sync

# Activate the virtual environment
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

选项2：使用pip

# Clone the repository
git clone https://github.com/jkear/kuzu-memory-graph-mcp.git
cd kuzu-memory-graph-mcp

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -e .

运行服务器

# Run using uv in development mode (recommended)
uv run kuzu-memory-server

# Or run directly with Python after activating the venv
python -m kuzu_memory_server

# Or use uvx for testing (installs from PyPI - for published package only)
# uvx kuzu-memory-server

⚙️ 配置

环境变量

MCP客户端配置

添加到您的MCP客户端配置中（例如，Claude Desktop） ~/Library/Application Support/Claude/claude_desktop_config.json):

用于开发（本地项目）

{
  "mcpServers": {
    "kuzu-memory": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/kuzu-memory-graph-mcp",
        "run",
        "kuzu-memory-server"
      ],
      "env": {
        "KUZU_MEMORY_DB_PATH": "/path/to/DBMS/memory.kuzu",
        "KUZU_DATABASES_DIR": "/path/to/DBMS",
        "KUZU_WRITABLE_DATABASES": "memory,prompt_engineering,research_papers"
      }
    }
  }
}

对于生产环境（已发布包）

{
  "mcpServers": {
    "kuzu-memory": {
      "command": "uvx",
      "args": ["kuzu-memory-server"],
      "env": {
        "KUZU_MEMORY_DB_PATH": "/path/to/DBMS/memory.kuzu",
        "KUZU_DATABASES_DIR": "/path/to/DBMS",
        "KUZU_WRITABLE_DATABASES": "memory,prompt_engineering,research_papers"
      }
    }
  }
}

📚 使用示例

多数据库设置

将您的数据库组织在一个专用目录中，并配置哪些数据库可以进行写入操作：

/path/to/databases/
├── memory.kuzu           # Writable database
├── prompt_engineer.kuzu  # Writable database
└── research_papers.kuzu  # Writable database

在您的环境中配置可写数据库：

export KUZU_WRITABLE_DATABASES="memory,prompt_engineer,research_papers"
export KUZU_DATABASES_DIR="/path/to/databases"

在可写数据库之间切换

使用 switch_primary_database 用于更改接收写操作的数据库的工具：

# Switch to prompt_engineer database
switch_primary_database(database="prompt_engineer")

# Now create entities in prompt_engineer
create_entity(
    database="prompt_engineer",
    name="Chain of Thought",
    entity_type="prompt_pattern",
    observations=["Improves reasoning", "Step-by-step thinking"]
)

数据库发现

通过MCP资源查询可用数据库：

# List all available databases
resource_result = await access_resource("kuzu://databases/list")
# Returns JSON with database metadata including writable status

创建实体

# Create a person entity in the 'memory' database
create_entity(
    database="memory",
    name="Jordan Kearfott",
    entity_type="person",
    observations=["Vibe engineer for Sales & Marketing", "Studies LLM Memory techniques", "Lives in Gainesville", "Mid at Python"]
)

# Switch to prompt_engineer and create an entity
switch_primary_database(database="prompt_engineer")
create_entity(
    database="prompt_engineer",
    name="Chain of Thought",
    entity_type="prompt_pattern",
    observations=["Improves reasoning", "Step-by-step thinking"]
)

# Switch to research_papers and create an entity
switch_primary_database(database="research_papers")
create_entity(
    database="research_papers",
    name="Attention Is All You Need",
    entity_type="paper",
    observations=["Transformer architecture", "Self-attention mechanism"]
)

建立关系

# Create relationship between entities in specific database
create_relationship(
    database="memory",
    from_entity="Sam Altman",
    to_entity="OpenAI",
    relationship_type="IS_CEO",
    confidence=0.9
)

搜索实体

# Search in specific database
search_entities(database="memory", query="software engineer", limit=10)

# Semantic search across research papers
semantic_search(database="research_papers", query="transformer architecture", limit=10, threshold=0.3)

# Get related entities in prompt database
get_related_entities(database="prompt_engineer", entity_name="Chain of Thought", max_depth=2)

跨数据库操作

# Get overview of all databases
summary = await get_graph_summary()  # No database param = all databases

# Get detailed summary of specific database
summary = await get_graph_summary(database="memory")

🛠️ MCP 工具与资源

服务器提供以下MCP工具和资源：

资源

工具

注写入操作（create_entity， create_relationship， add_observations) 仅适用于列表中列出的数据库 KUZU_WRITABLE_DATABASES. 使用 switch_primary_database 更改当前可写数据库。

🏗️ 建筑学

Kuzu内存图MCP服务器已进行全面重构，以消除双状态不一致性问题：

核心组件

1. MCP服务器层基于FastMCP的协议实现
2. 数据库管理器使用连接池实现的单源真数据库状态管理
3. 图数据库层KuzuDB用于实体和关系存储
4. 语义搜索层MLX/Sentence Transformers用于嵌入生成
5. 查询层使用向量相似度搜索执行Cypher查询

数据库连接架构

服务器实现了 混合连接池模式：

• WritableDatabaseManager（可翻译为“可写数据库管理器”）集中式数据库状态管理器
• 连接池数据库保持开放，按需创建连接
• “Single Source of Truth”翻译成中文是“唯一数据源”或“真理单一来源”。这个短语通常用于描述一个系统或环境中，所有信息和数据都来自同一个权威、准确且最新的来源，以确保信息的一致性和准确性所有数据库状态仅存在于WritableDatabaseManager中
• 自动恢复内置的连接健康检查和恢复功能
• 快速切换在\=0.11.2**高性能图数据库
• modelcontextprotocol>=0.1.0（模型上下文协议版本大于等于0.1.0）MCP协议实现
• sentence-transformers 版本大于等于 5.1.1文本嵌入模型
• mlx-embeddings>=0.0.4（中文可译为：“mlx-embeddings版本需大于等于0.0.4”）针对Apple Silicon优化的嵌入式（技术/功能）
• numpy版本大于等于2.3.3数值计算
• polars版本大于等于1.34.0数据操作
• pyarrow >= 21.0.0列式数据格式
• networkx版本大于等于3.5图算法
• scipy版本大于等于1.16.2科学计算
• mcp版本大于等于1.16.0MCP 客户端/服务器库

🔧 开发环境设置

1. 克隆仓库
2. 使用（以下工具/方法）进行安装 uv sync --dev
3. 使用（某工具/环境）运行测试 python test_server.py
4. 启动开发服务器，使用 uvx run . 或者 uvx run . kuzu-memory-server

🚀 部署

关于生产部署的考虑，请参阅 DEPLOYMENT.md 翻译为中文是：“部署说明.md” 或 “部署文件.md”（具体翻译可能根据上下文有所调整，但“DEPLOYMENT”通常指“部署”的意思，“.md”是Markdown文件的扩展名）。

📚 文档

应用生命周期管理

服务器使用上下文管理器进行资源管理，并支持动态数据库切换：

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with database and model initialization."""
    # Parse writable databases configuration
    writable_dbs_str = os.getenv('KUZU_WRITABLE_DATABASES', '')
    writable_databases = [db.strip() for db in writable_dbs_str.split(',') if db.strip()]
    
    # Get databases directory
    databases_dir = os.getenv('KUZU_DATABASES_DIR', './DBMS')
    
    # Load embedding models
    embedding_model = load_embeddings()
    tokenizer = load_tokenizer()
    
    # Initialize database manager
    db_manager.initialize(writable_databases, databases_dir, embedding_model, tokenizer)
    
    # Switch to initial primary database
    initial_db = writable_databases[0] if writable_databases else get_primary_db_name(db_path)
    success, msg = db_manager.switch_to(initial_db)
    
    try:
        yield AppContext(
            db_manager=db_manager,
            embedding_model=embedding_model,
            tokenizer=tokenizer,
            databases_dir=databases_dir
        )
    finally:
        # Cleanup all connections
        db_manager.cleanup()

嵌入生成

该服务器支持MLX（Apple Silicon）和Sentence Transformers：

def generate_embedding(model: Any, tokenizer: Any, text: str) -> list[float]:
    """Generate 384-dimensional embedding using MLX or fallback."""
    if not text or not text.strip():
        return [0.0] * 384
    
    # Try MLX first
    try:
        import mlx.core as mx
        inputs = tokenizer.encode(text.strip(), return_tensors="mlx")
        outputs = model(inputs)
        return outputs.text_embeds.tolist()
    except:
        # Fallback to sentence transformers
        embedding = embedding_model.encode(text.strip(), convert_to_numpy=True)
        return embedding.tolist()

数据库模式

图表模式是通过编程方式创建的：

# Entity node table
conn.execute("""
    CREATE NODE TABLE IF NOT EXISTS Entity (
        name STRING PRIMARY KEY,
        type STRING,
        observations STRING[],
        embedding FLOAT[384],
        created_date DATE DEFAULT current_date(),
        updated_date DATE DEFAULT current_date()
    )
""")

# Relationship table
conn.execute("""
    CREATE REL TABLE IF NOT EXISTS RELATED_TO (
        FROM Entity TO Entity,
        relationship_type STRING,
        confidence FLOAT DEFAULT 1.0,
        created_date DATE DEFAULT current_date()
    )
""")

测试

测试结构

该项目包含多种测试方法：

1. 基本功能测试 （test_server.py)

- 测试数据库初始化 - 验证嵌入生成 - 验证基本查询执行

1. 单元测试 （在 tests/ （目录）

- 单个部件测试 - 用于隔离的模拟依赖项 - 边缘情况验证

1. 集成测试

- 端到端工作流程测试 - MCP协议验证 - 性能基准

编写测试

实体创建示例测试：

import pytest
from unittest.mock import Mock, AsyncMock
from src.kuzu_memory_server import create_entity

@pytest.mark.asyncio
async def test_create_entity():
    # Setup mock context
    mock_ctx = Mock()
    mock_ctx.request_context.lifespan_context = Mock()
    mock_ctx.request_context.lifespan_context.conn = AsyncMock()
    
    # Mock database response
    mock_ctx.request_context.lifespan_context.conn.execute.return_value = Mock()
    mock_ctx.request_context.lifespan_context.conn.execute.return_value.has_next.return_value = False
    
    # Test entity creation
    result = await create_entity(
        mock_ctx,
        name="Test Entity",
        entity_type="test",
        observations=["Test observation"]
    )
    
    # Assertions
    assert result["status"] == "created"
    assert result["name"] == "Test Entity"
    assert result["type"] == "test"

测试数据管理

对于测试，请使用单独的数据库：

# In test setup
test_db_path = "./test_memory.kuzu"
os.environ["KUZU_MEMORY_DB_PATH"] = test_db_path

# Cleanup after tests
if os.path.exists(test_db_path):
    shutil.rmtree(test_db_path)

调试

记录日志

添加日志以调试问题：

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# In your code
logger.info(f"Creating entity: {name}")
logger.debug(f"Generated embedding dimension: {len(embedding)}")

常见的调试场景

1. 数据库连接问题:

   # Check database path and permissions
   print(f"text {var}", file=sys.stderr)f"Database path: {db_path}")
   print(f"text {var}", file=sys.stderr)f"Database exists: {os.path.exists(db_path)}")

1. 嵌入生成问题：

   # Test embedding generation
   test_embedding = generate_embedding(model, tokenizer, "test")
   print(f"text {var}", file=sys.stderr)f"Embedding dimension: {len(test_embedding)}")
   print(f"text {var}", file=sys.stderr)f"Sample values: {test_embedding[:5]}")

1. MCP工具注册：

   # List registered tools
   print(f"text {var}", file=sys.stderr)"Registered tools:", list(mcp.tools.keys()))

性能分析

使用Python内置的性能分析器：

import cProfile
import pstats

# Profile embedding generation
profiler = cProfile.Profile()
profiler.enable()

# Your code here
embedding = generate_embedding(model, tokenizer, text)

profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)

🤝 贡献（或“参与贡献”）

1. 为仓库创建分支（或：克隆仓库）
2. 创建一个特性分支（git checkout -b feature/amazing-feature)
3. 提交您的更改（git commit -m 'Add amazing feature'）
4. 推送到分支（git push origin feature/amazing-feature)
5. 提交拉取请求

📄 许可证

这个项目采用MIT许可证授权——详见 许可证 文件中有详细信息。

🙏 致谢

• KuzuDB（注：KuzuDB是一个数据库系统或技术的名称，直接翻译为中文可能无法准确传达其专业含义，因此在此保留原英文名称，若需具体解释或命名，需根据上下文或官方资料确定） 用于高性能图数据库
• FastMCP 对于MCP框架
• 句子变换模型（或句子表示模型） 用于嵌入模型
• MLX 针对Apple Silicon的优化