项目工具类开发详解

一、common.py：UTC 时间与 JSON 读写

设计思路

整个项目的时间戳统一使用UTC 时间，避免时区混乱。所有需要持久化的时间（trace 记录、版本号、入库报告）都用utc_now()，所有文件系统中的时间标识都用utc_file_stamp()。

# qa_core/common.py
import json
from datetime import datetime, timezone
from pathlib import Path

def utc_now() -> str:
    """返回 UTC ISO 时间字符串，用于版本管理、报告和 trace 的时间统一排序。
    格式: '2025-01-15T10:30:00.123456+00:00'
    """
    return datetime.now(timezone.utc).isoformat()

def utc_file_stamp() -> str:
    """返回适合放进文件名和版本号的 UTC 时间戳。
    格式: '20250115_103000'
    """
    return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")

def path_updated_at(path: str | Path) -> str:
    """返回文件修改时间的 UTC ISO 字符串，用于报告列表的排序。"""
    return datetime.fromtimestamp(Path(path).stat().st_mtime, timezone.utc).isoformat()

JSON 读写封装

项目统一使用ensure_ascii=False, indent=2格式写 JSON，保证中文字符可读、层级清晰。

# qa_core/common.py

def read_json(path: str | Path, default: Any = None) -> Any:
    """读取 JSON 文件，读取失败时返回 default —— 避免文件损坏导致进程崩溃。"""
    try:
        return json.loads(Path(path).read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return default

def read_json_dict(path: str | Path, default: dict[str, Any] | None = None) -> dict[str, Any]:
    """读取对象型 JSON，非对象或读取失败时返回字典默认值。
    用于旧 manifest 迁移、质量报告等对象型 JSON 文件。
    """
    payload = read_json(path, default=None)
    if isinstance(payload, dict):
        return payload
    return dict(default or {})

def write_json(path: str | Path, payload: Any) -> str:
    """按项目统一格式写入 JSON。自动创建父目录。返回绝对路径。"""
    output_path = Path(path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )
    return str(output_path)

列表报告工具

模板方法模式：通用的 JSON 文件扫描 + 按修改时间排序逻辑，被质量报告和状态页复用。

def list_json_reports(root: Path, glob_pattern: str, *, limit: int = 20) -> list[dict[str, Any]]:
    """列出目录下最近的 JSON 报告。
    每个条目包含 path / file_name / updated_at / payload 四个字段。
    被 `quality/ingestion.py` 的列表页复用。
    """
    if not root.exists():
        return []
    reports = []
    for path in sorted(root.glob(glob_pattern),
                       key=lambda item: item.stat().st_mtime, reverse=True):
        payload = read_json_dict(path)
        reports.append({
            "path": str(path), "file_name": path.name,
            "updated_at": path_updated_at(path), "payload": payload,
        })
        if limit and len(reports) >= limit:
            break
    return reports

二、utils.py：稳定哈希与指纹

设计思路

入库和检索都需要稳定的 ID 生成——同一个文件重新入库时产生相同的 chunk_id，才能正确覆盖旧数据。stable_hash()用 SHA-256 将任意数量的参数拼接后生成确定性的 hex 字符串。file_fingerprint()只读文件元数据（路径 + 修改时间 + 大小），不读内容，大幅减少增量入库的 IO。

# qa_core/utils.py
import hashlib
import os
from pathlib import Path

def stable_hash(*parts: object) -> str:
    """根据任意值创建稳定的 SHA-256 hex 字符串。

    关键特性：相同输入永远产生相同输出（确定性）。
    用于生成 chunk_id、faq_id、parent_id 等 Milvus 主键。

    例: stable_hash("hr_v2", "/data/入职流程.md", "chunk_3") 
        → 'a1b2c3d4e5f6...' (64 位 hex)
    """
    raw = "||".join("" if part is None else str(part) for part in parts)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def file_fingerprint(path: str | Path) -> str:
    """根据文件路径、修改时间和大小生成指纹，用于增量入库的变化判断。

    为什么不用内容哈希 (SHA-256 of content)？
    - 大文件（几十 MB 的 PDF）读内容做哈希太慢
    - 本地机器 CPU 有限，每轮入库读全量文件成本太高
    - 路径 + mtime + size 的组合在实际使用中已经足够可靠
    """
    p = Path(path)
    stat = p.stat()
    return stable_hash(str(p.resolve()), stat.st_mtime_ns, stat.st_size)

def normalize_source_from_path(path: str | Path) -> str:
    """从 '<source>_data' 目录名提取 source 标识。

    例: 'scenarios/enterprise_knowledge/data/hr_data' → 'hr'
        'data/legal_data' → 'legal'
    """
    name = os.path.basename(str(path)).replace("_data", "")
    return name or "default"

三、元数据 Store：MySQL 控制面

设计思路

当前项目的关键元数据已经进入 MySQL：知识库版本状态保存在kb_versions/kb_active_versions，文档增量入库状态保存在kb_document_manifests。这样版本激活、回滚、入库统计和增量跳过都可以被 API、脚本、测试和容器环境共享，不再依赖某个本地目录里的 JSON 文件。

质量报告仍然会使用common.py中的 JSON 读写工具生成可检查产物，但版本控制面和入库 manifest 只写 MySQL，避免同一份状态在文件和数据库之间分叉。

# qa_core/governance/kb_versions.py
class KnowledgeBaseVersionStore(_MySqlStore):
    """知识库版本状态机的 MySQL 存储实现。"""

    def __init__(self, scenario_id: str | None = None) -> None:
        """绑定业务场景；表结构由 bootstrap_mysql_schema() 统一初始化。"""

    def activate_version(self, kb_version: str) -> KnowledgeBaseVersion:
        """事务内切换 active 指针，不修改 Milvus 数据。"""

# qa_core/indexing/manifest.py
class IndexManifest(_MySqlStore):
    """文档增量入库清单的 MySQL 存储实现。"""

    def update(self, source, file_path, fingerprint, chunk_ids, *, scenario_id, kb_version):
        """记录文件指纹、kb_version、chunk_ids 和模型版本。"""

    def iter_records(self, *, scenario_id=None, source=None, kb_version=None):
        """按 scenario_id/source/kb_version 查询已入库文件。"""

四、schemas.py：Pydantic 数据模型

设计思路

所有 API 层的请求/响应都用 Pydantic BaseModel 定义，FastAPI 自动做参数校验和序列化。内部流转的数据结构用 Python dataclass（性能更好，不需要校验）。

# qa_core/schemas.py
from pydantic import BaseModel, Field

class RetrievalDebugRequest(BaseModel):
    """HTTP 检索诊断请求体。在线问答不复用该模型。"""
    query: str = Field(..., min_length=1)        # 必填，最短 1 字符
    source_filter: str | None = None              # 前端选择的业务分类
    session_id: str | None = None                 # 会话 ID，未传则自动生成
    scenario_id: str | None = None                # 场景 ID，未传用默认场景
    tenant_id: str | None = None                  # 租户 ID，未传默认 "default"
    dataset_id: str | None = None                 # 数据集 ID，未传默认 "default"
    visibility: str | None = None                 # 可见级别，未传默认 "public"
    user_role: str | None = None                  # 单一角色（前端简化入口）
    user_roles: list[str] = Field(default_factory=list)  # 多角色（正式入口）
    kb_version: str | None = None                 # 知识库版本，未传用 active 版本

class FeedbackRequest(BaseModel):
    """用户反馈载荷。rating 约束为 useful/not_useful，不允许自定义值。"""
    session_id: str | None = None
    question: str = Field(..., min_length=1)
    answer: str = Field(..., min_length=1)
    rating: str = Field(..., pattern="^(useful|not_useful)$")   # 仅允许这两个值
    comment: str | None = None
    sources: list[dict] = Field(default_factory=list)

class RetrievalDebugResponse(BaseModel):
    """检索调试响应，不包含最终答案，faq 和 doc 来源分开返回。"""
    query: str
    rewritten_query: str
    source_filter: str | None = None
    scenario_id: str | None = None
    tenant_id: str | None = None
    dataset_id: str | None = None
    visibility: str | None = None
    data_scope: dict | None = None
    kb_version: str | None = None
    intent: dict
    retrieval_plan: dict
    faq_sources: list[dict] = Field(default_factory=list)
    doc_sources: list[dict] = Field(default_factory=list)

内部流转的 dataclass

# qa_core/schemas.py (节选)
from dataclasses import dataclass, field

@dataclass
class RetrievalHit:
    """一次检索命中的封装：document + score。"""
    document: Any       # LangChain Document 对象
    score: float

@dataclass
class RetrievalResult:
    """一次完整检索的结果：命中列表 + 元信息。"""
    hits: list[RetrievalHit] = field(default_factory=list)
    query: str = ""
    source_type: str = ""      # "faq" 或 "doc"
    elapsed_ms: float = 0.0

    @property
    def top_score(self) -> float | None:
        """最高分，无命中时为 None。"""
        return self.hits[0].score if self.hits else None

五、config/settings.py：配置管理

设计思路

使用 pydantic-settings 从进程环境变量和本机.env加载全部配置。Docker Compose 部署时，.env.compose先由 Compose 注入到 API 容器的环境变量里，再由Settings读取。每个配置项有明确的默认值，注释标注了用途。model_config设置env_file=".env"，用于本机 API 调试；容器模式不要把.env.compose复制成.env。

# qa_core/config/settings.py (关键字段)
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)

    # ── 场景 ──
    active_scenario_id: str = Field(default="enterprise_knowledge", ...)
    scenario_config_dir: str = Field(default="scenarios", ...)

    # ── Milvus ──
    milvus_uri: str = Field(default="http://localhost:19530", ...)
    milvus_database: str = Field(default="", ...)

    # ── MySQL ──
    mysql_host: str = Field(default="localhost", ...)
    mysql_database: str = Field(default="qa_system", ...)

    # ── LLM (DashScope / OpenAI 兼容) ──
    llm_api_key: str = Field(default="", validation_alias="DASHSCOPE_API_KEY")
    llm_base_url: str = Field(default="https://dashscope.aliyuncs.com/compatible-mode/v1", ...)
    llm_model: str = Field(default="qwen-plus", ...)
    llm_temperature: float = Field(default=0.1, ge=0.0, le=1.0)
    llm_timeout: int = Field(default=60, ge=1)

    # ── 检索参数（所有默认值可用于 .env 覆盖）──
    faq_top_k: int = Field(default=20, ge=1)
    doc_top_k: int = Field(default=20, ge=1)
    faq_direct_score_threshold: float = Field(default=0.72, ge=0.0, le=1.0)
    max_context_chars: int = Field(default=6000, ge=1)
    max_context_doc_chars: int = Field(default=1600, ge=1)
    short_query_max_chars: int = Field(default=20, ge=1)

    # ── 历史摘要 ──
    history_summary_enabled: bool = Field(default=True)
    history_summary_after_messages: int = Field(default=14, ge=1)

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """进程级 Settings 单例。lru_cache 避免重复读取运行时配置。"""
    return Settings()

