sanguo_llmwiki/docs/02-design.md

sanguo_llmwiki 设计文档

1. 架构设计

1.1 系统架构

┌─────────────────────────────────────────────────────────────┐
│                     Claude Code                              │
│                                                               │
│  ┌─────────────────────┐    ┌─────────────────────────┐   │
│  │   Wiki Skills (11)   │    │   Wiki MCP Server       │   │
│  │   (一次性操作)        │    │   (Python 服务)          │   │
│  └─────────────────────┘    └─────────────────────────┘   │
│                                       │                     │
└───────────────────────────────────────┼─────────────────────┘
                                        │ MCP 协议
                                        │ (stdio/SSE)
┌───────────────────────────────────────▼─────────────────────┐
│                    Wiki MCP Server                           │
│                                                               │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                  MCP Protocol Layer                   │   │
│  └─────────────────────────────────────────────────────┘   │
│                          │                                  │
│  ┌───────────────────────▼───────────────────────────────┐ │
│  │                    Tool Layer                         │ │
│  │  query │ status │ lint │ linker │ taxonomy │ synthesize│ │
│  └───────────────────────┬───────────────────────────────┘ │
│                          │                                  │
│  ┌───────────────────────▼───────────────────────────────┐ │
│  │                  Service Layer                         │ │
│  │  indexer │ cache │ graph │ parser                    │ │
│  └───────────────────────┬───────────────────────────────┘ │
│                          │                                  │
│  ┌───────────────────────▼───────────────────────────────┐ │
│  │                  Storage Layer                         │ │
│  │              SQLite (WAL 模式)                         │ │
│  └─────────────────────────────────────────────────────┘   │
└───────────────────────────────────────┼─────────────────────┘
                                        │
┌───────────────────────────────────────▼─────────────────────┐
│                  Obsidian Wiki Vault                         │
│            /Volumes/KnowledgeBase/wiki-vault               │
│                                                               │
│  practices/ │ concepts/ │ entities/ │ projects/ │ skills/  │
└─────────────────────────────────────────────────────────────┘

1.2 部署架构

开发阶段：

Claude Code --stdio--> Wiki MCP Server (手动启动)

生产阶段：

PM2 --> Wiki MCP Server (SSE 模式)
  │
  └──> Claude Code --SSE--> Wiki MCP Server

---

2. 模块设计

2.1 MCP Protocol Layer

职责：

• MCP 协议解析和封装
• 工具注册和路由
• 错误处理和日志

接口：

class MCPServer:
    def register_tool(self, name: str, handler: Callable)
    def handle_call(self, name: str, params: dict) -> dict
    def log(self, level: str, message: str)

2.2 Tool Layer

工具列表：

工具	输入	输出	实现模块
wiki_query	query, tags, limit	results, citations	QueryTool
memory_bridge	tool_name, date_range	entries	MemoryTool
wiki_status	-	stats, pending	StatusTool
wiki_lint	path, level	issues, fixes	LintTool
cross_linker	path, dry_run	missing_links	LinkerTool
tag_taxonomy	path, enforce	conflicts	TaxonomyTool
wiki_synthesize	concepts, threshold	synthesis	SynthesizeTool
daily_update	-	updated, new	DailyTool

2.3 Service Layer

IndexerService（索引服务）：

class IndexerService:
    def index_page(self, path: str) -> None
    def index_batch(self, paths: List[str]) -> None
    def rebuild_index(self) -> None
    def get_dirty_pages(self) -> List[str]  # 增量更新

CacheService（缓存服务）：

class CacheService:
    def get(self, key: str) -> Optional[Any]
    def set(self, key: str, value: Any, ttl: int)
    def invalidate(self, pattern: str)

GraphService（图服务）：

class GraphService:
    def get_links(self, path: str) -> Set[str]
    def get_backlinks(self, path: str) -> Set[str]
    def find_orphans(self) -> Set[str]
    def find_missing_links(self) -> List[Tuple[str, str]]

ParserService（解析服务）：

class ParserService:
    def parse_frontmatter(self, content: str) -> dict
    def extract_links(self, content: str) -> List[str]
    def validate_page(self, path: str) -> List[str]  # 返回问题列表

2.4 Storage Layer

数据库连接（带并发保护）：

