乐于分享
好东西不私藏
当前位置：夜雨聆风 > 技术教程 > 软件教程 > openclaw6.9研究成果向量数据库Memory Search 局域网 BGE 模型替换指南

openclaw6.9研究成果向量数据库Memory Search 局域网 BGE 模型替换指南

当前时间： 2026-06-10 09:37:55 更新时间： 2026-06-10 分类：软件教程 评论(0)

openclaw6.9研究成果向量数据库Memory Search 局域网 BGE 模型替换指南

Memory Search 局域网 BGE 模型替换指南

场景：当 OpenClaw 的 memory_search 因 node-llama-cpp 缺失而不可用时，使用局域网已有的 BGE 模型作为 embedding provider。

完成时间：2026-06-09 08:05状态：✅ 已验证可用

问题背景

症状

memory_search is paused because the memory index was built with a different embedding provider/model/settings.

错误提示：index metadata is missing

根因

  • • node-llama-cpp 模块存在但导入失败（ESM TLA 问题）
  • • 本地 ONNX 模型文件 bge-small-zh-v1.5-onnx 不存在
  • • 官方的 chromadb-document-vectorizer 技能依赖 Sentence Transformers 或本地 ONNX

可用资源

局域网内另一台电脑（192.168.0.10）已部署：

  • • 模型：bge-small-zh-v1.5-f16.gguf（512 维嵌入）
  • • 服务：llama-cpp server，监听 0.0.0.0:8080
  • • API：/v1/embeddings 正常工作

解决方案概述

核心思路：创建一个局域网 API 调用版的 BGE 嵌入器，替代本地 ONNX 模型。

三步走策略：

  1. 1. 创建 tools/bge_onnx_embedder.py（局域网 API 调用版）
  2. 2. 修改 skills/chromadb-document-vectorizer-1.0.0/chromadb_document_vectorizer_simple.py
  3. 3. 重建 memory 向量库 + 修改 openclaw.json 配置

详细步骤

步骤 1：创建局域网 BGE 嵌入器

文件：tools/bge_onnx_embedder.py

#!/usr/bin/env python3"""BGE 嵌入器 - 局域网 API 调用版调用 192.168.0.10:8080 的 llama-cpp /v1/embeddings 接口替代本地 ONNX 模型，解决 node-llama-cpp 导入问题"""import jsonimport urllib.requestimport urllib.errorfrom typing import Listimport numpy as np# 局域网 BGE 模型 API 配置BGE_API_BASE = "http://192.168.0.10:8080"BGE_MODEL_NAME = "bge-small-zh-v1.5-f16.gguf"EMBEDDING_DIM = 512class BGEOnnxEmbedder:    """BGE 嵌入器 - 局域网 API 调用版"""    def __init__(self, model_path: str = None):        """        初始化嵌入器        Args:            model_path: 保留参数（兼容原接口），实际使用局域网 API        """        self.api_base = BGE_API_BASE        self.model_name = BGE_MODEL_NAME        self.dim = EMBEDDING_DIM        self._test_connection()    def _test_connection(self):        """测试 API 连接"""        try:            url = f"{self.api_base}/v1/models"            req = urllib.request.Request(url)            with urllib.request.urlopen(req, timeout=5) as response:                data = json.loads(response.read().decode('utf-8'))                models = data.get('data', [])                model_ids = [m['id'] for m in models]                if self.model_name in model_ids:                    print(f"[INFO] [OK] BGE 局域网 API 连接成功 - {self.model_name} ({self.dim}维)")                else:                    print(f"[WARN] 模型 {self.model_name} 未找到，可用模型: {model_ids}")        except Exception as e:            print(f"[ERROR] [FAIL] BGE 局域网 API 连接失败: {e}")            raise    def encode(self, texts: List[str], batch_size: int = 8) -> np.ndarray:        """        生成文本嵌入        Args:            texts: 文本列表            batch_size: 批量大小（保留参数，API 一次性处理）        Returns:            numpy 数组，形状 (len(texts), 512)        """        if not texts:            return np.array([]).reshape(0, self.dim)        embeddings = []        for text in texts:            try:                payload = {"input": text}                data = json.dumps(payload).encode('utf-8')                req = urllib.request.Request(                    f"{self.api_base}/v1/embeddings",                    data=data,                    headers={'Content-Type': 'application/json'}                )                with urllib.request.urlopen(req, timeout=30) as response:                    result = json.loads(response.read().decode('utf-8'))                    embedding_data = result['data'][0]['embedding']                    embeddings.append(embedding_data)            except Exception as e:                print(f"[ERROR] 文本嵌入失败: {str(e)[:100]}")                # 返回零向量作为降级                embeddings.append([0.0] * self.dim)        return np.array(embeddings, dtype=np.float32)    def encode_single(self, text: str) -> np.ndarray:        """生成单条文本嵌入"""        result = self.encode([text])        return result[0] if len(result) > 0 else np.zeros(self.dim, dtype=np.float32)def create_embedder() -> BGEOnnxEmbedder:    """创建嵌入器实例"""    return BGEOnnxEmbedder()if __name__ == "__main__":    import sys    import io    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')    print("[TEST] 测试 BGE 局域网嵌入器...")    embedder = BGEOnnxEmbedder()    # 测试嵌入    test_texts = ["测试中文嵌入", "OpenClaw 多 Agent 协作", "考试系统部署进度"]    for text in test_texts:        emb = embedder.encode_single(text)        print(f"  文本: {text[:20]}... | 维度: {emb.shape}, 非零元素: {np.count_nonzero(emb)}")    # 测试批量嵌入    batch_emb = embedder.encode(test_texts)    print(f"\n  批量嵌入: {batch_emb.shape}")    print("\n[OK] 测试完成!")