六、config/logging_config.py：结构化日志

设计思路

项目统一使用 Python 标准 logging + 自定义格式化器，输出 JSON 格式的结构化日志。get_logger(name)为每个模块提供带模块名的 logger 实例。

# qa_core/config/logging_config.py
import logging
import json
from datetime import datetime, timezone

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "time": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }, ensure_ascii=False)

def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JsonFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
    return logger

七、memory/base.py：MySQL 惰性连接基类

设计思路

ChatHistoryStore、FeedbackStore、KnowledgeBaseVersionStore和IndexManifest都需要 MySQL 连接，但不是每个请求都立即需要。_MySqlStore基类提供惰性初始化的引擎属性——第一次访问engine时才创建连接，创建后缓存复用。pool_pre_ping=True在连接被 MySQL 服务端断开后自动探活重连。

# qa_core/memory/base.py
from sqlalchemy import create_engine, text

class _MySqlStore:
    """MySQL 存储的公共基类。

    子类必须在访问 engine 之前设置 self.settings（提供 mysql_sync_uri）。
    子类: ChatHistoryStore, FeedbackStore, KnowledgeBaseVersionStore, IndexManifest
    """

    def __init__(self) -> None:
        self._engine = None

    @property
    def engine(self):
        """惰性创建的 SQLAlchemy 同步引擎。

        pool_pre_ping=True: 每次从连接池取连接前先发送 ping，
        避免 "MySQL server has gone away" 错误。
        """
        if self._engine is None:
            self._engine = create_engine(
                self.settings.mysql_sync_uri,
                pool_pre_ping=True
            )
        return self._engine

    # 表结构不写在 Store 基类里；启动期由 bootstrap 执行 runtime_schema.sql。

---