bootstrap.py•9.53 kB
#!/usr/bin/env python3
"""
MemOS Bootstrap Module
解决外部依赖问题的启动模块，在项目入口最早位置插入，
用于绕过硬依赖如ollama、neo4j等，支持轻量级测试和开发。
"""
import sys
import types
from typing import Any
def setup_mock_modules():
"""设置模拟模块来绕过硬依赖"""
# 模拟ollama模块
if "ollama" not in sys.modules:
ollama_mock = types.ModuleType("ollama")
# 创建基本的Client类模拟
class MockClient:
def __init__(self, *args, **kwargs):
pass
def list(self):
return {"models": []}
def embeddings(self, *args, **kwargs):
# 返回模拟的embedding结果
return {"embedding": [0.0] * 1024}
def chat(self, *args, **kwargs):
return {"message": {"content": "Mock response"}}
ollama_mock.Client = MockClient
sys.modules["ollama"] = ollama_mock
# 模拟neo4j模块
if "neo4j" not in sys.modules:
neo4j_mock = types.ModuleType("neo4j")
class MockGraphDatabase:
@staticmethod
def driver(*args, **kwargs):
return MockDriver()
class MockDriver:
def __init__(self):
pass
def session(self):
return MockSession()
def close(self):
pass
class MockSession:
def run(self, *args, **kwargs):
return MockResult()
def close(self):
pass
class MockResult:
def data(self):
return []
neo4j_mock.GraphDatabase = MockGraphDatabase
sys.modules["neo4j"] = neo4j_mock
# 模拟qdrant_client模块
if "qdrant_client" not in sys.modules:
qdrant_mock = types.ModuleType("qdrant_client")
class MockQdrantClient:
def __init__(self, *args, **kwargs):
pass
def search(self, *args, **kwargs):
return []
def upsert(self, *args, **kwargs):
return True
qdrant_mock.QdrantClient = MockQdrantClient
sys.modules["qdrant_client"] = qdrant_mock
# 模拟chonkie模块
if "chonkie" not in sys.modules:
chonkie_mock = types.ModuleType("chonkie")
class MockSentenceChunker:
def __init__(self, *args, **kwargs):
pass
def chunk(self, text):
# 简单的句子分割
return text.split('.')
chonkie_mock.SentenceChunker = MockSentenceChunker
sys.modules["chonkie"] = chonkie_mock
# 模拟openai模块
if "openai" not in sys.modules:
openai_mock = types.ModuleType("openai")
class MockOpenAI:
def __init__(self, *args, **kwargs):
pass
openai_mock.OpenAI = MockOpenAI
sys.modules["openai"] = openai_mock
# 模拟transformers模块
if "transformers" not in sys.modules:
transformers_mock = types.ModuleType("transformers")
class MockAutoTokenizer:
@staticmethod
def from_pretrained(*args, **kwargs):
return MockTokenizer()
class MockTokenizer:
def encode(self, text):
return [1, 2, 3] # 模拟token
def decode(self, tokens):
return "mock decoded text"
class MockAutoModelForCausalLM:
@staticmethod
def from_pretrained(*args, **kwargs):
return MockModel()
class MockModel:
def __init__(self):
pass
def generate(self, *args, **kwargs):
return [[1, 2, 3]] # 模拟生成结果
class MockDynamicCache:
def __init__(self, *args, **kwargs):
pass
class MockBitsAndBytesConfig:
def __init__(self, *args, **kwargs):
pass
transformers_mock.AutoTokenizer = MockAutoTokenizer
transformers_mock.AutoModelForCausalLM = MockAutoModelForCausalLM
transformers_mock.DynamicCache = MockDynamicCache
transformers_mock.BitsAndBytesConfig = MockBitsAndBytesConfig
sys.modules["transformers"] = transformers_mock
# 模拟pydantic模块
if "pydantic" not in sys.modules:
pydantic_mock = types.ModuleType("pydantic")
class MockBaseModel:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
def model_dump(self):
return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}
def model_validate(self, data):
return self.__class__(**data)
def mock_field(*args, **kwargs):
return None
def mock_field_validator(*args, **kwargs):
def decorator(func):
return func
return decorator
def mock_model_validator(*args, **kwargs):
def decorator(func):
return func
return decorator
class MockConfigDict(dict):
pass
pydantic_mock.BaseModel = MockBaseModel
pydantic_mock.Field = mock_field
pydantic_mock.field_validator = mock_field_validator
pydantic_mock.model_validator = mock_model_validator
pydantic_mock.ConfigDict = MockConfigDict
sys.modules["pydantic"] = pydantic_mock
# 模拟redis模块
if "redis" not in sys.modules:
redis_mock = types.ModuleType("redis")
class MockRedis:
def __init__(self, *args, **kwargs):
self._data = {}
def get(self, key):
return self._data.get(key)
def set(self, key, value, *args, **kwargs):
self._data[key] = value
return True
def delete(self, *keys):
for key in keys:
self._data.pop(key, None)
return len(keys)
def exists(self, key):
return key in self._data
def keys(self, pattern="*"):
return list(self._data.keys())
redis_mock.Redis = MockRedis
sys.modules["redis"] = redis_mock
# 模拟numpy模块
if "numpy" not in sys.modules:
numpy_mock = types.ModuleType("numpy")
class MockArray:
def __init__(self, data):
self.data = data if isinstance(data, list) else [data]
def __len__(self):
return len(self.data)
def __getitem__(self, key):
return self.data[key]
def tolist(self):
return self.data
def mean(self):
return sum(self.data) / len(self.data) if self.data else 0
def std(self):
return 1.0 # 简单模拟
def mock_array(data):
return MockArray(data)
def mock_zeros(shape):
if isinstance(shape, int):
return MockArray([0.0] * shape)
return MockArray([0.0] * shape[0])
def mock_ones(shape):
if isinstance(shape, int):
return MockArray([1.0] * shape)
return MockArray([1.0] * shape[0])
numpy_mock.array = mock_array
numpy_mock.zeros = mock_zeros
numpy_mock.ones = mock_ones
numpy_mock.mean = lambda x: sum(x) / len(x) if x else 0
numpy_mock.std = lambda x: 1.0
sys.modules["numpy"] = numpy_mock
# 添加别名
sys.modules["np"] = numpy_mock
def setup_environment_variables():
"""设置环境变量来控制行为"""
import os
# 设置轻量级模式
os.environ.setdefault("MEMOS_LITE_MODE", "true")
os.environ.setdefault("MEMOS_EMBEDDER", "debug-dummy")
os.environ.setdefault("MEMOS_SKIP_HEALTH_CHECK", "true")
def initialize():
"""初始化bootstrap环境"""
setup_mock_modules()
setup_environment_variables()
print("🚀 MemOS Bootstrap initialized - Lite mode enabled")
print(" 📦 Mock modules: ollama, neo4j, qdrant_client, chonkie, openai, transformers")
print(" 🔧 Environment: MEMOS_LITE_MODE=true")
# 自动初始化（当模块被导入时）
if __name__ != "__main__":
initialize()
if __name__ == "__main__":
# 如果直接运行此脚本，显示状态信息
initialize()
print("\n📋 Bootstrap Status:")
print(f" ollama: {'✅' if 'ollama' in sys.modules else '❌'}")
print(f" neo4j: {'✅' if 'neo4j' in sys.modules else '❌'}")
print(f" qdrant_client: {'✅' if 'qdrant_client' in sys.modules else '❌'}")
# 测试模拟模块
try:
import ollama
client = ollama.Client()
print(f" ollama.Client: ✅ {type(client)}")
except Exception as e:
print(f" ollama.Client: ❌ {e}")
try:
import neo4j
driver = neo4j.GraphDatabase.driver("bolt://localhost:7687")
print(f" neo4j.GraphDatabase: ✅ {type(driver)}")
except Exception as e:
print(f" neo4j.GraphDatabase: ❌ {e}")