关键点：

  • • 使用 urllib 直接调用局域网 HTTP API
  • • 兼容原接口 BGEOnnxEmbedder 类名
  • • 自动测试连接，失败则抛出异常
  • • 降级策略：嵌入失败时返回零向量

步骤 2：修改向量化工具配置

文件：skills/chromadb-document-vectorizer-1.0.0/chromadb_document_vectorizer_simple.py

修改 1：调整导入路径，支持 tools.bge_onnx_embedder

原代码：

try:    from tools.bge_onnx_embedder import BGEOnnxEmbedder    HAS_BGE_ONNX = Trueexcept Exception:    HAS_BGE_ONNX = FalseBGE_MODEL_PATH = r"C:\Users\Administrator\bge-small-zh-v1.5-onnx\onnx\model.onnx"

修改为：

try:    import sys    from pathlib import Path    # Add workspace root to path for tools import    workspace_root = Path(__file__).parent.parent.parent.parent    if str(workspace_root) not in sys.path:        sys.path.insert(0, str(workspace_root))    from tools.bge_onnx_embedder import BGEOnnxEmbedder    HAS_BGE_ONNX = Trueexcept Exception:    HAS_BGE_ONNX = False# BGE模型路径 - LAN API (192.168.0.10:8080)BGE_MODEL_PATH = "http://192.168.0.10:8080"

修改 2：默认启用 BGE ONNX（局域网版）

在 __init__ 方法中：

self.use_bge_onnx = use_bge_onnx and HAS_BGE_ONNXif self.use_bge_onnx:    self.bge_embedder = BGEOnnxEmbedder(bge_model_path)    print("[INFO] BGE 局域网 API 嵌入器已初始化 (192.168.0.10:8080)")

并将默认参数改为：

use_bge_onnx: bool = True,  # LAN API default: True

关键点：

  • • 通过 Path(__file__).parent.parent.parent.parent 获取 workspace 根目录
  • • 将 workspace 根目录加入 sys.path，确保 tools 模块可导入
  • • BGE_MODEL_PATH 改为 LAN API 地址（传递给嵌入器构造函数，虽然实际未使用）

步骤 3：重建 Memory 向量库

使用新嵌入器重新向量化所有记忆文件。

脚本：temp/rebuild_memory_vectors.py