class Database:
    def __init__(self, path: str):
        self.conn = sqlite3.connect(path, check_same_thread=False)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute("PRAGMA busy_timeout=10000")  # 10s
        self.lock = asyncio.Lock()
    
    async def execute(self, sql: str, params: tuple):
        async with self.lock:
            return self.conn.execute(sql, params)

---

3. 数据模型

3.1 WikiPage（页面模型）

@dataclass
class WikiPage:
    path: str               # wiki 相对路径
    title: str              # 标题
    category: str           # 分类（practices/concepts/...）
    tags: List[str]         # 标签列表
    summary: str            # 摘要（≤200 字符）
    content_hash: str       # MD5 哈希
    lifecycle: str          # draft/verified/archived/disputed
    sources: List[str]      # 来源页面
    created_at: datetime
    updated_at: datetime
    indexed_at: datetime
    
    def is_stale(self, days: int = 90) -> bool:
        return (datetime.now() - self.updated_at).days > days

3.2 WikiIndex（索引模型）

@dataclass
class WikiIndex:
    pages: Dict[str, WikiPage]
    links: Dict[str, Set[str]]      # source -> {targets}
    backlinks: Dict[str, Set[str]]   # target -> {sources}
    tags: Dict[str, Set[str]]       # tag -> {pages}
    orphans: Set[str]               # 无反向链接的页面
    stats: IndexStats
    
@dataclass
class IndexStats:
    total_pages: int
    total_links: int
    total_tags: int
    last_indexed: datetime
    dirty_pages: int

---

4. 接口设计

4.1 MCP Tool 接口

wiki_query:

{
  "name": "wiki_query",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {"type": "string"},
      "tags": {"type": "array", "items": {"type": "string"}},
      "limit": {"type": "integer", "default": 10}
    }
  }
}

wiki_status:

{
  "name": "wiki_status",
  "inputSchema": {
    "type": "object",
    "properties": {}
  }
}

4.2 内部服务接口

# IndexerService
async def index_page(path: str) -> IndexResult
async def get_dirty_pages() -> List[str]

# QueryService
async def search(query: str, limit: int) -> List[WikiPage]
async def search_by_tags(tags: List[str]) -> List[WikiPage]

# GraphService
async def get_links(path: str) -> Set[str]
async def find_orphans() -> Set[str]

---

5. 核心算法

5.1 增量更新算法

async def incremental_update():
    # 1. 获取所有 wiki 页面
    all_pages = scan_wiki_vault()
    
    # 2. 检查每个页面的哈希
    for page in all_pages:
        current_hash = md5(page.content)
        stored = db.get_page_hash(page.path)
        
        if stored != current_hash:
            # 3. 只重索引变化的页面
            await index_page(page)
    
    # 4. 处理删除的页面
    indexed_paths = db.get_all_indexed_paths()
    for path in indexed_paths:
        if path not in all_pages:
            db.delete_page(path)

5.2 查询算法

async def wiki_query(query: str, tags: List[str], limit: int):
    # 1. FTS5 全文搜索
    if query:
        results = fts_search(query, limit)
    
    # 2. 标签过滤
    if tags:
        results = filter_by_tags(results, tags)
    
    # 3. 按相关性排序
    sorted_results = rank_by_relevance(results, query)
    
    # 4. 返回带 [[wikilink]] 的结果
    return format_results(sorted_results[:limit])

5.3 cross_linker 算法

async def find_missing_links():
    # 1. 获取所有页面内容
    pages = load_all_pages()
    
    # 2. 提取所有 [[wikilinks]]
    all_links = extract_all_links(pages)
    
    # 3. 找出缺失的链接
    missing = []
    for source, targets in all_links.items():
        for target in targets:
            if not page_exists(target):
                missing.append((source, target))
    
    return missing

---

6. 配置设计

6.1 配置文件结构

# ~/.sanguo-llmwiki/config.yaml
wiki:
  vault_path: "/Volumes/KnowledgeBase/wiki-vault"
  index_path: "~/.sanguo-llmwiki/index.db"
  max_page_size_kb: 500

mcp:
  mode: "stdio"  # stdio 或 SSE
  host: "localhost"
  port: 8080

performance:
  query_timeout_ms: 5000
  cache_ttl_seconds: 3600
  fts_cache_size_mb: 100

logging:
  level: "INFO"
  file: "~/.sanguo-llmwiki/wiki-mcp.log"

