# md-vector-db 代码架构审计报告 **审计日期**: 2026-07-06 **审计范围**: `src/` 全部模块 + `scripts/ingest_obsidian.py` + 测试 **审计方法**: 逐文件静态分析,按安全性 → 可靠性 → 设计 → 测试四个维度评审 --- ## 总览 | 维度 | 评分 | 说明 | |------|------|------| | 安全性 | ⭐⭐⭐⭐ | 整体良好,认证/限流/路径防护到位 | | 可靠性 | ⭐⭐⭐ | 有 2 处吞异常 + 1 处内存泄漏风险 | | 设计 | ⭐⭐⭐ | 核心架构清晰,但 2 处封装泄漏 | | 测试 | ⭐⭐⭐ | 46 个测试覆盖主要路径,缺回归测试 | 共发现 **11 个问题**:3 个高、4 个中、4 个低。 --- ## 高优先级问题 ### H1. `config.py:75` — `AppConfig.__init__` 破坏 dataclass 构造 ```python # 问题代码 (config.py:66-79) @dataclass class AppConfig: chroma: ChromaConfig = field(default_factory=ChromaConfig) embed: EmbedConfig = field(default_factory=EmbedConfig) chunk: ChunkConfig = field(default_factory=ChunkConfig) server: ServerConfig = field(default_factory=ServerConfig) def __init__(self, **kwargs): # ← 覆盖了 dataclass 生成的 __init__ self.chroma = ChromaConfig(**kwargs.get("chroma", {})) self.embed = EmbedConfig(**kwargs.get("embed", {})) ... ``` **影响**: - `AppConfig(chroma=ChromaConfig())` — 正常 dataclass 构造方式会报 `TypeError` - `@dataclass` 的 `field(default_factory=...)` 全部被绕过 - 所有字段失去类型检查 **修复**: ```python @dataclass class AppConfig: chroma: ChromaConfig = field(default_factory=ChromaConfig) embed: EmbedConfig = field(default_factory=EmbedConfig) chunk: ChunkConfig = field(default_factory=ChunkConfig) server: ServerConfig = field(default_factory=ServerConfig) @classmethod def from_dict(cls, data: dict) -> "AppConfig": return cls( chroma=ChromaConfig(**data.get("chroma", {})), embed=EmbedConfig(**data.get("embed", {})), chunk=ChunkConfig(**data.get("chunk", {})), server=ServerConfig(**data.get("server", {})), ) def load_config(path: str | None = None) -> AppConfig: ... return AppConfig.from_dict(data) ``` **严重程度**: HIGH — 调用方如果用标准 dataclass 构造会直接崩溃。 --- ### H2. `deps.py:47-58` — `list_collections_with_stats` 只列出内存中的集合 ```python # 问题代码 def list_collections_with_stats(self) -> list[dict]: all_names = set(self._searchers.keys()) | set(self._ingestors.keys()) all_names.add(self.default_collection) # ↑ 只包含被访问过的集合!ChromaDB 中已有的其他集合不会出现 ``` **影响**:通过 CLI 或直接操作 ChromaDB 创建的集合,在 HTTP API 的 `/api/v1/collections` 中看不到。 **修复**:应从 ChromaDB 直接获取。 ```python def list_collections_with_stats(self) -> list[dict]: result = [] for coll in self.db.client.list_collections(): result.append({"name": coll.name, "count": coll.count()}) return result ``` **严重程度**: HIGH — 功能缺陷,`/api/v1/collections` 返回不完整数据。 --- ### H3. `auth.py:27` — RateLimiter 内存无限增长 ```python # 问题代码 class RateLimiter: def __init__(self, max_requests=30, window_seconds=60): self._store: dict[str, list[float]] = defaultdict(list) def is_allowed(self, client_id: str) -> bool: with self._lock: records = self._store[client_id] records[:] = [t for t in records if now - t < self.window] # ↑ 只清理时间戳,不清理空 key ``` **影响**:每个新 IP 都在 `_store` 中永久保留一个 key,长期运行后字典膨胀。 **修复**:在清理过期时间戳后,删除空的 key: ```python records[:] = [t for t in records if now - t < self.window] if not records: del self._store[client_id] ``` **严重程度**: HIGH — 长期运行会内存泄漏。 --- ## 中优先级问题 ### M1. `search.py:87-88` — `delete_by_source` 吞掉所有异常 ```python def delete_by_source(self, file_name: str) -> bool: try: with self.db.write_lock: ... except Exception: pass # ← 隐藏了 ChromaDB 损坏、磁盘满等严重错误 return False ``` **修复**:至少记录日志: ```python except Exception: logger.exception("删除文档失败: %s", file_name) return False ``` --- ### M2. `db.py` — 缺少 `list_collections()` 封装 `VectorDB` 没有暴露 `list_collections()`,导致: - `deps.py` 无法从 `db` 对象获取集合列表 - MCP server 需要绕过封装直接访问 `db.client` **修复**: ```python def list_collections(self) -> list: return self.client.list_collections() ``` --- ### M3. `ingest.py:237-245` — `_remove_by_source` 吞掉去重异常 ```python def _remove_by_source(self, file_name: str) -> None: try: ... except Exception: pass # collection 为空时 get 可能抛异常 ``` **影响**:如果去重失败(非空 collection),重复入库不会报错但会静默产生重复数据。 **修复**:区分 "collection 为空" 和真正的错误: ```python except ValueError: pass # collection 无数据时 ChromaDB 会抛 ValueError except Exception: logger.exception("去重检查失败: %s", file_name) ``` --- ### M4. `cli/main.py:47-48` — 硬编码回退值 ```python def _get_default_collection() -> str: return _cfg.chroma.collection_name if _cfg else "markdown_docs" ``` 与 `deps.py:27-29` 中的 `default_collection` 逻辑不一致(deps.py 支持 `MD_VECTOR_DB_COLLECTION` 环境变量)。 --- ## 低优先级问题 ### L1. `config.py:16` — 相对路径 `DEFAULT_CONFIG_PATH` ```python DEFAULT_CONFIG_PATH = "config.yaml" # 从不同 CWD 运行会找不到 ``` **建议**:在 `load_config` 中基于项目根目录解析。 --- ### L2. `cli/main.py:97-101` — 函数内重复 import ```python for fp in file_paths: from pathlib import Path as _Path # 每次循环都 import ... import glob as _glob ``` **修复**:移到文件顶部。 --- ### L3. 缺少死循环回归测试 `_split_single_paragraph` 修复(`start = max(start + 1, next_start)`)没有对应的单元测试:分隔符距 chunk 起点 < overlap 的场景。 ```python def test_separator_near_start_does_not_loop(self, splitter): """分隔符紧挨 start 时不会死循环 (回归测试).""" # 构造: 前 overlap 字符内就有分隔符的情况 text = "。" * 50 + "内容" * 300 # 分隔符密集在开头 chunks = splitter.split(f"# T\n{text}", source_file="test.md") assert len(chunks) > 0 # 不卡死即通过 ``` --- ### L4. `deps.py:11` — Logger 名称 ```python logger = logging.getLogger(__name__) # __name__ = "src.server.deps" ``` 与 `app.py:17` 的 `logging.getLogger("md-vector-db")` 不一致,日志输出时名称不统一。 --- ## 架构评价 ### 做得好的地方 - **依赖方向正确**: `config ← db ← embedder ← ingest/search ← server/cli`,核心层不依赖传输层 - **策略模式**: `Embedder` Protocol + 工厂函数 `create_embedder`,方便扩展新 Provider - **依赖注入**: `AppState` + `FastAPI Depends`,可测试可替换 - **线程安全**: `threading.Lock` 保护 ChromaDB 写操作 - **安全**: API Key 认证 + 速率限制 + 路径遍历防护三重保护 - **MCP 友好**: CLI `--json` 输出,支持 stdin 输入 ### 待改进 | 问题 | 说明 | |------|------| | 封装泄漏 | `db.client` 被外部直接访问(MCP server、deps.py),应通过 `VectorDB` 方法暴露 | | AppConfig 构造 | 自定义 `__init__` 与 dataclass 冲突 | | 集合发现 | 没有统一的 "列出所有集合" 方法,deps/MCP/CLI 各自实现 | --- ## 总结 | 优先级 | 数量 | 建议处理 | |--------|------|----------| | HIGH | 3 | H1 (AppConfig) 需立即修,H2/H3 尽快修 | | MEDIUM | 4 | 逐个修,影响小但累积会出问题 | | LOW | 4 | 择机修,不紧急 | **整体评价**: 代码架构设计良好,核心逻辑清晰。主要问题集中在配置系统的 dataclass 使用不当、集合发现的封装泄漏、以及几处静默吞异常。修复 H1~H3 后即可达到生产就绪状态。