2 Commits

Author SHA1 Message Date
iomgaa acc419a29b chore: add a mechanical wiki-vs-source alignment checker
七轮人工审查的 88 条发现里,签名/导出/字段序/列数/env 键这几类是机械可
比对的,不该靠人一轮轮追。五项检查全部由源码反推,已用注入历史错误的方式
验证有效: gather_bounded(coros, limit)、source_name 排进前 11 位、列数写
成 20、EXTRA_BODY 漏文档 —— 四条全部命中。

不并入 make ci: wiki 是独立仓库,仓库里没有它时自动跳过等于静默降级(违
P5),故做成显式的 make wiki-check WIKI=<path>。
2026-08-02 02:16:16 -04:00
iomgaa de7273598e docs: correct the scope normalization comment on Redis key impact
Both Redis backends have lowercased scope in their own constructors since
v1.0.0, so case never split the keyspace. What actually does is whitespace:
the backends lower but do not strip.
2026-08-02 00:38:42 -04:00
3 changed files with 233 additions and 4 deletions
+8 -1
View File
@@ -1,4 +1,4 @@
.PHONY: install test lint format check ci wiki .PHONY: install test lint format check ci wiki wiki-check
ENV := PolyGateway ENV := PolyGateway
@@ -24,3 +24,10 @@ ci: check test
wiki: wiki:
conda run -n $(ENV) python3 .claude/tools/research_wiki.py rebuild_index research-wiki/ conda run -n $(ENV) python3 .claude/tools/research_wiki.py rebuild_index research-wiki/
# 用户文档站(Gitea Wiki)与源码的机械对齐校验。wiki 是独立仓库,须显式给路径:
# make wiki-check WIKI=~/PolyGateway.wiki
# 不并入 ci: 仓库里没有 wiki,自动跳过等于静默降级(违 P5),宁可让人显式跑。
wiki-check:
@test -n "$(WIKI)" || (echo "用法: make wiki-check WIKI=<PolyGateway.wiki 克隆路径>" && exit 1)
conda run -n $(ENV) python3 tools/check_wiki_alignment.py --wiki $(WIKI)
+5 -3
View File
@@ -149,9 +149,11 @@ class GatewaySettings:
def _normalize(self) -> None: def _normalize(self) -> None:
"""把 `from_env` 一直在做的规范化补到构造路上,两条路必须产出同一个值。 """把 `from_env` 一直在做的规范化补到构造路上,两条路必须产出同一个值。
`scope` 最要紧: 它直接进 Redis key(`pgw:limit:{scope}:…`/`pgw:gate:{scope}:…`)。 `scope` 的 strip 才是要紧的那一半: 它进 Redis key(`pgw:limit:{scope}:…`
一个进程走 `from_env("LLM")` 拿到 "llm"、另一个直接构造传 "LLM",同一逻辑 /`pgw:gate:{scope}:…`),而两个 Redis 后端在构造函数里只 `.lower()` **不 strip**
scope 的限流与熔断状态会分裂两套命名空间,各记各的,治理静默失效且不报错 ——`"llm "` 会产出 `pgw:limit:llm :…`,与 `from_env` 路的进程分裂两套命名空间。
大小写则不会: 后端自 v1.0.0 起各自 lower,`from_settings` 传 "LLM" 也落在同一
套 key 上(此处 lower 只为让 `GatewaySettings.scope` 属性两路取值一致)。
空串归 None 同理: 留着空串会骗过 `is None` 判断,把错误推迟到 redis 客户端 空串归 None 同理: 留着空串会骗过 `is None` 判断,把错误推迟到 redis 客户端
抛连接串解析异常。`telemetry_pg_dsn` 的驱动后缀因为要看 backend 且需告警, 抛连接串解析异常。`telemetry_pg_dsn` 的驱动后缀因为要看 backend 且需告警,
+220
View File
@@ -0,0 +1,220 @@
#!/usr/bin/env python3
"""机械校验 Gitea Wiki 与源码的可比对事实(签名/导出/字段序/列清单/env 键)。
**只查机械可比对的部分**——机制语义、行为口径这类需要读懂代码才能判断的断言
不在此列(那部分靠 `解释-治理行为` 的适用性总表做单一事实源 + 人工审查)。
设计动机: 2026-08 对 wiki 做了七轮人工审查,88 条发现里有相当一部分属于
"机械可校验却写错"——`gather_bounded(coros, limit)`(实为 keyword-only 的
`concurrency`)、`LLMResponse` 字段表把 `source_name` 排进前 11 位、遥测列数
写成 20/21(实为 22 列)、`__version__` 在自称"全集"的页面缺席。这类偏差不该
靠人一轮轮追,故收敛为脚本。
用法(wiki 是独立仓库,须显式给路径;**不做 skip 静默降级**):
python3 tools/check_wiki_alignment.py --wiki /path/to/PolyGateway.wiki
"""
from __future__ import annotations
import argparse
import dataclasses
import inspect
import re
import sys
from pathlib import Path
import polygateway
from polygateway import EmbeddingClient, GatewayClient, LLMResponse
from polygateway.config import _SOURCE_FIELDS
from polygateway.ocr import OcrClient
from polygateway.providers import register_provider
from polygateway.telemetry.sqlite import _COLUMNS as TELEMETRY_COLUMNS
# 参数名允许在 wiki 里以别名出现的白名单(仅限确无歧义的自解释形参)
_PARAM_ALIASES: dict[str, set[str]] = {"env": {"env"}}
# (符号, 可调用对象) —— 这些的签名必须在 wiki 里逐参数出现
_SIGNATURE_TARGETS = [
("GatewayClient.chat", GatewayClient.chat),
("GatewayClient.from_env", GatewayClient.from_env),
("EmbeddingClient.from_env", EmbeddingClient.from_env),
("EmbeddingClient.embed", EmbeddingClient.embed),
("OcrClient.from_env", OcrClient.from_env),
("OcrClient.recognize_text", OcrClient.recognize_text),
("gather_bounded", polygateway.gather_bounded),
("register_provider", register_provider),
]
def _wiki_text(wiki: Path) -> dict[str, str]:
"""读全部 .md;文件名(不含后缀)→ 正文。"""
pages = {p.stem: p.read_text(encoding="utf-8") for p in sorted(wiki.glob("*.md"))}
if not pages:
raise SystemExit(f"错误: {wiki} 下没有 .md 文件,路径是否指向 wiki 克隆?")
return pages
def check_exports_documented(pages: dict[str, str]) -> list[str]:
"""`__all__` 每一项都得在某页出现过(R1 漏 gather_bounded、R6 漏 __version__)。"""
blob = "\n".join(pages.values())
missing = [name for name in polygateway.__all__ if name not in blob]
return [f"__all__ 的 {name!r} 在全部 wiki 页面中零命中(页首自称『顶层导出全集』)"
for name in missing]
def check_signatures(pages: dict[str, str]) -> list[str]:
"""提到某个公共可调用的那一行,必须列全它的参数名。
只查「参数名是否出现」,不查顺序与类型——后者用自然语言表述合法。
历史命中: gather_bounded 的 concurrency 被写成 limit;EmbeddingClient/
OcrClient 的 from_env 用省略号承接 chat 的关键字集合,掩盖了没有 cache=。
"""
problems = []
for label, func in _SIGNATURE_TARGETS:
symbol = label.split(".")[-1]
params = [
p.name
for p in inspect.signature(func).parameters.values()
if p.name not in ("self", "cls")
]
# 找出提到该符号的所有行,任一行列全即算通过
lines = [
line
for text in pages.values()
for line in text.splitlines()
if f"`{symbol}`" in line or f"{symbol}(" in line
]
if not lines:
problems.append(f"{label}: wiki 里找不到任何提及")
continue
best_missing: list[str] | None = None
for line in lines:
missing = [
p for p in params
if p not in line and not (_PARAM_ALIASES.get(p, set()) & set(line.split()))
]
if not missing:
best_missing = []
break
if best_missing is None or len(missing) < len(best_missing):
best_missing = missing
if best_missing:
problems.append(
f"{label}: 没有任何一行列全参数,最接近的一行仍缺 {best_missing}"
f"(实际签名 {inspect.signature(func)})"
)
return problems
def check_llmresponse_field_order(pages: dict[str, str]) -> list[str]:
"""字段表出现顺序须与 dataclass 声明顺序一致。
R2 命中: wiki 把 source_name 与 model/provider 并成一行(位置 5),而它实为
第 12 个字段。迁移中的三项目按位置构造 fake,照 wiki 写会静默错位。
"""
page = pages.get("参考-公共API")
if page is None:
return ["缺少 参考-公共API.md"]
# 只在 LLMResponse 小节内找: 别的类型(EmbeddingResponse 等)也有同名字段,
# 全页搜索会命中它们、把顺序判断带偏
start = page.find("## LLMResponse")
if start < 0:
return ["参考-公共API.md 缺少 `## LLMResponse` 小节"]
end = page.find("\n## ", start + 1)
section = page[start : end if end > 0 else len(page)]
declared = [f.name for f in dataclasses.fields(LLMResponse)]
positions = []
for name in declared:
# 取该字段在小节内最早的出现位置(表格首列可能写成 `a / b` 合并形式)
cands = [
section.find(pat)
for pat in (f"| {name} ", f"{name} /", f"/ {name} ", f"| {name}\n")
]
hits = [i for i in cands if i >= 0]
positions.append((name, min(hits) if hits else -1))
documented = [n for n, i in positions if i >= 0]
missing = [n for n, i in positions if i < 0]
problems = [f"LLMResponse 字段 {missing} 未在 参考-公共API 的字段表出现"] if missing else []
ordered = sorted((i, n) for n, i in positions if i >= 0)
actual = [n for _, n in ordered]
expected = [n for n in declared if n in documented]
if actual != expected:
problems.append(
f"LLMResponse 字段表顺序与声明顺序不符\n"
f" wiki 顺序: {actual}\n"
f" 声明顺序: {expected}"
)
return problems
def check_telemetry_columns(pages: dict[str, str]) -> list[str]:
"""遥测列清单与 sqlite 后端的 _COLUMNS 对齐(created_at 由 DDL 生成,单列)。"""
page = pages.get("指南-遥测与成本")
if page is None:
return ["缺少 指南-遥测与成本.md"]
expected = [*TELEMETRY_COLUMNS, "created_at"]
missing = [c for c in expected if c not in page]
problems = [f"遥测列 {missing} 未在 指南-遥测与成本 出现"] if missing else []
# 列数声明: 表 = _COLUMNS + created_at;端口 = _COLUMNS。
# 页内**所有** "N 列" 声明都必须等于真实列数——只查"正确值是否出现"会被
# 漏改的旧数字骗过(它们同时存在时检查照样通过)
table_n, port_n = len(expected), len(TELEMETRY_COLUMNS)
declared_counts = {int(m) for m in re.findall(r"(\d+)\s*列", page)}
if not declared_counts:
problems.append(f"指南-遥测与成本 未声明表列数(应为 {table_n} 列)")
elif declared_counts != {table_n}:
wrong = sorted(declared_counts - {table_n})
problems.append(
f"指南-遥测与成本 的列数声明 {wrong} 与实际 {table_n} 列不符"
f"(端口是 {port_n} 参数,两者差 created_at)"
)
return problems
def check_source_env_fields(pages: dict[str, str]) -> list[str]:
"""`_SOURCE_FIELDS` 的每个 FIELD 段都得在 参考-配置键 出现(R1 命中 EXTRA_BODY)。"""
page = pages.get("参考-配置键")
if page is None:
return ["缺少 参考-配置键.md"]
# 用词边界匹配: `f in page` 会让 EXTRA_BODY 被 EXTRA_BODYY 蒙混过关
missing = [f for f in _SOURCE_FIELDS if not re.search(rf"\b{re.escape(f)}\b", page)]
return [f"源键 FIELD 段 {missing} 未在 参考-配置键 文档化"] if missing else []
_CHECKS = [
("顶层导出覆盖", check_exports_documented),
("公共签名参数", check_signatures),
("LLMResponse 字段序", check_llmresponse_field_order),
("遥测列清单", check_telemetry_columns),
("源 env 键覆盖", check_source_env_fields),
]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--wiki", required=True, type=Path, help="PolyGateway.wiki 克隆目录")
args = parser.parse_args()
if not args.wiki.is_dir():
raise SystemExit(f"错误: {args.wiki} 不是目录")
pages = _wiki_text(args.wiki)
failed = 0
for label, check in _CHECKS:
problems = check(pages)
if problems:
failed += len(problems)
print(f"{label}")
for p in problems:
print(f" {p}")
else:
print(f"{label}")
print()
if failed:
print(f"{failed} 处机械偏差 —— wiki 与源码不一致")
return 1
print(f"{len(pages)} 页机械校验通过")
return 0
if __name__ == "__main__":
sys.exit(main())