md-vector-db/docs/代码审计/代码架构审计报告.md

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 构造

# 问题代码 (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=...) 全部被绕过
• 所有字段失去类型检查

修复：

@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 只列出内存中的集合

# 问题代码
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 直接获取。

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 内存无限增长

# 问题代码
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：

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 吞掉所有异常

def delete_by_source(self, file_name: str) -> bool:
    try:
        with self.db.write_lock:
            ...
    except Exception:
        pass  # ← 隐藏了 ChromaDB 损坏、磁盘满等严重错误
    return False

修复：至少记录日志：

except Exception:
    logger.exception("删除文档失败: %s", file_name)
    return False

---

M2. db.py — 缺少 list_collections() 封装

VectorDB 没有暴露 list_collections()，导致：

• deps.py 无法从 db 对象获取集合列表
• MCP server 需要绕过封装直接访问 db.client

修复：

def list_collections(self) -> list:
    return self.client.list_collections()

---

M3. ingest.py:237-245 — _remove_by_source 吞掉去重异常

def _remove_by_source(self, file_name: str) -> None:
    try:
        ...
    except Exception:
        pass  # collection 为空时 get 可能抛异常

影响：如果去重失败（非空 collection），重复入库不会报错但会静默产生重复数据。

修复：区分 "collection 为空" 和真正的错误：

except ValueError:
    pass  # collection 无数据时 ChromaDB 会抛 ValueError
except Exception:
    logger.exception("去重检查失败: %s", file_name)

---

M4. cli/main.py:47-48 — 硬编码回退值

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

DEFAULT_CONFIG_PATH = "config.yaml"  # 从不同 CWD 运行会找不到

建议：在 load_config 中基于项目根目录解析。

---

L2. cli/main.py:97-101 — 函数内重复 import

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 的场景。

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 名称

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 后即可达到生产就绪状态。