#!/usr/bin/env python3"""重建 Memory 向量库 - 使用局域网 BGE 模型"""import sysimport iosys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')from pathlib import Pathimport os# Add workspace root to pathworkspace = Path(__file__).parent.parentif str(workspace) not in sys.path:    sys.path.insert(0, str(workspace))skill_dir = workspace / "skills" / "chromadb-document-vectorizer-1.0.0"if str(skill_dir) not in sys.path:    sys.path.insert(0, str(skill_dir))from chromadb_document_vectorizer_simple import DocumentVectorizerprint("[INFO] 开始重建 Memory 向量库...")# 创建向量化工具 - 启用 BGE LAN APIpersist_dir = workspace / "skills" / "chromadb-document-vectorizer-1.0.0" / "chroma_data"vectorizer = DocumentVectorizer(    persist_directory=str(persist_dir),    collection_name="memory",    use_real_embedding=False,    use_hybrid_search=True,    use_bge_onnx=True,  # 启用局域网 BGE    chunk_size=300,    cache_size=500)# 清空旧数据print("\n[STEP 1] 清空旧向量库...")vectorizer.clear_collection()# 收集需要向量化的文件memory_dir = workspace / "memory"learn_dir = workspace / "learn"memory_md = workspace / "MEMORY.md"files_to_vectorize = []if memory_md.exists():    files_to_vectorize.append(("MEMORY.md", str(memory_md)))if memory_dir.exists():    for f in sorted(memory_dir.glob("*.md")):        if f.name not in ["events.jsonl", "short-term-recall.json"]:            files_to_vectorize.append((f"memory/{f.name}", str(f)))if learn_dir.exists():    for f in sorted(learn_dir.glob("*.md")):        files_to_vectorize.append((f"learn/{f.name}", str(f)))print(f"\n[STEP 2] 共找到 {len(files_to_vectorize)} 个文件")# 向量化print("\n[STEP 3] 开始向量化...")total_chunks = 0for name, path in files_to_vectorize:    result = vectorizer.vectorize_file(path)    if result["status"] == "success":        chunks = result["chunks_added"]        total_chunks += chunks        print(f"  [OK] {name}: {chunks} 个切片")    else:        print(f"  [FAIL] {name}: {result.get('message', 'unknown error')}")# 统计print("\n" + "="*50)stats = vectorizer.get_collection_stats()print(f"[INFO] 向量库重建完成!")print(f"  - 总文档切片数: {stats['total_documents']}")print(f"  - 总文件数: {stats['total_files']}")print("="*50)# 测试搜索print("\n[STEP 4] 测试语义搜索...")test_queries = ["部署进度", "考试系统", "模块化架构"]for q in test_queries:    results = vectorizer.search_vectors(q, top_k=3, use_cache=False)    if results:        print(f"\n  查询: \"{q}\"")        for i, r in enumerate(results, 1):            print(f"    [{i}] 相似度: {r['similarity']:.4f}")            print(f"        {r['content'][:80]}...")print("\n[OK] 重建完成!")

运行：

python "C:\Users\Administrator\.openclaw\workspace\temp\rebuild_memory_vectors.py"

输出示例：

[INFO] [OK] BGE 局域网 API 连接成功 - bge-small-zh-v1.5-f16.gguf (512维)[STEP 1] 清空旧向量库...[STEP 2] 共找到 102 个文件[STEP 3] 开始向量化...  [OK] MEMORY.md: 42 个切片  [OK] memory/2026-06-03.md: 18 个切片  ...[INFO] 已保存 1164 个文档切片到磁盘[STEP 4] 测试语义搜索...  查询: "部署进度"    [1] 相似度: 0.5866        **当前状态**：方案已制定，等待用户审批后执行...

步骤 4：修改 OpenClaw 全局配置

文件：openclaw.json

4.1 新增 bge-lan-512 提供商

在 models.providers 中添加：

"bge-lan-512": {  "baseUrl": "http://192.168.0.10:8080/v1",  "api": "openai-compatible",  "models": [    {      "id": "bge-small-zh-v1.5-f16.gguf",      "name": "BGE Small Zh v1.5 (LAN 512-dim)",      "input": ["text"],      "cost": { "input": 0, "output": 0 }    }  ]}

4.2 修改 agents.defaults.memorySearch

原配置：

"memorySearch": {  "provider": "custom-integrate-api-nvidia-com",  "model": "nvidia/nv-embed-v1"}

修改为：

"memorySearch": {  "provider": "bge-lan-512",  "model": "bge-small-zh-v1.5-f16.gguf",  "enabled":true}

4.3 验证 JSON 有效性