6.2 环境变量

# 环境变量优先级高于配置文件
WIKI_VAULT_PATH=/custom/path
WIKI_INDEX_PATH=/custom/index.db
MCP_MODE=sse
LOG_LEVEL=DEBUG

---

7. 错误处理

7.1 错误分类

错误类型	处理方式
Wiki 路径不存在	启动失败，返回友好错误
索引文件损坏	自动重建 + WARN 日志
SQLite 写入失败	回滚事务 + 重试 1 次
页面解析失败	记录日志 + 跳过该页面
MCP 协议错误	返回标准错误格式
查询超时	返回部分结果 + WARN

7.2 错误响应格式

{
  "success": false,
  "error": {
    "code": "INDEX_CORRUPTED",
    "message": "索引文件损坏，正在自动重建",
    "details": {"rebuilding": true}
  }
}

---

8. 安全考虑

虽然是本地系统，但仍需考虑：

1. 路径安全：验证路径在 wiki vault 范围内（防止路径遍历）
2. 资源限制：限制查询返回数量、内存使用
3. 日志脱敏：日志中不记录敏感内容

---

9. 性能优化

9.1 索引优化

• 使用 FTS5 全文搜索索引
• 标签单独建立索引
• 定期 VACUUM（每周）

9.2 查询优化

• 查询结果缓存（TTL 1 小时）
• 限制返回数量（默认 10）
• 使用 prepared statements

9.3 并发优化

• SQLite WAL 模式
• asyncio.Lock 写入串行化
• busy_timeout=10s

---

10. Wiki Skills 设计

10.1 Skill 模板

每个 Skill 遵循统一结构：

---
name: wiki-xxx
description: >
  简短描述（1-2 句）
  触发条件
---

# Wiki XXX

## 使用场景
用户何时触发这个 Skill

## 操作步骤
1. ...
2. ...

## 输出格式
...

10.2 Skills 列表

优先级 P0（核心）：

1. wiki-setup - 初始化 wiki
2. wiki-ingest - 蒸馏文档
3. wiki-capture - 保存对话

优先级 P1（重要）： 4. wiki-rebuild - 重建 wiki 5. data-ingest - 录入数据 6. ingest-url - 抓取 URL 7. wiki-agent - 录入历史 8. wiki-export - 导出

优先级 P2（可选）： 9. wiki-research - 研究 10. impl-validator - 验证 11. graph-colorize - 着色

---

11. 测试设计

11.1 单元测试

覆盖所有 Service 层的核心逻辑：

• IndexerService 测试
• QueryService 测试
• GraphService 测试
• ParserService 测试

11.2 集成测试

• MCP 协议层测试（使用 MCP SDK mock）
• SQLite 操作测试
• Wiki 解析测试

11.3 E2E 测试

使用真实 wiki 数据集测试：

• 查询场景
• 搜索场景
• 索引更新场景

---

12. 部署设计

12.1 开发部署

# 手动启动
cd ~/.openclaw/sanguo_projects/sanguo_llmwiki
python -m mcp_server.main

12.2 生产部署

# PM2 配置
cat > ecosystem.config.cjs << 'EOF'
module.exports = {
  apps: [{
    name: 'wiki-mcp',
    script: 'python',
    args: '-m mcp_server.main',
    cwd: '/Users/chufeng/.openclaw/sanguo_projects/sanguo_llmwiki',
    instances: 1,
    autorestart: true,
    watch: false,
    max_memory_restart: '500M',
    env: {
      PYTHONUNBUFFERED: '1'
    }
  }]
}
EOF

pm2 start ecosystem.config.cjs
pm2 save

12.3 MCP 配置

stdio 模式（开发）：

{
  "mcpServers": {
    "wiki": {
      "command": "python",
      "args": ["-m", "mcp_server.main"],
      "cwd": "/Users/chufeng/.openclaw/sanguo_projects/sanguo_llmwiki",
      "env": {
        "WIKI_VAULT_PATH": "/Volumes/KnowledgeBase/wiki-vault"
      }
    }
  }
}

SSE 模式（生产）：

{
  "mcpServers": {
    "wiki": {
      "type": "sse",
      "url": "http://localhost:8080/sse"
    }
  }
}

---

文档版本：v1.0 创建时间：2026-06-26