跳到主要內容

會話存儲

Hermes Agent 使用 SQLite 數據庫(~/.hermes/state.db)來持久化會話元數據、完整消息歷史記錄以及模型配置,適用於 CLI 和網關會話。這取代了早期的每個會話使用 JSONL 文件的方法。

源文件:hermes_state.py

架構概覽

~/.hermes/state.db (SQLite, WAL mode)
├── sessions — Session metadata, token counts, billing
├── messages — Full message history per session
├── messages_fts — FTS5 virtual table for full-text search
└── schema_version — Single-row table tracking migration state

關鍵設計決策:

  • WAL 模式:支持併發讀取 + 單個寫入者(網關多平臺)
  • FTS5 虛擬表:在所有會話消息中實現快速文本搜索
  • 會話血緣關係:通過 parent_session_id 鏈(由上下文壓縮觸發的拆分)
  • 來源標籤clitelegramdiscord 等):用於平臺過濾
  • 批處理運行器和強化學習軌跡 不存儲於此(由獨立系統管理)

SQLite 模式

Sessions 表

CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
source TEXT NOT NULL,
user_id TEXT,
model TEXT,
model_config TEXT,
system_prompt TEXT,
parent_session_id TEXT,
started_at REAL NOT NULL,
ended_at REAL,
end_reason TEXT,
message_count INTEGER DEFAULT 0,
tool_call_count INTEGER DEFAULT 0,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
cache_read_tokens INTEGER DEFAULT 0,
cache_write_tokens INTEGER DEFAULT 0,
reasoning_tokens INTEGER DEFAULT 0,
billing_provider TEXT,
billing_base_url TEXT,
billing_mode TEXT,
estimated_cost_usd REAL,
actual_cost_usd REAL,
cost_status TEXT,
cost_source TEXT,
pricing_version TEXT,
title TEXT,
FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
);

CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_title_unique
ON sessions(title) WHERE title IS NOT NULL;

Messages 表

CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL REFERENCES sessions(id),
role TEXT NOT NULL,
content TEXT,
tool_call_id TEXT,
tool_calls TEXT,
tool_name TEXT,
timestamp REAL NOT NULL,
token_count INTEGER,
finish_reason TEXT,
reasoning TEXT,
reasoning_details TEXT,
codex_reasoning_items TEXT
);

CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);

備註:

  • tool_calls 以 JSON 字符串形式存儲(工具調用對象的序列化列表)
  • reasoning_detailscodex_reasoning_items 以 JSON 字符串形式存儲
  • reasoning 存儲提供方暴露的原始推理文本
  • 時間戳為 Unix 紀元浮點數(time.time()
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
content,
content=messages,
content_rowid=id
);

FTS5 表通過三個觸發器與 messages 表保持同步,這些觸發器在 messages 表執行 INSERT、UPDATE 和 DELETE 操作時觸發:

CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content)
VALUES('delete', old.id, old.content);
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content)
VALUES('delete', old.id, old.content);
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;

模式版本與遷移

當前模式版本:6

schema_version 表存儲一個整數。初始化時,_init_schema() 檢查當前版本並按順序應用遷移:

版本變更
1初始模式(sessions、messages、FTS5)
2在 messages 表中添加 finish_reason
3在 sessions 表中添加 title
4title 上添加唯一索引(允許 NULL,非 NULL 值必須唯一)
5添加計費列:cache_read_tokenscache_write_tokensreasoning_tokensbilling_providerbilling_base_urlbilling_modeestimated_cost_usdactual_cost_usdcost_statuscost_sourcepricing_version
6在 messages 表中添加推理列:reasoningreasoning_detailscodex_reasoning_items

每個遷移使用 ALTER TABLE ADD COLUMN 包裹在 try/except 中,以處理列已存在的情況(冪等性)。每次成功遷移後,版本號遞增。

寫入競爭處理

多個 hermes 進程(網關 + CLI 會話 + worktree Agent)共享一個 state.dbSessionDB 類通過以下方式處理寫入競爭:

  • 短 SQLite 超時時間(1 秒),而非默認的 30 秒
  • 應用層重試,帶有隨機抖動(20–150ms,最多 15 次重試)
  • BEGIN IMMEDIATE 事務,使鎖競爭在事務開始時即被暴露
  • 定期 WAL 檢查點:每成功寫入 50 次執行一次(PASSIVE 模式)

這避免了“車隊效應”——即 SQLite 的確定性內部退避機制導致所有競爭寫入者在同一時間間隔重試。

_WRITE_MAX_RETRIES = 15
_WRITE_RETRY_MIN_S = 0.020 # 20毫秒
_WRITE_RETRY_MAX_S = 0.150 # 150毫秒
_CHECKPOINT_EVERY_N_WRITES = 50

常見操作

初始化

from hermes_state import SessionDB

db = SessionDB() # 默認:`~/.hermes/state.db`
db = SessionDB(db_path=Path("/tmp/test.db")) # 自定義路徑

創建和管理會話