# PowerShell& "C:\Users\Administrator\AppData\Local\Programs\Python\Python312\python.exe" -c "import jsonpath = r'C:\Users\Administrator\.openclaw\openclaw.json'with open(path) as f:    d = json.load(f)print('memorySearch:', json.dumps(d['agents']['defaults']['memorySearch'], indent=2))print('bge-lan-512 provider:', json.dumps(d['models']['providers'].get('bge-lan-512'), indent=2))print('JSON valid: OK')"

预期输出：

memorySearch: {  "provider": "bge-lan-512",  "model": "bge-small-zh-v1.5-f16.gguf",  "enabled":true}bge-lan-512 provider: {  "baseUrl": "http://192.168.0.10:8080/v1",  "api": "openai-compatible",  "models": [...]}

验证测试

1. 嵌入器连通性测试

python "C:\Users\Administrator\.openclaw\workspace\tools\bge_onnx_embedder.py"

预期输出：

[TEST] 测试 BGE 局域网嵌入器...[INFO] [OK] BGE 局域网 API 连接成功 - bge-small-zh-v1.5-f16.gguf (512维)  文本: 测试中文嵌入... | 维度: (512,), 非零元素: 512  文本: OpenClaw 多 Agent 协作... | 维度: (512,), 非零元素: 512  批量嵌入: (3, 512)[OK] 测试完成!

2. 向量库重建验证

运行 rebuild_memory_vectors.py，检查输出：

  • • 总文档切片数: 1164
  • • 总文件数: 102
  • • 语义搜索测试返回相关片段（相似度 > 0.5）

3. OpenClaw Memory Search 状态

重启 OpenClaw 网关后，运行：

openclaw memory status --deep

应显示：

Provider: bge-lan-512Model: bge-small-zh-v1.5-f16.ggufIndex: readyTotal chunks: 1164

4. 实际使用测试

在对话中调用 memory_search：

# 示例results = await memory_search("部署进度", top_k=5)for r in results:    print(f"{r['similarity']:.2%} - {r['content'][:100]}")

应返回语义相关结果，而非空或降级提示。

注意事项

网络依赖

  • • OpenClaw 网关必须能访问 192.168.0.10:8080
  • • 如果局域网模型服务重启，OpenClaw 会抛出连接错误（降级为空结果）
  • • 建议将 BGE 模型服务设置为开机自启

性能

  • • 局域网嵌入延迟约 100-300ms/条
  • • 批量请求未优化（逐条发送），如需性能可修改 encode() 为批量发送
  • • 向量库重建时约需 5-10 分钟（取决于文件数量）

兼容性

  • • 本方案绕过了 node-llama-cpp 的 ESM 导入问题
  • • 不依赖本地 ONNX 运行时
  • • 512 维向量与原有 chroma_data 不兼容，必须重建

故障排查

问题
检查点
HAS_BGE_ONNX = False tools/bge_onnx_embedder.py

 是否在 workspace 根目录？sys.path 是否包含 workspace？
连接失败
 192.168.0.10:8080

 是否可达？防火墙是否开放？
memory_search 仍不可用
 openclaw.json

 的 memorySearch.provider 是否改为 bge-lan-512？是否重启网关？
搜索结果不相关
BGE 模型是否为中文优化？文本是否被正确分块（chunk_size=300）？

恢复默认（如需）

恢复 OpenAI embeddings

"memorySearch": {  "provider": "openai",  "model": "text-embedding-3-small"}

恢复本地 ONNX 模型

  1. 1. 删除 tools/bge_onnx_embedder.py
  2. 2. 恢复 chromadb_document_vectorizer_simple.py 原配置
  3. 3. 安装依赖：pip install onnxruntime sentence-transformers
  4. 4. 下载模型：bge-small-zh-v1.5-onnx 到 C:\Users\Administrator\bge-small-zh-v1.5-onnx\onnx\model.onnx
  5. 5. 重建向量库

参考文档

  • • OpenClaw Memory 配置：docs/concepts/memory-search.md
  • • 提供商配置：docs/reference/memory-config.md
  • • BGE 模型：https://github.com/FlagOpen/FlagEmbedding

更新日期：2026-06-09适用场景：node-llama-cpp 不可用，但有局域网 BGE embedding API 可用风险等级：低（仅修改配置，无数据丢失）维护人：张有德