Hermes + Obsidian + LLM-wiki 三工具联动架构设计
文档版本：v1.0 | 日期：2026-06-09 | 作者：系统架构设计
1. 三工具职责划分
在构建个人或团队 AI 知识库体系时，三个工具各司其职、相互协作，形成一个完整的知识生命周期管理闭环。
| 工具 | 定位 | 核心职责 | 对外接口 |
|---|---|---|---|
| Hermes | 摄取 + 路由 Agent | 多源信息摄取、清洗、分类、提炼、路由分发 | CLI / REST API / Webhook / Claude Code Hooks |
| Obsidian | Canonical 存储层 | 人类可读的结构化存储、双链知识图谱、版本控制 | Local REST API（端口 27123）/ 本地文件系统 |
| LLM-wiki | 知识提炼引擎 | 从碎片笔记中提炼结构化 wiki 条目、建立语义关联 | 批处理命令 / 增量 Watch 模式 / HTTP API |
1.1 工具定位的哲学
三工具的分工基于以下核心原则：
- 关注点分离（Separation of Concerns）：摄取、存储、提炼三个关注点分别由独立工具负责，互不耦合。
- 人机协同（Human-in-the-Loop）：Obsidian 作为 canonical 存储，保证人类随时可以直接介入编辑，不依赖任何 AI 工具即可使用。
- 渐进式增强：三工具可以独立工作（退化模式），也可以全量联动（增强模式），新用户可以从单个工具开始，逐步接入其余组件。
1.2 数据所有权模型
原始数据所有权：用户（本地文件系统）
└── Obsidian Vault = Single Source of Truth
├── Hermes 写入 → 原始摄取产物
├── LLM-wiki 写入 → 提炼产物
└── 用户直接编辑 → 主动创作
所有工具都只是 Obsidian Vault 的读写客户端，任何工具的损坏或迁移都不会导致数据丢失。
2. 完整数据流架构
2.1 顶层数据流
graph TD A[信息源: 网页/文档/对话/代码仓库] --> B[Hermes Agent] B --> C{LLM 分类路由} C -->|原始笔记 / 待处理| D[Obsidian: inbox/] C -->|已结构化经验| E[Obsidian: experiences/] C -->|代码片段| F[Obsidian: snippets/] C -->|参考资料| G[Obsidian: references/] D --> H[LLM-wiki 提炼引擎] H -->|提炼完成| E H -->|建立双向链接| D E --> I[文件变化监听器 Watchdog] F --> I G --> I I --> J[增量向量化处理器] J --> K[向量数据库 ChromaDB] E --> L[Git 版本控制] F --> L G --> L K --> M[AI 消费: RAG 查询接口] E --> N[人类消费: Obsidian UI 直接阅读] M --> O[Claude / GPT 等 LLM 的上下文增强]
2.2 Hermes 内部数据流
graph LR subgraph 输入适配器层 U[URL Fetcher] FI[File Importer] TI[Text Input] AI[API Receiver] WH[Webhook Listener] end subgraph 预处理层 CL[HTML 清洗器] FMT[格式转换器] CHUNK[文本分块器] end subgraph LLM 分析层 CLS[分类器] SUM[摘要生成器] TAG[标签提取器] LINK[关联推断器] end subgraph 输出路由层 ROUTER[路由决策器] OBS[Obsidian Writer] VEC[Vector Indexer] end U --> CL FI --> FMT TI --> CHUNK AI --> CHUNK WH --> CHUNK CL --> CHUNK FMT --> CHUNK CHUNK --> CLS CLS --> SUM SUM --> TAG TAG --> LINK LINK --> ROUTER ROUTER --> OBS ROUTER --> VEC
2.3 LLM-wiki 提炼流水线
sequenceDiagram participant W as Watchdog participant LW as LLM-wiki participant LLM as LLM API participant OBS as Obsidian REST participant VEC as Vector DB W->>LW: 检测到 inbox/ 新文件 LW->>OBS: GET /vault/inbox/{file} OBS-->>LW: 原始笔记内容 LW->>LW: 判断是否可提炼（长度/质量阈值） LW->>LLM: 发送提炼 Prompt LLM-->>LW: 结构化 wiki 条目（Markdown） LW->>OBS: PUT /vault/experiences/{slug}.md LW->>OBS: PATCH /vault/inbox/{file}（追加 processed 标记） LW->>OBS: POST 建立双向链接 LW->>VEC: 更新向量索引 LW-->>W: 完成通知
3. Hermes 架构详解
3.1 整体架构
Hermes 是整个系统的”神经中枢”，负责将非结构化的外部信息转化为 Obsidian 可消费的 Markdown 笔记。
hermes/
├── adapters/ # 输入适配器层
│ ├── url_fetcher.py # 网页抓取（支持 JS 渲染）
│ ├── file_importer.py # 本地文件导入（PDF/Word/TXT）
│ ├── text_handler.py # 直接文本输入
│ ├── api_receiver.py # REST API 接收器
│ └── webhook_listener.py # Webhook 服务器
├── processors/ # 内容预处理层
│ ├── html_cleaner.py # HTML 清洗（去广告/导航）
│ ├── format_converter.py # 格式转换（PDF→MD / HTML→MD）
│ └── chunker.py # 智能文本分块
├── analysis/ # LLM 分析层
│ ├── classifier.py # 内容分类（主题/类型）
│ ├── summarizer.py # 摘要生成
│ ├── tagger.py # 标签提取
│ └── linker.py # 关联推断（与已有笔记）
├── routing/ # 输出路由层
│ ├── router.py # 路由决策逻辑
│ └── obsidian_writer.py # Obsidian REST API 写入
├── state/ # 状态管理
│ ├── queue.py # 摄取队列（基于 SQLite）
│ └── retry.py # 错误重试机制
└── main.py # CLI 入口
3.2 核心摄取流程代码
# hermes/core/ingestion_pipeline.py
import asyncio
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class ContentType(Enum):
RAW_NOTE = "inbox"
EXPERIENCE = "experiences"
CODE_SNIPPET = "snippets"
REFERENCE = "references"
@dataclass
class IngestRequest:
source: str # URL / 文件路径 / 原始文本
source_type: str # "url" | "file" | "text" | "api"
hint: Optional[str] = None # 用户提供的分类提示
@dataclass
class IngestResult:
content_type: ContentType
title: str
body: str # 最终 Markdown 内容
tags: list[str]
related_notes: list[str] # 推断的关联笔记路径
confidence: float # 分类置信度
class IngestionPipeline:
def __init__(self, llm_client, obsidian_client, config):
self.llm = llm_client
self.obsidian = obsidian_client
self.config = config
async def ingest(self, request: IngestRequest) -> IngestResult:
# Step 1: 获取原始内容
raw_content = await self._fetch(request)
# Step 2: 清洗和格式化
clean_content = await self._preprocess(raw_content, request.source_type)
# Step 3: LLM 分析
analysis = await self._analyze(clean_content, hint=request.hint)
# Step 4: 路由并写入 Obsidian
await self._route_and_write(analysis)
return analysis
async def _analyze(self, content: str, hint: Optional[str]) -> IngestResult:
prompt = f"""
你是一个知识管理助手。请分析以下内容并：
1. 判断内容类型（raw_note/experience/snippet/reference）
2. 生成一个简洁的标题
3. 将内容转化为标准 Markdown 格式
4. 提取 3-5 个标签
5. 识别可能的关联主题
{'用户提示: ' + hint if hint else ''}
内容：
{content[:4000]}
请以 JSON 格式返回结果。
"""
response = await self.llm.complete(prompt)
return self._parse_analysis(response)
async def _route_and_write(self, result: IngestResult):
# 根据内容类型决定写入路径
path_map = {
ContentType.RAW_NOTE: "inbox",
ContentType.EXPERIENCE: "experiences",
ContentType.CODE_SNIPPET: "snippets",
ContentType.REFERENCE: "references",
}
folder = path_map[result.content_type]
filename = self._slugify(result.title)
await self.obsidian.create_note(
path=f"{folder}/{filename}.md",
content=self._render_note(result)
)
def _render_note(self, result: IngestResult) -> str:
tags_str = "\n".join(f" - {t}" for t in result.tags)
links_str = "\n".join(f"- [[{n}]]" for n in result.related_notes)
return f"""---
tags:
{tags_str}
created: {self._now_iso()}
source_type: hermes_ingestion
---
# {result.title}
{result.body}
## 相关笔记
{links_str}
"""3.3 状态管理与错误重试
# hermes/state/queue.py
import sqlite3
import json
from datetime import datetime
from enum import Enum
class JobStatus(Enum):
PENDING = "pending"
PROCESSING = "processing"
DONE = "done"
FAILED = "failed"
class IngestionQueue:
"""基于 SQLite 的持久化摄取队列，支持崩溃恢复"""
def __init__(self, db_path: str = "~/.hermes/queue.db"):
self.conn = sqlite3.connect(db_path)
self._init_schema()
def _init_schema(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
source_type TEXT NOT NULL,
hint TEXT,
status TEXT DEFAULT 'pending',
retry_count INTEGER DEFAULT 0,
error_msg TEXT,
created_at TEXT,
updated_at TEXT
)
""")
self.conn.commit()
def enqueue(self, source: str, source_type: str, hint: str = None) -> int:
now = datetime.utcnow().isoformat()
cursor = self.conn.execute(
"INSERT INTO jobs (source, source_type, hint, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
(source, source_type, hint, now, now)
)
self.conn.commit()
return cursor.lastrowid
def mark_failed(self, job_id: int, error: str):
self.conn.execute(
"UPDATE jobs SET status='failed', error_msg=?, retry_count=retry_count+1 WHERE id=?",
(error, job_id)
)
self.conn.commit()
def get_retryable(self, max_retries: int = 3):
"""获取可重试的失败任务（retry_count < max_retries）"""
return self.conn.execute(
"SELECT * FROM jobs WHERE status='failed' AND retry_count < ?",
(max_retries,)
).fetchall()4. Obsidian Local REST API 集成
4.1 API 端点全览
Obsidian Local REST API 插件默认监听 https://127.0.0.1:27123，提供以下核心端点：
| 端点 | 方法 | 功能 |
|---|---|---|
/vault/{path} | GET | 读取文件内容 |
/vault/{path} | PUT | 创建或覆盖文件 |
/vault/{path} | PATCH | 追加内容到文件 |
/vault/{path} | DELETE | 删除文件 |
/vault/ | GET | 列出目录内容 |
/search/simple/?query= | GET | 全文搜索 |
/active/ | GET | 获取当前活跃文件 |
/commands/ | POST | 执行 Obsidian 命令 |
4.2 认证方式
Obsidian Local REST API 使用 Bearer Token 认证，token 在插件设置中生成：
Authorization: Bearer <your-api-key>
注意：默认使用自签名 TLS 证书，需要在客户端禁用证书验证（verify=False）或导入证书。
4.3 完整 Python 客户端封装
# obsidian_client/client.py
import httpx
import json
from pathlib import Path
from typing import Optional, Union
from dataclasses import dataclass
@dataclass
class ObsidianNote:
path: str
content: str
frontmatter: dict = None
class ObsidianClient:
"""
Obsidian Local REST API 的 Python 客户端封装。
支持同步和异步两种使用方式。
"""
def __init__(
self,
base_url: str = "https://127.0.0.1:27123",
api_key: str = None,
verify_ssl: bool = False,
):
self.base_url = base_url.rstrip("/")
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "text/markdown",
}
self.verify_ssl = verify_ssl
self._client = httpx.Client(
headers=self.headers,
verify=self.verify_ssl,
timeout=30.0
)
# ── 文件操作 ──────────────────────────────────────────────────────────
def read_note(self, path: str) -> Optional[str]:
"""读取笔记内容，path 为相对于 vault 根目录的路径"""
resp = self._client.get(f"{self.base_url}/vault/{path}")
if resp.status_code == 404:
return None
resp.raise_for_status()
return resp.text
def create_note(self, path: str, content: str) -> bool:
"""创建或覆盖笔记（PUT 语义）"""
resp = self._client.put(
f"{self.base_url}/vault/{path}",
content=content.encode("utf-8")
)
resp.raise_for_status()
return resp.status_code in (200, 204)
def append_to_note(self, path: str, content: str) -> bool:
"""追加内容到笔记末尾（PATCH 语义）"""
resp = self._client.patch(
f"{self.base_url}/vault/{path}",
content=content.encode("utf-8"),
headers={**self.headers, "Content-Type": "text/markdown"}
)
resp.raise_for_status()
return resp.status_code in (200, 204)
def delete_note(self, path: str) -> bool:
"""删除笔记"""
resp = self._client.delete(f"{self.base_url}/vault/{path}")
return resp.status_code in (200, 204, 404)
def list_directory(self, path: str = "") -> list[str]:
"""列出目录中的文件"""
url = f"{self.base_url}/vault/{path}/" if path else f"{self.base_url}/vault/"
resp = self._client.get(url)
resp.raise_for_status()
return resp.json().get("files", [])
# ── 搜索 ──────────────────────────────────────────────────────────────
def search(self, query: str, context_length: int = 200) -> list[dict]:
"""全文搜索，返回匹配的笔记路径和上下文摘要"""
resp = self._client.get(
f"{self.base_url}/search/simple/",
params={"query": query, "contextLength": context_length}
)
resp.raise_for_status()
return resp.json()
# ── 辅助方法 ──────────────────────────────────────────────────────────
def ensure_directory_exists(self, folder_path: str):
"""通过创建占位文件来确保目录存在（Obsidian 没有独立的目录创建 API）"""
placeholder = f"{folder_path}/.gitkeep"
if not self.read_note(placeholder):
self.create_note(placeholder, "")
def upsert_note(self, path: str, content: str) -> str:
"""
幂等的创建/更新操作。
返回 "created" 或 "updated"。
"""
existing = self.read_note(path)
self.create_note(path, content)
return "updated" if existing else "created"
def add_backlink(self, source_path: str, target_path: str):
"""在 source 笔记末尾追加指向 target 的反向链接"""
target_name = Path(target_path).stem # 去掉 .md 后缀
link_section = f"\n\n## 反向链接\n- [[{target_name}]]\n"
# 检查是否已存在该链接（避免重复）
existing = self.read_note(source_path) or ""
if f"[[{target_name}]]" not in existing:
self.append_to_note(source_path, link_section)
# ── 使用示例 ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
client = ObsidianClient(
base_url="https://127.0.0.1:27123",
api_key="your-api-key-here"
)
# 创建一篇新笔记
note_content = """---
tags:
- python
- async
created: 2026-06-09
---
# Python 异步编程要点
asyncio 的核心模型是单线程事件循环，通过协程实现并发。
## 关键概念
- `async def` 定义协程函数
- `await` 挂起当前协程，让出控制权
- `asyncio.gather()` 并发运行多个协程
"""
result = client.upsert_note("experiences/python-async-notes.md", note_content)
print(f"笔记操作结果: {result}")
# 搜索
results = client.search("asyncio 事件循环")
for r in results:
print(f"找到: {r['filename']} — {r['context'][:100]}")5. LLM-wiki 提炼流程
5.1 提炼触发策略
# llm_wiki/refiner.py
import os
import re
import asyncio
from pathlib import Path
from datetime import datetime
from obsidian_client import ObsidianClient
# 可提炼的判断条件
MIN_WORD_COUNT = 100 # 最少 100 词才值得提炼
MAX_AGE_DAYS = 30 # inbox 中超过 30 天的笔记强制处理
PROCESSED_MARKER = "<!-- llm-wiki:processed -->"
class WikiRefiner:
def __init__(self, obsidian: ObsidianClient, llm_client):
self.obsidian = obsidian
self.llm = llm_client
async def run_refine_cycle(self):
"""扫描 inbox，批量提炼"""
inbox_files = self.obsidian.list_directory("inbox")
md_files = [f for f in inbox_files if f.endswith(".md") and f != ".gitkeep"]
print(f"[llm-wiki] 发现 {len(md_files)} 个待处理文件")
tasks = []
for filename in md_files:
path = f"inbox/{filename}"
content = self.obsidian.read_note(path)
if content and self._should_refine(content):
tasks.append(self._refine_single(path, content))
# 并发提炼（最多 3 个并发，避免 API 限流）
semaphore = asyncio.Semaphore(3)
async def bounded_refine(coro):
async with semaphore:
return await coro
results = await asyncio.gather(
*[bounded_refine(t) for t in tasks],
return_exceptions=True
)
success = sum(1 for r in results if not isinstance(r, Exception))
print(f"[llm-wiki] 提炼完成: {success}/{len(tasks)} 成功")
def _should_refine(self, content: str) -> bool:
"""判断内容是否值得提炼"""
if PROCESSED_MARKER in content:
return False
word_count = len(content.split())
return word_count >= MIN_WORD_COUNT
async def _refine_single(self, inbox_path: str, raw_content: str):
"""提炼单篇笔记"""
print(f"[llm-wiki] 提炼: {inbox_path}")
# Step 1: 调用 LLM 生成结构化 wiki 条目
wiki_entry = await self._call_llm_refine(raw_content)
# Step 2: 生成目标路径（基于 LLM 返回的 slug）
slug = wiki_entry.get("slug") or self._auto_slug(wiki_entry.get("title", "untitled"))
target_path = f"experiences/{slug}.md"
# Step 3: 写入 experiences/
final_content = self._render_wiki_entry(wiki_entry)
action = self.obsidian.upsert_note(target_path, final_content)
print(f"[llm-wiki] → {target_path} ({action})")
# Step 4: 在 inbox 原文末尾追加处理标记和反向链接
marker = f"\n\n{PROCESSED_MARKER}\n提炼时间: {datetime.now().isoformat()}\n提炼结果: [[{slug}]]\n"
self.obsidian.append_to_note(inbox_path, marker)
# Step 5: 建立双向链接
self.obsidian.add_backlink(target_path, inbox_path)
# Step 6: 触发向量重新索引（发送事件）
await self._notify_reindex(target_path)
async def _call_llm_refine(self, raw_content: str) -> dict:
"""调用 LLM 将原始笔记提炼为结构化 wiki 条目"""
system_prompt = """你是一个知识工程师，擅长将碎片化笔记整理为结构清晰的 wiki 条目。
请将用户提供的笔记内容整理为标准 wiki 格式，返回 JSON：
{
"title": "简洁准确的标题",
"slug": "kebab-case-filename",
"summary": "一句话摘要（< 100 字）",
"tags": ["tag1", "tag2"],
"body": "完整的 Markdown 内容，包含层级标题、代码块、要点列表",
"related_topics": ["相关主题1", "相关主题2"]
}"""
response = await self.llm.complete(
system=system_prompt,
user=raw_content[:6000]
)
return json.loads(response)
def _render_wiki_entry(self, entry: dict) -> str:
tags = "\n".join(f" - {t}" for t in entry.get("tags", []))
related = "\n".join(f"- [[{r}]]" for r in entry.get("related_topics", []))
return f"""---
title: {entry['title']}
tags:
{tags}
summary: {entry.get('summary', '')}
created: {datetime.now().strftime('%Y-%m-%d')}
source: llm-wiki-refiner
---
# {entry['title']}
> {entry.get('summary', '')}
{entry['body']}
## 相关主题
{related}
"""6. 增量索引机制
6.1 文件变化监听
# vector_indexer/watcher.py
import asyncio
import hashlib
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileModifiedEvent, FileCreatedEvent, FileDeletedEvent
class VaultChangeHandler(FileSystemEventHandler):
"""监听 Obsidian Vault 的文件变化，触发增量向量索引"""
IGNORED_FOLDERS = {".obsidian", ".git", ".trash"}
WATCHED_FOLDERS = {"experiences", "snippets", "references"} # 不监听 inbox（未提炼）
def __init__(self, indexer):
self.indexer = indexer
self._pending_events = asyncio.Queue()
def on_modified(self, event):
if not event.is_directory and self._should_index(event.src_path):
asyncio.run_coroutine_threadsafe(
self._pending_events.put(("upsert", event.src_path)),
asyncio.get_event_loop()
)
def on_created(self, event):
if not event.is_directory and self._should_index(event.src_path):
asyncio.run_coroutine_threadsafe(
self._pending_events.put(("upsert", event.src_path)),
asyncio.get_event_loop()
)
def on_deleted(self, event):
if not event.is_directory and self._should_index(event.src_path):
asyncio.run_coroutine_threadsafe(
self._pending_events.put(("delete", event.src_path)),
asyncio.get_event_loop()
)
def on_moved(self, event):
# 重命名 = 删除旧 + 创建新
if self._should_index(event.src_path):
asyncio.run_coroutine_threadsafe(
self._pending_events.put(("delete", event.src_path)),
asyncio.get_event_loop()
)
if self._should_index(event.dest_path):
asyncio.run_coroutine_threadsafe(
self._pending_events.put(("upsert", event.dest_path)),
asyncio.get_event_loop()
)
def _should_index(self, path: str) -> bool:
p = Path(path)
if not p.suffix == ".md":
return False
for part in p.parts:
if part in self.IGNORED_FOLDERS:
return False
# 只索引 WATCHED_FOLDERS 中的内容
for folder in self.WATCHED_FOLDERS:
if folder in p.parts:
return True
return False
async def process_events(self):
"""事件处理循环，带防抖（1秒内同一文件的多个事件合并）"""
pending = {} # path → (action, timestamp)
while True:
try:
action, path = await asyncio.wait_for(
self._pending_events.get(), timeout=1.0
)
pending[path] = action
except asyncio.TimeoutError:
# 1秒无新事件，处理积压
if pending:
for path, action in pending.items():
await self.indexer.handle_change(action, path)
pending.clear()6.2 向量数据库增量更新
# vector_indexer/indexer.py
import chromadb
from chromadb.utils import embedding_functions
from pathlib import Path
import hashlib
class IncrementalVectorIndexer:
"""
基于 ChromaDB 的增量向量索引器。
只重新索引变更的文件，通过内容哈希判断是否真正变化。
"""
def __init__(self, chroma_url: str, collection_name: str = "obsidian_vault"):
self.client = chromadb.HttpClient(host=chroma_url)
self.ef = embedding_functions.DefaultEmbeddingFunction()
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.ef,
metadata={"hnsw:space": "cosine"}
)
async def handle_change(self, action: str, file_path: str):
doc_id = self._path_to_id(file_path)
if action == "delete":
self._delete_document(doc_id)
elif action == "upsert":
await self._upsert_document(file_path, doc_id)
async def _upsert_document(self, file_path: str, doc_id: str):
content = Path(file_path).read_text(encoding="utf-8")
content_hash = hashlib.md5(content.encode()).hexdigest()
# 检查哈希是否变化（避免无意义的重新嵌入，节省 API 费用）
existing = self.collection.get(ids=[doc_id], include=["metadatas"])
if existing["ids"] and existing["metadatas"][0].get("hash") == content_hash:
return # 内容未变，跳过
# 按段落分块（每块最多 500 词）
chunks = self._chunk_document(content)
chunk_ids = [f"{doc_id}#{i}" for i in range(len(chunks))]
metadatas = [
{"source": file_path, "chunk_index": i, "hash": content_hash}
for i in range(len(chunks))
]
# 先删除旧的所有 chunk
self._delete_document(doc_id)
# 批量 upsert 新 chunk
self.collection.upsert(
ids=chunk_ids,
documents=chunks,
metadatas=metadatas
)
print(f"[indexer] 已索引: {Path(file_path).name} ({len(chunks)} chunks)")
def _delete_document(self, doc_id: str):
"""删除文档的所有 chunk（通过前缀匹配）"""
existing = self.collection.get(
where={"source": {"$like": f"%{doc_id}%"}}
)
if existing["ids"]:
self.collection.delete(ids=existing["ids"])
def _chunk_document(self, content: str, max_words: int = 500) -> list[str]:
paragraphs = content.split("\n\n")
chunks, current_chunk, current_words = [], [], 0
for para in paragraphs:
words = len(para.split())
if current_words + words > max_words and current_chunk:
chunks.append("\n\n".join(current_chunk))
current_chunk, current_words = [para], words
else:
current_chunk.append(para)
current_words += words
if current_chunk:
chunks.append("\n\n".join(current_chunk))
return chunks
def _path_to_id(self, path: str) -> str:
return Path(path).stem.replace(" ", "_")7. 三工具的配置和部署
7.1 最小可用部署（单机 Docker）
# docker-compose.yml
version: "3.9"
services:
hermes:
image: ghcr.io/yourorg/hermes:latest
restart: unless-stopped
volumes:
- ${VAULT_PATH}:/vault:rw # Obsidian Vault 挂载
- ./hermes_data:/data # 队列数据库
environment:
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
- OBSIDIAN_API_URL=http://host.docker.internal:27123
- OBSIDIAN_API_KEY=${OBSIDIAN_API_KEY}
- HERMES_QUEUE_PATH=/data/queue.db
- CHROMA_URL=http://chroma:8000
- LOG_LEVEL=INFO
ports:
- "8765:8765" # Webhook 监听端口
extra_hosts:
- "host.docker.internal:host-gateway"
llm-wiki:
image: ghcr.io/yourorg/llm-wiki:latest
restart: unless-stopped
volumes:
- ${VAULT_PATH}:/vault:rw
environment:
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
- OBSIDIAN_API_URL=http://host.docker.internal:27123
- OBSIDIAN_API_KEY=${OBSIDIAN_API_KEY}
- CHROMA_URL=http://chroma:8000
- REFINE_INTERVAL=3600 # 每小时执行一次提炼
- WATCH_MODE=true # 同时监听文件变化
extra_hosts:
- "host.docker.internal:host-gateway"
chroma:
image: chromadb/chroma:0.5.0
restart: unless-stopped
ports:
- "8000:8000"
volumes:
- ./chroma_data:/chroma/chroma
environment:
- CHROMA_SERVER_AUTH_PROVIDER=chromadb.auth.token.TokenAuthServerProvider
- CHROMA_SERVER_AUTH_CREDENTIALS=${CHROMA_TOKEN}
- CHROMA_SERVER_AUTH_TOKEN_TRANSPORT_HEADER=X-Chroma-Token
# 可选: 向量检索 API（供 Claude/GPT 等 LLM 调用）
rag-api:
image: ghcr.io/yourorg/rag-api:latest
restart: unless-stopped
ports:
- "8001:8001"
environment:
- CHROMA_URL=http://chroma:8000
- CHROMA_TOKEN=${CHROMA_TOKEN}
- API_KEY=${RAG_API_KEY}
depends_on:
- chroma# .env
VAULT_PATH=/home/user/obsidian-vault
ANTHROPIC_API_KEY=sk-ant-...
OBSIDIAN_API_KEY=your-obsidian-key
CHROMA_TOKEN=your-chroma-token
RAG_API_KEY=your-rag-api-key7.2 Vault 目录结构规范
obsidian-vault/
├── inbox/ # Hermes 写入的原始摄取内容（待处理）
│ ├── 2026-06-09-python-async.md
│ └── 2026-06-09-docker-tips.md
├── experiences/ # LLM-wiki 提炼的结构化知识（最终形态）
│ ├── python-async-programming.md
│ └── docker-best-practices.md
├── snippets/ # 代码片段
│ └── python-retry-decorator.md
├── references/ # 参考资料（技术文档/论文摘要）
│ └── obsidian-rest-api-reference.md
├── projects/ # 项目专属笔记（手动维护）
└── journal/ # 日记/会议记录（不参与向量索引）
8. 触发机制设计
8.1 四种触发方式
graph LR subgraph 触发方式 M[手动 CLI] FW[文件监听 Watchdog] CR[定时 Cron] WH[Webhook / HTTP] CCH[Claude Code Hooks] end subgraph 处理目标 HI[Hermes 摄取] LR[LLM-wiki 提炼] VI[向量重建索引] end M --> HI M --> LR FW --> VI FW --> LR CR --> LR CR --> VI WH --> HI CCH --> HI
8.2 CLI 接口设计
# 摄取单个 URL
hermes ingest --url "https://example.com/article" --hint "Python 性能优化"
# 摄取本地文件
hermes ingest --file ./notes.txt
# 批量摄取（stdin 或文件列表）
cat urls.txt | hermes ingest --batch
# 手动触发 LLM-wiki 提炼循环
llm-wiki refine --dry-run # 预览模式，不实际写入
llm-wiki refine # 执行提炼
# 重建全量向量索引
llm-wiki reindex --folder experiences
# 查询 RAG（测试用）
llm-wiki query "Python asyncio 的并发模型是什么"8.3 Claude Code Hooks 集成
在 .claude/settings.json 中配置 Hook，实现 Claude Code 会话结束后自动归档：
{
"hooks": {
"Stop": [
{
"matcher": "",
"hooks": [
{
"type": "command",
"command": "hermes ingest --text \"$CLAUDE_SESSION_SUMMARY\" --hint \"claude-session\" 2>/dev/null || true"
}
]
}
]
}
}8.4 定时任务配置
# crontab -e
# 每天凌晨 2 点执行 LLM-wiki 提炼（低峰期避免影响使用）
0 2 * * * docker exec llm-wiki llm-wiki refine >> /var/log/llm-wiki-cron.log 2>&1
# 每周日重建全量向量索引（保证索引完整性）
0 3 * * 0 docker exec llm-wiki llm-wiki reindex --folder experiences >> /var/log/reindex-cron.log 2>&19. 可观测性
9.1 结构化日志格式
所有三个工具统一采用 JSON 结构化日志，方便聚合和查询：
{
"timestamp": "2026-06-09T02:15:30.123Z",
"level": "INFO",
"service": "hermes",
"event": "ingest_completed",
"source_url": "https://example.com/article",
"content_type": "experience",
"target_path": "experiences/python-perf-notes.md",
"duration_ms": 2341,
"token_usage": {
"input": 1200,
"output": 800
}
}{
"timestamp": "2026-06-09T02:16:01.456Z",
"level": "WARNING",
"service": "llm-wiki",
"event": "refine_skipped",
"reason": "below_word_threshold",
"file": "inbox/short-note.md",
"word_count": 42,
"threshold": 100
}9.2 关键指标监控
| 指标名 | 类型 | 说明 | 告警阈值 |
|---|---|---|---|
hermes_ingest_total | Counter | 总摄取数量 | — |
hermes_ingest_errors_total | Counter | 摄取错误数 | 错误率 > 10% |
hermes_queue_depth | Gauge | 当前队列积压深度 | > 50 |
llm_wiki_refine_duration_seconds | Histogram | 单次提炼耗时 | P99 > 30s |
vector_index_documents_total | Gauge | 向量库文档总数 | — |
llm_token_cost_usd_total | Counter | 累计 LLM 费用 | 日费用 > $5 |
obsidian_api_errors_total | Counter | Obsidian API 错误数 | > 0（即告警） |
9.3 Prometheus + Grafana 部署
# docker-compose.yml 追加
prometheus:
image: prom/prometheus:v2.51.0
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
grafana:
image: grafana/grafana:10.3.0
ports:
- "3000:3000"
volumes:
- ./grafana_data:/var/lib/grafana# prometheus.yml
global:
scrape_interval: 30s
scrape_configs:
- job_name: hermes
static_configs:
- targets: ["hermes:8766"] # /metrics 端点
- job_name: llm-wiki
static_configs:
- targets: ["llm-wiki:8767"]10. 替代方案对比
| 方案 | 优点 | 缺点 | 适合谁 |
|---|---|---|---|
| Hermes + Obsidian + LLM-wiki（本方案） | 完全本地、高度可定制、数据自主、可编程集成 | 需要自行搭建维护，初始配置成本高 | 有一定工程能力的个人开发者、研究员 |
| Notion + AI（Notion AI / 第三方插件） | 开箱即用、界面美观、多人协作 | 数据存储在 Notion 云端、API 费用高、定制能力有限 | 非技术用户、小团队协作 |
| AnythingLLM | 界面友好、支持多种 LLM 后端、文档上传简单 | 知识组织能力弱（无双链）、定制能力有限、本地部署资源占用大 | 希望有 GUI 的小团队 |
| Khoj | 开源完整、支持 Obsidian 插件、跨平台搜索 | 配置复杂、文档不完善、社区较小 | 有 Linux 运维经验的高级用户 |
| Mem.ai | AI 自动组织强、搜索体验好 | 完全云端、价格较贵（$14.99/月）、无法自托管 | 愿意付费、不在意数据隐私的用户 |
| Logseq + LLM 插件 | 大纲式思维友好、本地优先 | LLM 集成生态不如 Obsidian 成熟 | 大纲式笔记爱好者 |
10.1 选型决策树
你有编程能力，且希望数据完全本地？
└── 是 → 本方案（Hermes + Obsidian + LLM-wiki）
└── 否 →
你愿意为托管服务付费？
└── 是 → Mem.ai（个人）或 Notion AI（团队）
└── 否 →
你接受 GUI 配置复杂度？
└── 是 → Khoj（功能最完整的开源方案）
└── 否 → AnythingLLM（最容易上手的本地方案）
附录：快速启动指南
A. 环境准备清单
# 1. 安装 Obsidian 并启用 Local REST API 插件
# 插件市场搜索: "Local REST API"，安装后记录 API Key
# 2. 克隆项目
git clone https://github.com/yourorg/knowledge-stack.git
cd knowledge-stack
# 3. 复制并填写环境变量
cp .env.example .env
vim .env
# 4. 启动所有服务
docker compose up -d
# 5. 验证服务状态
docker compose ps
curl http://localhost:8001/health # RAG API 健康检查
# 6. 测试摄取
docker exec hermes hermes ingest \
--url "https://docs.python.org/3/library/asyncio.html" \
--hint "Python 异步编程"
# 7. 检查 Obsidian Vault 中是否出现新笔记
ls ~/obsidian-vault/inbox/B. 故障排查速查
| 症状 | 可能原因 | 解决方法 |
|---|---|---|
| Obsidian API 返回 401 | API Key 错误或 Obsidian 未打开 | 检查 .env 中的 OBSIDIAN_API_KEY；确保 Obsidian 正在运行 |
| LLM 调用超时 | 网络问题或 API Key 无效 | 检查 ANTHROPIC_API_KEY；测试 curl api.anthropic.com 连通性 |
| 向量索引未更新 | Watchdog 未监听到变化 | 检查 Vault 挂载路径是否正确；重启 llm-wiki 服务 |
| ChromaDB 连接拒绝 | Chroma 服务未启动 | docker compose up -d chroma |
| 摄取队列积压 | LLM API 限流或网络慢 | 调低 MAX_CONCURRENT_INGESTS 环境变量 |