第一步向量记忆落地详细步骤指南
技能内容完整。现在把这份完整指南直接输出给你：
🔥 火山Embedding向量记忆落地Hermes — 完整指南
🎯 这是什么
将火山引擎的 doubao-embedding-vision Embedding模型集成到Hermes Agent的记忆系统中，让 session_search 工具从纯关键词搜索（FTS5）升级为关键词+语义向量双路召回。
为什么需要这个？
FTS5关键词搜索的致命缺陷：用户换一种说法就搜不到。
| 用户查询 | FTS5 | 向量搜索 |
|---|---|---|
| 「系统空间不够了怎么办」 | ❌ 0结果 | ✅ sim=0.58（找到"磁盘清理"session） |
| 「怎么连远程主机传文件」 | ❌ 0结果 | ✅ sim=0.60（找到"FTP上传"session） |
| 「服务器安全检查」 | ❌ 0结果 | ✅ sim=0.61（找到"安全审计"session） |
向量搜索让Hermes具备了语义理解能力——即使用户的措辞和原文完全不同，也能通过语义相似度找到相关对话。
📋 前置条件
| 条件 | 说明 |
|---|---|
| Hermes Agent | 已安装并可正常运行 |
| 火山引擎API | 已开通，有有效API Key |
| Python环境 | Hermes自带venv中有 `httpx`（0.28+） |
| 无需额外依赖 | 纯Python实现，不需要numpy/faiss/chroma |
🔑 火山引擎Embedding API信息
| 项目 | 值 |
|---|---|
| 模型名称 | `doubao-embedding-vision` |
| API端点(Base URL) | `https://ark.cn-beijing.volces.com/api/coding/v3` |
| Embedding端点 | Base URL + `/embeddings` |
| 认证方式 | Bearer Token（API Key格式：`ark-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx-xxxxx`） |
| 向量维度 | 2048 |
| 批量支持 | 是，`input` 字段支持字符串数组 |
| API兼容 | OpenAI Embeddings API格式 |
API调用示例
curl -s -X POST "https://ark.cn-beijing.volces.com/api/coding/v3/embeddings" \
H "Content-Type: application/json" \
H "Authorization: Bearer ark-你的API密钥" \
d '{
"model": "doubao-embedding-vision",
"input": ["你好世界", "测试文本"]
}'
正常返回格式：
{
"data": [
{"embedding": [0.0294, -0.0134, ...], "index": 0},
{"embedding": [0.0156, -0.0089, ...], "index": 1}
],
"model": "doubao-embedding-vision",
"usage": {"prompt_tokens": 8, "total_tokens": 8}
}
🚀 落地步骤（5步）
Step 1: 配置memorySearch（Hermes配置文件）
在Hermes的 ~/.hermes/config.yaml 中的 agents.defaults 下添加 memorySearch 配置块：
agents:
defaults:
memorySearch:
provider: openai
model: doubao-embedding-vision
remote:
baseUrl: https://ark.cn-beijing.volces.com/api/coding/v3
apiKey: ark-你的API密钥
⚠️ 位置注意：必须放在 agents.defaults 下，与 model 配置同级。
Step 2: 修改hermes_state.py（数据库层）
这是核心改动，涉及5个部分：
2.1 升级Schema版本
找到文件顶部的 SCHEMA_VERSION，从当前值+1：
SCHEMA_VERSION = 7 # 原来是6
2.2 添加EMBEDDING_SQL表定义
在 FTS_SQL 常量之后添加：
EMBEDDING_SQL = """
CREATE TABLE IF NOT EXISTS message_embeddings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
msg_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
session_id TEXT NOT NULL,
embedding BLOB NOT NULL,
text_hash TEXT NOT NULL,
dims INTEGER NOT NULL DEFAULT 2048,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_embeddings_msg_id ON message_embeddings(msg_id);
CREATE INDEX IF NOT EXISTS idx_embeddings_session_id ON message_embeddings(session_id);
CREATE INDEX IF NOT EXISTS idx_embeddings_text_hash ON message_embeddings(text_hash);
"""
设计要点：
embedding 存为BLOB（float32数组打包），每条2048维 = 8KB，1000条≈8MB
text_hash 用于去重：文本未变则跳过重新embedding，节省API调用
msg_id 外键关联messages表，CASCADE删除
2.3 添加辅助函数
在 _text_hash() 函数附近添加：
import struct as _struct
import hashlib as _hashlib
def _pack_embedding(vec: list) -> bytes:
"""Pack a float32 vector into a compact BLOB (little-endian)."""
return _struct.pack(f"<{len(vec)}f", *vec)
def _unpack_embedding(blob: bytes) -> list:
"""Unpack a BLOB back into a float32 list."""
n = len(blob) // 4
return list(_struct.unpack(f"<{n}f", blob))
def _cosine_similarity(a: list, b: list) -> float:
"""Pure-Python cosine similarity — no numpy needed.
For ~1000 vectors of 2048 dims this runs in <100 ms.
"""
dot = 0.0
norm_a = 0.0
norm_b = 0.0
for x, y in zip(a, b):
dot += x * y
norm_a += x * x
norm_b += y * y
if norm_a == 0.0 or norm_b == 0.0:
return 0.0
return dot / ((norm_a * norm_b) ** 0.5)
为什么纯Python：Hermes venv由uv管理，没有pip，无法安装numpy。实测1000×2048向量全扫描<100ms，完全够用。
2.4 添加Schema迁移（_migrate方法）
(1/6)
在 _migrate() 方法的迁移链末尾添加：
if current_version < 7:
v7: add message_embeddings table for vector semantic search
try:
cursor.executescript(EMBEDDING_SQL)
except sqlite3.OperationalError:
pass # Table already exists
cursor.execute("UPDATE schema_version SET version = 7")
同时在 _init_schema() 末尾的FTS5 setup之后添加：
Embeddings table setup (for vector semantic search)
try:
cursor.execute("SELECT * FROM message_embeddings LIMIT 0")
except sqlite3.OperationalError:
cursor.executescript(EMBEDDING_SQL)
2.5 添加Embedding方法到SessionDB类
在 search_messages() 方法之后、search_sessions() 方法之前添加以下4个方法：
方法1: store_embedding — 存储消息的向量
def store_embedding(self, msg_id: int, session_id: str, embedding: list, text: str) -> None:
"""Store an embedding vector for a message."""
th = _text_hash(text)
blob = _pack_embedding(embedding)
def _do(conn):
row = conn.execute(
"SELECT text_hash FROM message_embeddings WHERE msg_id = ?",
(msg_id,),
).fetchone()
existing_hash = row[0] if row else None
if existing_hash == th:
return # Already up-to-date
if row:
conn.execute(
"""UPDATE message_embeddings
SET embedding = ?, text_hash = ?, dims = ?, created_at = ?
WHERE msg_id = ?""",
(blob, th, len(embedding), time.time(), msg_id),
)
else:
conn.execute(
"""INSERT INTO message_embeddings
(msg_id, session_id, embedding, text_hash, dims, created_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(msg_id, session_id, blob, th, len(embedding), time.time()),
)
self._execute_write(_do)
方法2: search_by_vector — 向量相似度搜索
(2/6)
def search_by_vector(
self,
query_vector: list,
top_k: int = 20,
exclude_session_ids: List[str] = None,
role_filter: List[str] = None,
min_similarity: float = 0.4,
) -> List[Dict[str, Any]]\:
"""Find messages whose embeddings are most similar to query_vector.
Uses pure-Python cosine similarity.
"""
exclude_set = set(exclude_session_ids or [])
with self._lock:
if role_filter:
placeholders = ",".join("?" for _ in role_filter)
cursor = self._conn.execute(f"""
SELECT e.msg_id, e.session_id, e.embedding, m.role,
SUBSTR(m.content, 1, 200) AS content_snippet,
s.source, s.model, s.started_at AS session_started
FROM message_embeddings e
JOIN messages m ON m.id = e.msg_id
JOIN sessions s ON s.id = e.session_id
WHERE m.role IN ({placeholders})
""", role_filter)
else:
cursor = self._conn.execute("""
SELECT e.msg_id, e.session_id, e.embedding, m.role,
SUBSTR(m.content, 1, 200) AS content_snippet,
s.source, s.model, s.started_at AS session_started
FROM message_embeddings e
JOIN messages m ON m.id = e.msg_id
JOIN sessions s ON s.id = e.session_id
""")
rows = cursor.fetchall()
scored = []
for row in rows:
session_id = row[1]
if session_id in exclude_set:
continue
blob = row[2]
try:
vec = _unpack_embedding(blob)
except Exception:
continue
sim = _cosine_similarity(query_vector, vec)
if sim >= min_similarity:
scored.append((sim, row))
scored.sort(key=lambda x: x[0], reverse=True)
scored = scored[:top_k]
results = []
for sim, row in scored:
results.append({
"msg_id": row[0],
"session_id": row[1],
"role": row[3],
"content_snippet": row[4],
"similarity": round(sim, 4),
"source": row[5],
"model": row[6],
"session_started": row[7],
})
return results
方法3: get_unembedded_messages — 查找未向量化消息（用于回填）
def get_unembedded_messages(self, limit: int = 100) -> List[Dict[str, Any]]\:
"""Return messages that don't have embeddings yet (for backfill)."""
with self._lock:
cursor = self._conn.execute("""
SELECT m.id, m.session_id, m.role, m.content
FROM messages m
LEFT JOIN message_embeddings e ON e.msg_id = m.id
WHERE e.id IS NULL
AND m.content IS NOT NULL
AND LENGTH(m.content) > 10
AND m.role IN ('user', 'assistant')
ORDER BY m.timestamp DESC
LIMIT ?
""", (limit,))
return [dict(row) for row in cursor.fetchall()]
方法4: embedding_count — 统计已向量化的消息数
def embedding_count(self) -> int:
"""Count total stored embeddings."""
with self._lock:
cursor = self._conn.execute("SELECT COUNT(*) FROM message_embeddings")
return cursor.fetchone()[0]
Step 3: 修改session_search_tool.py（搜索集成层）
3.1 添加import
在文件顶部的import区域添加：
import httpx
3.2 添加3个辅助函数
在 _HIDDEN_SESSION_SOURCES 定义之后添加：
函数1: _get_memory_search_config — 读取Hermes配置中的memorySearch
(3/6)
def _get_memory_search_config() -> Optional[Dict[str, Any]]\:
"""Read memorySearch config from Hermes config file."""
try:
import yaml
from pathlib import Path
config_path = Path.home() / ".hermes" / "config.yaml"
if not config_path.exists():
return None
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
ms = cfg.get("agents", {}).get("defaults", {}).get("memorySearch")
if not ms or not isinstance(ms, dict):
return None
remote = ms.get("remote", {})
return {
"provider": ms.get("provider", "openai"),
"model": ms.get("model"),
"baseUrl": remote.get("baseUrl"),
"apiKey": remote.get("apiKey"),
}
except Exception as e:
logging.debug("Failed to read memorySearch config: %s", e)
return None
函数2: _call_embedding_api — 调用火山Embedding API
def _call_embedding_api(texts: List[str], config: Dict[str, Any]) -> Optional[List[List[float]]]\:
"""Call the Volcano/OpenAI-compatible embedding API synchronously."""
if not config.get("baseUrl") or not config.get("apiKey") or not config.get("model"):
return None
url = config["baseUrl"].rstrip("/") + "/embeddings"
headers = {
"Authorization": f"Bearer {config['apiKey']}",
"Content-Type": "application/json",
}
payload = {
"model": config["model"],
"input": texts,
}
try:
with httpx.Client(timeout=30.0) as client:
resp = client.post(url, json=payload, headers=headers)
resp.raise_for_status()
data = resp.json()
return [item["embedding"] for item in data.get("data", [])]
except Exception as e:
logging.warning("Embedding API call failed: %s", e)
return None
函数3: _vector_recall — 执行向量语义搜索
def _vector_recall(
query: str,
db,
config: Dict[str, Any],
exclude_session_ids: List[str] = None,
top_k: int = 30,
) -> List[Dict[str, Any]]\:
"""Perform vector semantic search to supplement FTS5 keyword search."""
vectors = _call_embedding_api([query], config)
if not vectors or not vectors[0]\:
return []
query_vector = vectors[0]
try:
results = db.search_by_vector(
query_vector=query_vector,
top_k=top_k,
exclude_session_ids=exclude_session_ids,
role_filter=["user", "assistant"],
min_similarity=0.4,
)
except Exception as e:
logging.warning("Vector search failed: %s", e)
return []
normalized = []
for r in results:
normalized.append({
"session_id": r["session_id"],
"role": r["role"],
"snippet": r.get("content_snippet", ""),
"rank": float(r["similarity"]),
"session_started": r.get("session_started"),
"source": r.get("source", "unknown"),
"model": r.get("model"),
"search_type": "vector",
"similarity": r["similarity"],
})
return normalized
3.3 修改session_search()主函数（合并双路召回）
找到 session_search() 函数中的FTS5搜索代码，做3处修改：
修改1: FTS5结果打标签 + 触发向量搜索
在 raw_results = db.search_messages(...) 之后添加：
(4/6)
Tag FTS5 results
for r in raw_results:
r["search_type"] = "fts5"
Vector semantic search -- supplement FTS5 with embedding similarity
embed_config = _get_memory_search_config()
vector_results = []
if embed_config:
try:
vector_results = _vector_recall(
query=query,
db=db,
config=embed_config,
exclude_session_ids=list(_HIDDEN_SESSION_SOURCES),
top_k=30,
)
logging.info(
"Vector recall returned %d results for query '%s'",
len(vector_results), query,
)
except Exception as e:
logging.warning("Vector recall failed (non-fatal): %s", e)
修改2: 合并去重
Merge FTS5 + vector results, dedup by (session_id, snippet)
seen_keys = set()
merged = []
for r in raw_results + vector_results:
key = (r.get("session_id", ""), r.get("snippet", "")[:80])
if key not in seen_keys:
seen_keys.add(key)
merged.append(r)
修改3: 后续代码使用 merged 替代 raw_results
将 if not raw_results: 改为 if not merged:，将分组循环中的 for result in raw_results: 改为 for result in merged:。
Step 4: 历史消息Embedding回填
使用以下脚本对已有消息批量生成embedding：
import sys, time, logging
sys.path.insert(0, '<hermes-agent目录>') # 替换为实际路径
from hermes_state import SessionDB
from tools.session_search_tool import _get_memory_search_config, _call_embedding_api
logging.basicConfig(level=logging.INFO)
db = SessionDB()
config = _get_memory_search_config()
unembedded = db.get_unembedded_messages(limit=1000)
print(f"需要回填的消息数量: {len(unembedded)}")
BATCH_SIZE = 10 # 火山API批量限制
total_stored = 0
errors = 0
for i in range(0, len(unembedded), BATCH_SIZE):
batch = unembedded[i:i+BATCH_SIZE]
texts = [m['content'] for m in batch]
try:
vectors = _call_embedding_api(texts, config)
if not vectors or len(vectors) != len(batch):
errors += len(batch)
continue
for j, msg in enumerate(batch):
db.store_embedding(msg['id'], msg['session_id'], vectors[j], msg['content'])
total_stored += 1
batch_num = i // BATCH_SIZE + 1
total_batches = (len(unembedded) + BATCH_SIZE - 1) // BATCH_SIZE
print(f" Batch {batch_num}/{total_batches}: {total_stored} stored")
if i + BATCH_SIZE < len(unembedded):
time.sleep(0.5) # 避免限流
except Exception as e:
print(f" Batch failed: {e}")
errors += len(batch)
重试失败的（逐条）
remaining = db.get_unembedded_messages(limit=1000)
for msg in remaining:
try:
vectors = _call_embedding_api([msg['content']], config)
if vectors and vectors[0]\:
db.store_embedding(msg['id'], msg['session_id'], vectors[0], msg['content'])
total_stored += 1
time.sleep(0.3)
except Exception:
errors += 1
print(f"完成! 成功: {total_stored}, 失败: {errors}, DB总数: {db.embedding_count()}")
回填速度参考：338条消息，34批次，耗时约46秒（含0.5s间隔）。
Step 5: 验证
import sys
sys.path.insert(0, '<hermes-agent目录>') # 替换为实际路径
from hermes_state import SessionDB
from tools.session_search_tool import _get_memory_search_config, _vector_recall
db = SessionDB()
config = _get_memory_search_config()
1. 检查embedding数量
print(f"Embedding count: {db.embedding_count()}")
2. 语义搜索测试
results = _vector_recall("磁盘清理", db, config, top_k=3)
for r in results:
print(f" sim={r['similarity']\:.4f} | {r['snippet'][:60]}")
3. 无关话题应过滤
results2 = _vector_recall("做菜食谱", db, config, top_k=3)
assert len(results2) == 0, "无关话题应该被0.4阈值过滤"
4. 完整session_search测试
from tools.session_search_tool import session_search
result = session_search(query="系统空间不够", db=db)
print(f"Search result: {result[:200]}")
🏗️ 架构图
(5/6)
用户查询 "系统空间不够了怎么办"
│
├──→ FTS5关键词搜索 ──→ 精确匹配（0结果：没提"磁盘"）
│
└──→ 向量语义搜索 ──→ 火山Embedding API
│
query vector (2048维)
│
SQLite全扫描cosine相似度
│
sim=0.58 找到"磁盘清理"session ✅
│
└──→ 合并去重 ──→ 返回语义相关结果
📊 性能数据
| 指标 | 数值 |
|---|---|
| 向量维度 | 2048 (doubao-embedding-vision) |
| 单条BLOB大小 | 8,192 bytes (2048 × 4) |
| 1000条向量DB体积 | ~8MB |
| 向量全扫描耗时 (1000条) | <100ms |
| Embedding API延迟 | ~0.3s/请求 |
| 回填速度 | ~7条/秒（批量10条） |
| 相似度阈值 | 0.4（经验值，平衡召回率和精度） |
⚠️ 注意事项与陷阱
1. Schema迁移是安全的：IF NOT EXISTS确保重复运行不出错
2. cosine相似度阈值：0.3太松（无关话题0.32也能过），0.4是经验最佳值
3. SQLite Row类型：代码中处理了 sqlite3.Row 和普通 tuple 两种返回格式
4. Embedding API限流：批量回填时每批之间sleep 0.5秒
5. 向量搜索是兜底：如果Embedding API不可用，自动降级为纯FTS5，不会报错
6. text_hash去重：相同文本不会重复调用API，文本变更才更新embedding
7. Hermes venv无pip：必须用纯Python实现，不能依赖numpy/faiss
8. BLOB编码：小端序float32（<Nf），与火山API返回的float32列表一致
🔧 故障排除
| 问题 | 原因 | 解决 |
|---|---|---|
| embedding_count()=0 | 表刚创建，还没回填 | 运行Step 4回填脚本 |
| 向量搜索0结果 | API Key无效或网络不通 | 用curl测试API连通性 |
| "做菜食谱"也返回结果 | min_similarity太低 | 调高到0.4 |
| 回填batch失败 | 个别文本过长或含特殊字符 | 逐条重试可恢复 |
| Hermes启动报错 | hermes_state.py语法错误 | `python3 -c "import py_compile; py_compile.compile('hermes_state.py', doraise=True)"` |
| schema_version未更新 | _migrate()没执行到v7分支 | 检查current_version < 7逻辑 |
📈 后续优化方向
1. 实时Embedding：在 append_message() 中自动调用Embedding API，新消息即时向量化
2. 增量回填cron：设置定时任务每小时检查未向量化的消息
3. 混合排序：FTS5和向量结果用RRF(Reciprocal Rank Fusion)融合排序
4. 缓存query vector：相同查询复用向量，减少API调用
5. sqlite-vec扩展：消息量>10万时，可替换纯Python为sqlite-vec加速
以上完整指南已保存为技能 volcano-embedding-vector-memory，其他Hermes实例加载这个技能后即可照着5步走完整个落地流程 🔥 (6/6)
干的好，如何实现实时Embedding：在 append_message() 中自动调用Embedding API，新消息即时向量化如何详细落地的，一步一步做一个详细总结，以便于其他Hermes能照着你的指南逐步和你一样把向量数据库落地，把详细指南写出来