MBE 开放核心实现方案

一、仓库架构

GitHub: mises-ai/
│
├── mbe-core/                    # 🔓 公开 (MIT License)
│   ├── src/
│   │   ├── core/
│   │   │   ├── engine.py        # 基础引擎
│   │   │   ├── memory_base.py   # 记忆接口（抽象类）
│   │   │   └── config.py
│   │   ├── knowledge/
│   │   │   ├── expert_router.py # 基础路由
│   │   │   ├── dynamic_expert.py# 动态专家
│   │   │   └── rag_basic.py     # 基础 RAG
│   │   ├── api/                 # REST API
│   │   └── llm/                 # LLM 抽象层
│   ├── knowledge_bases/
│   │   └── experts/
│   │       ├── general_assistant/  # 通用助手
│   │       └── example_expert/     # 示例专家
│   ├── docker-compose.yml
│   ├── requirements.txt
│   └── README.md
│
├── mbe-pro/                     # 🔒 私有 (商业许可)
│   ├── titans/
│   │   ├── memory.py            # TITANS 完整实现
│   │   ├── short_term.py
│   │   ├── mid_term.py
│   │   └── long_term.py
│   ├── miras/
│   │   ├── matcher.py           # MIRAS 多尺度匹配
│   │   └── titans_rag.py
│   ├── moe/
│   │   └── expert_routing.py    # 43专家 MoE
│   ├── hope/
│   │   └── continuous_learning.py
│   └── experts/                 # 预训练专家
│       ├── chinese_civil_lawyer/
│       ├── nutritionist/
│       └── psychologist/
│
└── mbe-enterprise/              # 🔒 私有 (企业许可)
    ├── multi_tenant/            # 多租户
    ├── sso/                     # 企业 SSO
    ├── marketplace/             # 专家市场完整版
    ├── analytics/               # 数据分析
    └── ha/                      # 高可用集群

二、功能分层

社区版 (mbe-core) - 免费

专业版 (mbe-pro) - $99/月

企业版 (mbe-enterprise) - 定制

三、技术实现

3.1 插件化架构

核心接口定义 (mbe-core):

# src/core/memory_base.py (开源)

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class MemoryInterface(ABC):
    """记忆系统抽象接口"""
    
    @abstractmethod
    def encode(self, user_id: str, text: str) -> None:
        """编码并存储信息"""
        pass
    
    @abstractmethod
    def retrieve(self, user_id: str, query: str, top_k: int = 5) -> list:
        """检索相关记忆"""
        pass
    
    @abstractmethod
    def get_context(self, user_id: str, query: str) -> Dict[str, Any]:
        """获取上下文"""
        pass


class SimpleMemory(MemoryInterface):
    """简单记忆实现（开源版）- 基于 Redis"""
    
    def __init__(self):
        self.redis = get_redis()
    
    def encode(self, user_id: str, text: str) -> None:
        # 简单存储最近 N 条对话
        key = f"memory:{user_id}"
        self.redis.lpush(key, text)
        self.redis.ltrim(key, 0, 50)  # 保留最近50条
    
    def retrieve(self, user_id: str, query: str, top_k: int = 5) -> list:
        key = f"memory:{user_id}"
        return self.redis.lrange(key, 0, top_k - 1)
    
    def get_context(self, user_id: str, query: str) -> Dict[str, Any]:
        history = self.retrieve(user_id, query)
        return {"conversation_history": history}

高级实现 (mbe-pro):

# titans/memory.py (闭源)

from mbe_core.memory_base import MemoryInterface

class TITANSMemory(MemoryInterface):
    """TITANS 完整记忆实现（商业版）"""
    
    def __init__(self):
        self.short_term = ShortTermMemory(slots=64)
        self.mid_term = MidTermMemory(slots=256)
        self.long_term = LongTermMemory(slots=512)
        self.surprise_detector = SurpriseDetector()
    
    def encode(self, user_id: str, text: str) -> None:
        embedding = self._text_to_embedding(text)
        
        # 惊喜检测门控
        surprise = self.surprise_detector.calculate(embedding)
        
        # 短期记忆
        self.short_term.write(user_id, embedding, surprise)
        
        # 条件性巩固到中期
        if self._should_consolidate(user_id):
            self._consolidate_to_mid_term(user_id)
    
    def retrieve(self, user_id: str, query: str, top_k: int = 5) -> list:
        query_embedding = self._text_to_embedding(query)
        
        # 多层检索
        short_results = self.short_term.attention_retrieve(user_id, query_embedding)
        mid_results = self.mid_term.pattern_retrieve(user_id, query_embedding)
        long_results = self.long_term.semantic_retrieve(user_id, query_embedding)
        
        # 融合排序
        return self._merge_and_rank(short_results, mid_results, long_results, top_k)
    
    # ... 完整实现

3.2 动态加载机制

# src/core/loader.py (开源)

import importlib
import os
from loguru import logger