# 創建一個新的session
db.create_session(
session_id="sess_abc123",
source="cli",
model="anthropic/claude-sonnet-4.6",
user_id="user_1",
parent_session_id=None, # 或以前的 session ID 血統
)

# 結束一個session
db.end_session("sess_abc123", end_reason="user_exit")

# 重新打開一個session(清除ended_at/end_reason)
db.reopen_session("sess_abc123")

存儲消息

msg_id = db.append_message(
session_id="sess_abc123",
role="assistant",
content="Here's the answer...",
tool_calls=[{"id": "call_1", "function": {"name": "terminal", "arguments": "{}"}}],
token_count=150,
finish_reason="stop",
reasoning="Let me think about this...",
)

檢索消息

# 包含所有元數據的原始消息
messages = db.get_messages("sess_abc123")

# OpenAI 對話格式(用於 API 重放)
conversation = db.get_messages_as_conversation("sess_abc123")
# 返回:[{"role": "user", "content": "..."}, {"role": "assistant", ...}]

會話標題

# 設置標題(在非NULL標題中必須是唯一的)
db.set_session_title("sess_abc123", "Fix Docker Build")

# 按頭銜解析(返回血統中最新的)
session_id = db.resolve_session_by_title("Fix Docker Build")

# 自動生成譜系中的下一個頭銜
next_title = db.get_next_title_in_lineage("Fix Docker Build")
# 返回: "Fix Docker Build #2"

search_messages() 方法支持 FTS5 查詢語法,並對用戶輸入進行自動清理。

results = db.search_messages("docker deployment")

FTS5 查詢語法

語法示例含義
關鍵詞docker deployment兩個詞同時匹配(隱式 AND)
引號短語"exact phrase"精確短語匹配
布爾 ORdocker OR kubernetes任一詞匹配
布爾 NOTpython NOT java排除該詞
前綴匹配deploy*前綴匹配
# 只搜索 CLI sessions
results = db.search_messages("error", source_filter=["cli"])

# 排除 gateway sessions
results = db.search_messages("bug", exclude_sources=["telegram", "discord"])

# 僅搜索用戶消息
results = db.search_messages("help", role_filter=["user"])

搜索結果格式

每個結果包含:

  • idsession_idroletimestamp
  • snippet — FTS5 生成的片段,包含 >>>match<<< 標記
  • context — 匹配消息前後各一條消息(內容截斷至 200 字符)
  • sourcemodelsession_started — 來自父會話

_sanitize_fts5_query() 方法處理邊緣情況:

  • 剝離不匹配的引號和特殊字符
  • 將連字符詞用引號包裹(chat-send"chat-send"
  • 移除懸空的布爾操作符(hello ANDhello

會話血緣關係

會話可通過 parent_session_id 形成鏈式結構。這發生在網關中上下文壓縮觸發會話拆分時。

查詢:查找會話血緣關係

-- 查找 session 的所有祖先
WITH RECURSIVE lineage AS (
SELECT * FROM sessions WHERE id = ?
UNION ALL
SELECT s.* FROM sessions s
JOIN lineage l ON s.id = l.parent_session_id
)
SELECT id, title, started_at, parent_session_id FROM lineage;

-- 查找 session 的所有後代
WITH RECURSIVE descendants AS (
SELECT * FROM sessions WHERE id = ?
UNION ALL
SELECT s.* FROM sessions s
JOIN descendants d ON s.parent_session_id = d.id
)
SELECT id, title, started_at FROM descendants;

查詢:最近會話及預覽

SELECT s.*,
COALESCE(
(SELECT SUBSTR(m.content, 1, 63)
FROM messages m
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
ORDER BY m.timestamp, m.id LIMIT 1),
''
) AS preview,
COALESCE(
(SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
s.started_at
) AS last_active
FROM sessions s
ORDER BY s.started_at DESC
LIMIT 20;

查詢:令牌使用統計

-- 按 model 總計 tokens
SELECT model,
COUNT(*) as session_count,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(estimated_cost_usd) as total_cost
FROM sessions
WHERE model IS NOT NULL
GROUP BY model
ORDER BY total_cost DESC;

-- Sessions 使用率最高的 token
SELECT id, title, model, input_tokens + output_tokens AS total_tokens,
estimated_cost_usd
FROM sessions
ORDER BY total_tokens DESC
LIMIT 10;

導出與清理

# 導出帶有消息的單個 session
data = db.export_session("sess_abc123")

# 將所有 sessions(帶有消息)導出為字典列表
all_data = db.export_all(source="cli")

# 刪除舊的sessions(僅結束sessions)
deleted_count = db.prune_sessions(older_than_days=90)
deleted_count = db.prune_sessions(older_than_days=30, source="telegram")

# 清除消息但保留 session 記錄
db.clear_messages("sess_abc123")

# 刪除session和所有消息
db.delete_session("sess_abc123")

數據庫位置

默認路徑:~/.hermes/state.db

該路徑由 hermes_constants.get_hermes_home() 解析得出,默認為 ~/.hermes/,或由 HERMES_HOME 環境變量指定。

數據庫文件、WAL 文件(state.db-wal)和共享內存文件(state.db-shm)均創建在同一個目錄中。