fusion (Yasinyaman)

融合

DuckDB支持LLM工具的内存分析引擎。

Fusion通过以下方式连接到PostgreSQL/MySQL数据库 扭曲 REST API，将数据加载到DuckDB中进行快速柱状分析，并通过MCP和OpenAI函数调用为LLM公开10个工具。

特性

• 10个LLM工具 — list_sources, describe_table, query_data, search_data, aggregate_data, create_view, list_views, refresh_view, load_table, cache_stats
• 双格式 --MCP（模型上下文协议）和OpenAI函数调用格式中的工具定义
• 3访问层 -MCP服务器（stdio）、REST API（FastAPI/HTTP）、Python SDK
• 查询下推 --尽可能将查询直接路由到源数据库，避免不必要的数据传输
• 懒加载 --仅在查询中实际引用时从源获取表数据
• SQL护栏 --阻止破坏性SQL（DROP、DELETE、INSERT）以保护数据完整性
• LRU缓存 --查询结果缓存，具有可配置的TTL，响应时间为毫秒
• 物化视图 --具有计划自动刷新功能的预先计算的聚合表
• 跨来源联盟 --在单个查询中跨多个数据库（PostgreSQL+MySQL）进行JOIN
• 自动发现 --自动从Warp中发现所有数据库和表

建筑

┌─────────────────────────────────────────────────────────────────────────────┐
│  1. Data Source Layer                                                       │
│  ┌──────────────┐    REST     ┌─────────────────┐                          │
│  │ PostgreSQL   │ ──────────► │                 │                          │
│  │ MySQL        │             │ WarpConnector    │  auto-discovery           │
│  └──────────────┘             │ (query pushdown) │  pagination, schema       │
│       Warp REST API           └────────┬────────┘                          │
└────────────────────────────────────────┼──────────────────────────────────┘
                                          │
┌─────────────────────────────────────────▼──────────────────────────────────┐
│  2. DuckDB Core Layer                                                       │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │ OLAPEngine                                                            │  │
│  │  • DuckDB (in-memory, columnar)   • QueryCache (LRU + TTL)            │  │
│  │  • SchemaCatalog (multi-source)   • MaterializedViewManager           │  │
│  │  • FetchStrategy (lazy load)      • SQLGuardrails (SELECT only)       │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────┬──────────────────────────────────┘
                                          │
┌─────────────────────────────────────────▼──────────────────────────────────┐
│  3. LLM Tool Layer                                                          │
│  ┌─────────────┐  ┌──────────────────┐  ┌─────────────────┐                 │
│  │ ToolExecutor│  │ 10 tools         │  │ MCP / REST / SDK│                 │
│  │ (dispatch)  │─►│ query_data, etc. │─►│ → LLM → Result  │                 │
│  └─────────────┘  └──────────────────┘  └─────────────────┘                 │
└─────────────────────────────────────────────────────────────────────────────┘

安装

pip install -e .

可选依赖关系：

pip install -e ".[mcp]"     # MCP Server support
pip install -e ".[rest]"    # REST API (FastAPI + uvicorn)
pip install -e ".[dev]"     # Development (pytest, ruff, mypy)
pip install -e ".[all]"     # Everything

快速开始

开发包

from fusion import OLAPEngine

engine = OLAPEngine(memory_limit="4GB")
engine.connect_source("mydb", {
    "type": "warp",
    "base_url": "http://localhost:8000",
    "database": "mydb",
})

executor = engine.get_tool_executor()

# Discover available data
sources = executor.list_sources()

# Run an analytical query (auto-loads referenced tables)
result = executor.query_data("SELECT * FROM mydb.orders LIMIT 10")

# Aggregate data
agg = executor.aggregate_data(
    table="mydb.orders",
    group_by="status",
    agg_column="amount",
    agg_func="SUM",
)

# Create a materialized view
executor.create_view(
    name="daily_revenue",
    sql="SELECT status, SUM(amount) as total FROM mydb.orders GROUP BY status",
    refresh="hourly",
)

MCP服务器（克劳德桌面/光标）

fusion-mcp --warp-url http://localhost:8000 --database mydb

在中配置 claude_desktop_config.json:

{
  "mcpServers": {
    "fusion": {
      "command": "fusion-mcp",
      "args": ["--warp-url", "http://localhost:8000", "--database", "mydb"]
    }
  }
}

自动发现所有数据库：

fusion-mcp --warp-url http://localhost:8000 --auto-discover

REST API

fusion-rest --warp-url http://localhost:8000 --auto-discover --port 9000

Swagger用户界面位于 http://localhost:9000/docs.关键端点：

OpenAI函数调用

from fusion import get_openai_tools, OLAPEngine

engine = OLAPEngine()
engine.connect_source("mydb", {"type": "warp", "base_url": "http://localhost:8000"})
executor = engine.get_tool_executor()

# Get tool definitions for OpenAI Chat Completions API
tools = get_openai_tools()

# When the LLM makes a tool call:
result = executor.execute("query_data", {"sql": "SELECT ..."})

工具

扭曲设置

Fusion使用 扭曲 作为其数据源网关：

git clone https://github.com/yasinyaman/warp.git
cd warp
docker compose up -d

Warp提供了一个RESTneneneba API，用于联合对PostgreSQL和MySQL数据库的访问。

项目结构

fusion/
├── __init__.py              # Public API exports
├── engine.py                # OLAPEngine — main orchestration
├── cache.py                 # QueryCache (LRU + TTL)
├── catalog.py               # SchemaCatalog — multi-source metadata
├── guardrails.py            # SQLGuardrails — blocks destructive SQL
├── result.py                # QueryResult — format conversions
├── strategy.py              # FetchStrategy — smart table loading
├── exceptions.py            # Custom exception hierarchy
├── connectors/
│   ├── base.py              # BaseConnector (abstract)
│   └── warp.py              # WarpConnector (Warp REST API)
├── tools/
│   ├── definitions.py       # 10 tool schemas (OpenAI + MCP)
│   ├── executor.py          # ToolExecutor — routes tool calls
│   ├── mcp_server.py        # MCP Server (stdio transport)
│   └── rest_server.py       # REST API Server (FastAPI)
└── views/
    └── materialized.py      # MaterializedViewManager

发展

pip install -e ".[all]"
pytest tests/ -v           # 236 tests
ruff check fusion/         # Lint
python -m demo.demo        # Demo with synthetic data

需求

• Python 3.10+
• DuckDB 1.2+
• 扭曲 （数据源网关）

许可证

Apache 2.0——请参阅 许可证 了解详情。