def load_memory_backend():
    """动态加载记忆后端"""
    
    # 检查是否有 Pro 许可证
    license_key = os.getenv("MBE_PRO_LICENSE")
    
    if license_key and validate_license(license_key):
        try:
            # 尝试加载 TITANS
            titans = importlib.import_module("mbe_pro.titans.memory")
            logger.info("✅ TITANS Memory loaded (Pro)")
            return titans.TITANSMemory()
        except ImportError:
            logger.warning("Pro license valid but module not found")
    
    # 回退到社区版
    from .memory_base import SimpleMemory
    logger.info("📦 Simple Memory loaded (Community)")
    return SimpleMemory()


def load_expert_router():
    """动态加载专家路由"""
    
    license_key = os.getenv("MBE_PRO_LICENSE")
    
    if license_key and validate_license(license_key):
        try:
            moe = importlib.import_module("mbe_pro.moe.expert_routing")
            logger.info("✅ MoE Router loaded (Pro)")
            return moe.MoEExpertRouter()
        except ImportError:
            pass
    
    from .expert_router import BasicExpertRouter
    logger.info("📦 Basic Router loaded (Community)")
    return BasicExpertRouter()

3.3 许可证验证

# src/core/license.py (开源)

import hashlib
import time
import requests
from typing import Optional

class LicenseManager:
    """许可证管理"""
    
    VALIDATION_URL = "https://license.mises-ai.com/validate"
    
    def __init__(self):
        self._license_key: Optional[str] = None
        self._license_type: str = "community"
        self._valid_until: int = 0
        self._features: list = []
    
    def validate(self, license_key: str) -> bool:
        """验证许可证"""
        try:
            # 在线验证
            response = requests.post(
                self.VALIDATION_URL,
                json={"key": license_key},
                timeout=5
            )
            
            if response.status_code == 200:
                data = response.json()
                self._license_key = license_key
                self._license_type = data.get("type", "community")
                self._valid_until = data.get("valid_until", 0)
                self._features = data.get("features", [])
                return True
                
        except Exception:
            # 离线验证（基于签名）
            return self._offline_validate(license_key)
        
        return False
    
    def has_feature(self, feature: str) -> bool:
        """检查功能权限"""
        if self._license_type == "enterprise":
            return True
        return feature in self._features
    
    @property
    def is_pro(self) -> bool:
        return self._license_type in ["pro", "enterprise"]
    
    @property
    def is_enterprise(self) -> bool:
        return self._license_type == "enterprise"


# 全局实例
_license_manager = None

def get_license_manager() -> LicenseManager:
    global _license_manager
    if _license_manager is None:
        _license_manager = LicenseManager()
        key = os.getenv("MBE_LICENSE_KEY")
        if key:
            _license_manager.validate(key)
    return _license_manager

四、发布流程

4.1 版本发布

# .github/workflows/release.yml (mbe-core)

name: Release

on:
  push:
    tags:
      - 'v*'

jobs:
  release:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Build Docker Image
        run: |
          docker build -t mises-ai/mbe-core:${{ github.ref_name }} .
          
      - name: Push to Docker Hub
        run: |
          echo ${{ secrets.DOCKER_TOKEN }} | docker login -u mises-ai --password-stdin
          docker push mises-ai/mbe-core:${{ github.ref_name }}
          
      - name: Create GitHub Release
        uses: softprops/action-gh-release@v1
        with:
          files: |
            dist/*.whl
            dist/*.tar.gz

4.2 PyPI 发布

# pyproject.toml (mbe-core)

[project]
name = "mbe-core"
version = "3.6.0"
description = "Mises Behavior Engine - AI Decision Platform"
license = {text = "MIT"}
authors = [{name = "MBE Team", email = "team@mises-ai.com"}]
readme = "README.md"

[project.optional-dependencies]
pro = ["mbe-pro"]  # 商业版依赖

五、用户体验

5.1 社区版安装

# Docker (推荐)
docker run -d -p 8000:8000 mises-ai/mbe-core:latest

# 或 pip
pip install mbe-core
mbe-core serve

5.2 升级到专业版

# 1. 获取许可证
# 访问 https://mises-ai.com/pricing 购买

# 2. 安装专业版
pip install mbe-pro

# 3. 配置许可证
export MBE_LICENSE_KEY="your-license-key"

# 4. 重启服务
docker-compose restart
# 日志会显示: ✅ TITANS Memory loaded (Pro)

5.3 功能对比提示

# src/api/upgrade_prompt.py

def check_and_prompt_upgrade(feature: str, user_request: str):
    """检查功能并提示升级"""
    
    license = get_license_manager()
    
    if not license.has_feature(feature):
        return {
            "available": False,
            "message": f"'{feature}' 需要 Pro 版本",
            "upgrade_url": "https://mises-ai.com/pricing",
            "fallback": get_community_alternative(feature)
        }
    
    return {"available": True}

用户看到的提示：

{
  "success": true,
  "answer": "面包发酵温度建议24-26°C...",
  "memory_type": "simple",
  "upgrade_hint": {
    "message": "升级到 Pro 版可获得 TITANS 记忆系统，AI 会记住你的所有对话偏好",
    "features": ["3层神经记忆", "43专家智能路由", "持续学习优化"],
    "url": "https://mises-ai.com/pricing"
  }
}

六、定价策略

七、时间线

文档版本: 1.0 | 更新: 2026-01-26