知识库编译查询：让 AI 从「读文档」变成「查数据」

- entity: 张总
  type: person
  fact: 认为方案 A 技术可行
  evidence: PoC 验证，P99 延迟降低 40%
  time: 2026-05-23
  context: 产品周会，技术方案评审环节
# generated by hugo AI

topic: 方案 A 的评估
timeline:
  - date: 2026-04-10
    event: 首次提出，进入技术评估
    source: 产品周会
  - date: 2026-05-02
    event: 架构组否决，原因是运维成本高
    source: 架构评审会
  - date: 2026-05-23
    event: 重新提出，新方案采用云原生架构降低成本
    source: 产品周会
  - date: 2026-05-23
    status: pending
    decision: 待架构组二次评审
# generated by hugo AI

facts:
  - P99 延迟降低 40%
  - PoC 在预发环境运行
  - 样本量 1000 请求

judgments:
  - who: 张总
    judgment: 技术上可行
    confidence: medium
    basis: PoC 数据

decisions:
  - what: 进入架构评审
    status: pending
    owner: 架构组
    deadline: 2026-05-30
# generated by hugo AI

  [会议转写稿 15000 字] ──→ [AI: 生成会议纪要] ──→ 平庸的摘要

  ┌─────────────────────────────────────────────────────────────┐
  │ Stage 1: 编译                                               │
  │ [会议转写稿] ──→ [实体/决策/行动项/分歧/开放问题] ──→ JSON    │
  ├─────────────────────────────────────────────────────────────┤
  │ Stage 2: 查询                                               │
  │ [编译结果 + 知识库] ──→ [召回相关历史 + 溯源矛盾] ──→ 上下文  │
  ├─────────────────────────────────────────────────────────────┤
  │ Stage 3: 综合                                               │
  │ [编译结果 + 历史上下文] ──→ [AI: 生成带洞察的会议纪要]        │
  └─────────────────────────────────────────────────────────────┘

from dataclasses import dataclass, field
from enum import Enum
from datetime import date
from typing import Optional


@dataclass
class KnowledgeUnit:
    """编译后的知识单元——知识库的最小存储单位。"""
    id: str
    source: str
    compiled_at: str
    entities: list[str] = field(default_factory=list)
    facts: list[str] = field(default_factory=list)
    judgments: list[dict] = field(default_factory=list)
    decisions: list[dict] = field(default_factory=list)
    action_items: list[dict] = field(default_factory=list)
    timeline: list[dict] = field(default_factory=list)
    topic: str = ""
    raw_excerpt: str = ""
# generated by hugo AI


class QueryPattern(Enum):
    """四种查询模式，对应不同的分析任务。"""
    RECALL = "recall"        # 召回：找所有相关记录
    COMPARE = "compare"      # 对比：两个实体的差异
    TRACE = "trace"          # 溯源：决策的演变过程
    INSIGHT = "insight"      # 洞察：跨文档发现模式
# generated by hugo AI


COMPILE_PROMPT: str = """你是一个知识编译器。从以下内容中提取结构化知识单元。

## 提取规则
1. entities: 参与的人、项目、产品、技术方案
2. facts: 可验证的客观事实（带数据）
3. judgments: 主观判断（谁说的，信心程度）
4. decisions: 做出的决策（状态、负责人、截止时间）
5. action_items: 待办事项（负责人、截止时间）
6. timeline: 时间线事件

## 原则
- 事实与判断必须分离
- 每个单元必须自包含（不依赖原文也能理解）
- 保留关键数据（百分比、日期、人名）

## 输入
{raw_content}

输出 JSON 格式的知识单元。
"""
# generated by hugo AI


QUERY_TEMPLATES: dict[str, str] = {
    "recall": (
        "根据以下编译结果，回答：{query}\n\n"
        "编译结果：\n{compiled_units}\n\n"
        "要求：引用具体来源，标注时间。"
    ),
    "compare": (
        "对比以下两组编译结果，回答：{query}\n\n"
        "A组：\n{units_a}\n\nB组：\n{units_b}\n\n"
        "输出结构化对比表，标注数据来源和时间。"
    ),
    "trace": (
        "追踪以下决策的完整生命周期：{query}\n\n"
        "时间线编译：\n{timeline_units}\n\n"
        "按时间顺序输出：提出→讨论→修改→决定→执行→复盘。"
    ),
    "insight": (
        "综合以下多源编译结果，发现非显而易见的模式：{query}\n\n"
        "编译结果：\n{compiled_units}\n\n"
        "要求：模式必须有数据支撑，不是主观臆测。"
    ),
}
# generated by hugo AI


def filter_units(
    units: list[KnowledgeUnit],
    query: str,
    time_range: Optional[tuple[date, date]] = None,
) -> list[KnowledgeUnit]:
    """按关键词和时间范围过滤知识单元。"""
    results: list[KnowledgeUnit] = []
    for unit in units:
        if time_range:
            unit_date = date.fromisoformat(unit.compiled_at[:10])
            if not (time_range[0] <= unit_date <= time_range[1]):
                continue
        # 简单的关键词匹配，生产环境用向量检索
        searchable = " ".join(
            unit.entities + unit.facts + [unit.topic]
        )
        if query.lower() in searchable.lower():
            results.append(unit)
    return results


def build_query_prompt(
    pattern: QueryPattern,
    query: str,
    units: list[KnowledgeUnit],
    **kwargs: str,
) -> str:
    """根据查询模式构建 prompt。"""
    import json

    compiled_json = json.dumps(
        [u.__dict__ for u in units],
        ensure_ascii=False,
        indent=2,
    )
    template = QUERY_TEMPLATES[pattern.value]
    return template.format(
        query=query,
        compiled_units=compiled_json,
        **kwargs,
    )
# generated by hugo AI