feat(provenance): URI 统一 + resolver + 强验证(T12 架构审查整改)

- T12 (OV3, P1): 新建 src/genesis/parsers/resolver.py
  - parse_source_uri: 解析 file.xlsx#Sheet!CellRef → SourceRef(格式非法 raise URIError)
  - provenance_to_uri: Provenance 还原 URI(与 build_source_uri 互逆)
  - resolve_source_uri: StructuredSource 内定位真实 CellValue
  - validate_source_uris: 批量强验证 → ValidationResult(resolved/unresolved)
    格式错误或源中不存在一律 unresolved(防 QA#8 编造 URI 作弊)
- URI 唯一生成入口 build_source_uri(provenance.py),无散落不一致
- 新增 test_resolver.py(13 用例);同步 design.md §9.2 + §6.8 第五步
- TDD: RED(模块缺失)→ GREEN(聚焦 10 passed)→ 全量 231 passed / 100.00%(1191 stmts/298 br)
This commit is contained in:
lhl
2026-08-12 22:35:56 +08:00
parent 6a54580ca4
commit 69aec7716c
4 changed files with 270 additions and 1 deletions
+1
View File
@@ -78,4 +78,5 @@
| 2026-08-11 | Agent 实现 | T9(架构审查整改):CJK 保守 token 估算(Issue11)。token.py approximate_token_count 重写:新增 _is_cjk_charCJK 统一表意/扩展A/假名/韩文/兼容/全角六大 Unicode 范围)+ _CJK_TOKENS_PER_CHAR=1.5(旧逻辑 4 字符 1 token 对中文/日文严重低估,裁剪失效致 API 超限);CJK 字符按 1.5 token/字符,其余仍 4 字符 1 token,最少 1 token;同步 config-design.md token_estimation 注释;新增 4 用例(纯 CJK 保守/ASCII 不回归/混合文本/全角符号);TDD 验证 RED((設計)4 字符仅 1 token)→ GREEN(聚焦 11 passed)→ 全量 186 passed 覆盖 100.00%995 stmts/252 br),fail_under=99 达标 | src/genesis/inference/token.py, tests/test_inference_token.py, docs/config-design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free |
| 2026-08-11 | Agent 实现 | T5(架构审查整改):删死配置 + 同步文档(Issue5 + OV1)。config.py 删除 QdrantStoreConfig 类与 VectorStoreConfig.qdrant 字段、AppConfig.task_queue.redis_url;同步更新 6 处文档(api-design §1/§4.3/§5.2/§5.3/§6.2TaskQueue 标注 v1 仅 InMemory、Redis/Valkey 为 v2 预留;rag-layer §9Storage Adapter 仅 ChromaAdapter、移除切换流程/工厂 qdrant 分支;agent-runtime §3.1/§3.5design §5.5/§8.4.1config-design env/app.yaml/rag.yaml/docker composeweb-ui §4.1+ tests/fixtures/rag.yaml 去 qdrant 段;历史评审记录(design-review/web-ui-review/phase1 plan)保留原样不改写;新增 3 用例(QdrantStoreConfig 已删/vector_store 无 qdrant 字段/task_queue 无 redis_url+ 同步 2 个既有 qdrant 依赖用例;TDD 验证 RED(三处死配置存在)→ GREEN(聚焦 10 passed)→ 全量 189 passed 覆盖 100.00%991 stmts/252 br),fail_under=99 达标 | src/genesis/config.py, tests/test_config.py, tests/fixtures/rag.yaml, docs/api-design.md, docs/rag-layer-design.md, docs/agent-runtime-design.md, docs/design.md, docs/config-design.md, docs/web-ui-design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free |
| 2026-08-11 | Agent 实现 | T6+T11(架构审查整改,Lane B):v1 rerank 精排 + bge-m3 多语言切换(Issue6 + OV2)。config.py 新增 RerankConfigenabled=True/model=BAAI/bge-reranker-v2-m3/device=cpu)挂入 RagConfigEmbeddingConfig.model 默认 bge-small-zh-v1.5 → BAAI/bge-m3(实际语料日文);rag-layer-design.md 新增 §2.3 多语言与日文样本验证、§6.3 Rerank 精排(窗口=RRF top-10、候选≤top_k 跳过、故障降级 RRF 原序),原 §6.3-6.6 顺延 6.4-6.7;选型表/依赖表/manifest/流程图 bge-small-zh → bge-m3config-design.md embedding 默认 + 新增 rerank 段;design.md §5.5 与 implementation-plan 4.3 同步;新增 tests/test_rag_design_consistency.py 一致性门禁(6 用例:代码默认/fixture 同步/4 文档用 bge-m3+reranker/无 legacy 引用);TDD 验证 RED(默认模型仍旧+rerank 字段不存在)→ GREEN(聚焦 13 passed)→ 全量 198 passed 覆盖 100.00%996 stmts/252 br),fail_under=99 达标 | src/genesis/config.py, tests/test_config.py, tests/test_rag_design_consistency.py, tests/fixtures/rag.yaml, docs/rag-layer-design.md, docs/config-design.md, docs/design.md, docs/implementation-plan.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free |
| 2026-08-11 | Agent 实现 | T12(架构审查整改,P1):URI 统一 + resolver + 强验证(OV3)。新建 src/genesis/parsers/resolver.pyparse_source_uri(解析 file.xlsx#Sheet!CellRef → SourceRef,格式非法 raise URIError)、provenance_to_uriProvenance 还原,与 build 互逆)、resolve_source_uriStructuredSource 内定位真实 CellValue)、validate_source_uris(批量强验证 → ValidationResult(resolved/unresolved),格式错误或源中不存在一律 unresolved,防 QA#8 编造 URI 作弊);统一 URI 唯一生成入口 build_source_uriprovenance.py),formatting_detector 经其生成,无散落不一致;新建 tests/test_resolver.py13 用例:解析/往返/定位/批量验证/防御分支);同步 design.md §9.2 机制化说明 + §6.8 第五步存在性校验引用;TDD 验证 RED(模块缺失)→ GREEN(聚焦 10 passed)→ 全量 231 passed 覆盖 100.00%1191 stmts/298 br),fail_under=99 达标 | src/genesis/parsers/resolver.py, tests/test_resolver.py, docs/design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free |
| 2026-08-11 | Agent 实现 | T14+T16(架构审查整改,Lane A):DataGate 机制化 + 任务级持久化(OV5 + OV7)。新建 src/genesis/orchestrator/ 包:datagate.pyDataGate.load(source, selector):子集加载 + 规模保护 max_total_rows=500 无 selector 拒绝全量 + token 预算 max_total_tokens=8000 复用 CJK 保守估算 + 未知表容错)+ task_queue.pyTaskQueue ABC + PersistentTaskQueue SQLite 落盘:enqueue/poll/update_status/get/cancel/recover/close + 幂等去重 §5.3 + recover 将 running→failed、pending 保留);新建 tests/test_datagate.py8 用例:子集/规模保护/1000 行 Excel selector 放行/token 预算/未知表/空 selector+ tests/test_task_queue.py11 用例:CRUD/幂等缓存/重开不丢/recover 语义/防御分支);同步 agent-runtime-design.md(§4.2 原则→机制 + §3.5/3.6 任务级恢复已实现)、api-design.md §5.2/5.3、design.md §8.4.1TDD 验证 RED(模块缺失/错误消息不匹配)→ GREEN(聚焦 8+8 passed)→ 全量 218 passed 覆盖 100.00%1140 stmts/278 br),fail_under=99 达标 | src/genesis/orchestrator/__init__.py, src/genesis/orchestrator/datagate.py, src/genesis/orchestrator/task_queue.py, tests/test_datagate.py, tests/test_task_queue.py, docs/agent-runtime-design.md, docs/api-design.md, docs/design.md, _AI_USAGE_LOG.md | deepseek-v4-flash-free |
+11 -1
View File
@@ -897,7 +897,8 @@ style_map:
│ │
│ ⑤ 校验输出: │
│ 内容块结构校验(schema 校验)→ 失败则重试 │
│ source_uris 存在性校验(引用的 URI 必须在输入中存在)
│ source_uris 存在性校验(引用的 URI 必须在输入中定位,
│ T12 resolver.validate_source_uris → unresolved 非空即失败)│
│ │
│ ⑥ 渲染: │
│ 产出 chapter_html → 前端实时预览 │
@@ -1374,6 +1375,15 @@ DB设计 生成过程中发生错误
例: "要件定義.xlsx#機能一覧!C3"
```
> **T12 机制化(OV3)**:URI 的解析/存在性验证已落地为 `src/genesis/parsers/resolver.py`
> - `parse_source_uri(uri)`:解析为 SourceRef(格式非法 → `URIError`
> - `provenance_to_uri(prov)`:从 Provenance 还原 URI(与 build 互逆)
> - `resolve_source_uri(uri, source)`:在 StructuredSource 内定位真实单元格
> - `validate_source_uris(uris, source)` → ValidationResult(resolved, unresolved)
>
> 该机制支撑 design.md §6.8 第五步「source_uris 存在性校验」——QA 校验时
> 所有引用的 URI 必须能在输入源中定位,否则落入 unresolved(防 QA#8 编造 URI 作弊)。
### 9.3 Provenance Chain
```
+100
View File
@@ -0,0 +1,100 @@
"""URI resolver 与强验证(T12OV3)。
背景:design.md §9.2 定义 Citation URI 格式 `file.xlsx#SheetName!ColumnRow`
但仅 `build_source_uri` 存在,无解析、无存在性验证。OV3 裁定将其机制化:
- parse_source_uri:把 URI 解析为结构化 SourceRef(格式不一致即报错)
- provenance_to_uri:从 Provenance 还原 URI(与 build 互为逆)
- resolve_source_uri:在 StructuredSource 内定位真实单元格(存在性校验)
- validate_source_uris:批量强验证,区分 resolved/unresolved(防 QA#8 作弊——
编造的 URI 无法在源中定位,必落入 unresolved)
"""
from __future__ import annotations
from dataclasses import dataclass
from genesis.data_models import CellValue, Provenance, StructuredSource
from genesis.parsers.provenance import build_source_uri
class URIError(ValueError):
"""URI 格式非法(不符合 file.xlsx#SheetName!CellRef)。"""
@dataclass(frozen=True)
class SourceRef:
"""URI 解析后的结构化定位。"""
file_name: str
sheet_name: str
cell_ref: str
def parse_source_uri(uri: str) -> SourceRef:
"""解析 `file.xlsx#SheetName!C3` → SourceRef。
Raises:
URIError: 缺 `#` / 缺 `!` / 任一分段为空。
"""
if not isinstance(uri, str) or "#" not in uri or "!" not in uri:
raise URIError(f"URI 格式非法(期望 file.xlsx#SheetName!CellRef: {uri!r}")
file_part, rest = uri.split("#", 1)
if not file_part or "!" not in rest:
raise URIError(f"URI 格式非法(期望 file.xlsx#SheetName!CellRef: {uri!r}")
sheet_name, cell_ref = rest.split("!", 1)
if not sheet_name or not cell_ref:
raise URIError(f"URI 格式非法(Sheet/Cell 段不可为空): {uri!r}")
return SourceRef(file_name=file_part, sheet_name=sheet_name, cell_ref=cell_ref)
def provenance_to_uri(prov: Provenance) -> str:
"""从 Provenance 还原 URI(与 build_source_uri 互逆)。"""
cell_ref = f"{prov.column}{prov.row}"
return build_source_uri(prov.file_name, prov.sheet_name, cell_ref)
def resolve_source_uri(uri: str, source: StructuredSource) -> CellValue | None:
"""在 StructuredSource 中定位 URI 指向的真实单元格;不存在返回 None。"""
ref = parse_source_uri(uri)
for table in source.tables:
for row in table.rows:
for cell in row.values():
if _matches(cell, ref):
return cell
return None
def validate_source_uris(uris: list[str], source: StructuredSource) -> "ValidationResult":
"""批量强验证:把 URI 分为可在源中定位(resolved)与不可定位(unresolved)。
格式错误或源中不存在的 URI 一律归入 unresolved —— 供 QA 校验断言
「所有引用的 URI 必须存在于输入中」(design.md §6.8 第五步,T12 落地)。
"""
resolved: list[str] = []
unresolved: list[str] = []
for uri in uris:
try:
if resolve_source_uri(uri, source) is not None:
resolved.append(uri)
else:
unresolved.append(uri)
except URIError:
unresolved.append(uri)
return ValidationResult(resolved=resolved, unresolved=unresolved)
@dataclass
class ValidationResult:
resolved: list[str]
unresolved: list[str]
def _matches(cell: CellValue, ref: SourceRef) -> bool:
prov = cell.provenance
if prov is None:
return False
return (
prov.file_name == ref.file_name
and prov.sheet_name == ref.sheet_name
and f"{prov.column}{prov.row}" == ref.cell_ref
)
+158
View File
@@ -0,0 +1,158 @@
"""URI resolver + 强验证测试(T12OV3)。
OV3 裁定:可追溯性标准未操作性定义(URI 格式不一致、无 resolver、QA#8 可作弊)
→ 统一 URI + resolver + 强验证。
本文件测试:URI 解析、provenance 往返、结构化源内定位、批量强验证(防 QA 作弊)。
"""
from __future__ import annotations
import pytest
from genesis.data_models import (
CellComment,
CellValue,
ExcelTable,
ParsedTemplate,
Provenance,
RuleDocument,
SheetType,
StructuredSource,
)
from genesis.parsers.provenance import build_source_uri
from genesis.parsers.resolver import (
SourceRef,
URIError,
ValidationResult,
parse_source_uri,
provenance_to_uri,
resolve_source_uri,
validate_source_uris,
)
# ---------- 构造小源 ----------
def _cell(file_name: str, sheet: str, row: int, col: str, value: str) -> CellValue:
return CellValue(
value=value,
provenance=Provenance(
file_name=file_name, sheet_name=sheet, row=row, column=col, column_header="x"
),
)
def _source_with(cells: list[CellValue]) -> StructuredSource:
table = ExcelTable(
name="機能一覧",
detected_type=SheetType.FUNCTION,
extraction_method="structured",
headers=["v"],
rows=[{"v": c} for c in cells],
)
return StructuredSource(
tables=[table],
template=ParsedTemplate(file_name="t.docx", sections=[], placeholders={}, styles={}),
rule_docs=[],
image_analyses=[],
existing_system=None,
comments=[],
)
# ---------- URI 解析 ----------
def test_parse_source_uri():
ref = parse_source_uri("要求.xlsx#機能一覧!C3")
assert ref.file_name == "要求.xlsx"
assert ref.sheet_name == "機能一覧"
assert ref.cell_ref == "C3"
def test_parse_source_uri_rejects_missing_hash():
with pytest.raises(URIError):
parse_source_uri("要求.xlsx機能一覧!C3")
def test_parse_source_uri_rejects_missing_bang():
with pytest.raises(URIError):
parse_source_uri("要求.xlsx#機能一覧")
def test_parse_source_uri_rejects_empty_file():
with pytest.raises(URIError):
parse_source_uri("#機能一覧!C3")
def test_parse_source_uri_rejects_empty_sheet_or_cell():
with pytest.raises(URIError):
parse_source_uri("f.xlsx#!C3")
with pytest.raises(URIError):
parse_source_uri("f.xlsx#Sheet!")
def test_resolve_skips_cell_without_provenance():
"""CellValue.provenance 为 None 时不匹配任何 URI_matches 防御分支)。"""
orphan = CellValue(value="x", provenance=None)
source = _source_with([orphan])
assert resolve_source_uri("f.xlsx#機能一覧!C3", source) is None
# ---------- provenance 往返 ----------
def test_provenance_to_uri_roundtrip():
prov = Provenance(file_name="f.xlsx", sheet_name="S", row=3, column="C", column_header="h")
uri = provenance_to_uri(prov)
assert uri == "f.xlsx#S!C3"
ref = parse_source_uri(uri)
assert ref == SourceRef(file_name="f.xlsx", sheet_name="S", cell_ref="C3")
def test_build_and_parse_roundtrip():
uri = build_source_uri("要求.xlsx", "機能一覧", "A5")
ref = parse_source_uri(uri)
assert ref == SourceRef(file_name="要求.xlsx", sheet_name="機能一覧", cell_ref="A5")
# ---------- 结构化源内定位 ----------
def test_resolve_finds_existing_cell():
cell = _cell("f.xlsx", "機能一覧", 3, "C", "登録")
source = _source_with([cell])
found = resolve_source_uri("f.xlsx#機能一覧!C3", source)
assert found is not None
assert found.value == "登録"
def test_resolve_returns_none_for_missing():
cell = _cell("f.xlsx", "機能一覧", 3, "C", "登録")
source = _source_with([cell])
assert resolve_source_uri("f.xlsx#機能一覧!D9", source) is None
def test_resolve_distinguishes_sheet():
cell = _cell("f.xlsx", "機能一覧", 3, "C", "登録")
source = _source_with([cell])
# 不同 Sheet 的同 cell_ref 不应命中
assert resolve_source_uri("f.xlsx#画面一覧!C3", source) is None
# ---------- 批量强验证(防 QA 作弊) ----------
def test_validate_source_uris_split():
cell = _cell("f.xlsx", "機能一覧", 3, "C", "登録")
source = _source_with([cell])
uris = ["f.xlsx#機能一覧!C3", "f.xlsx#機能一覧!D9", "invalid-uri"]
result = validate_source_uris(uris, source)
assert isinstance(result, ValidationResult)
assert result.resolved == ["f.xlsx#機能一覧!C3"]
assert result.unresolved == ["f.xlsx#機能一覧!D9", "invalid-uri"]
def test_validate_rejects_malformed_separately():
"""格式错误的 URI 应归入 unresolvedQA#8 不可靠 URI 无效,防作弊)。"""
cell = _cell("f.xlsx", "機能一覧", 3, "C", "登録")
source = _source_with([cell])
result = validate_source_uris(["garbage-without-hash"], source)
assert result.resolved == []
assert result.unresolved == ["garbage-without-hash"]