分析能力骨架:app/analysis 契约(结论的工件种类、台账条目、写作目录前缀 A、引用标记 a、分析卡节点类型、五个算子与模式、方法版本、判定码、算不了的原因码、附带说明码、工具名)、设置缺省 / 取值范围 / 叫法与校验、阈值合并(缺省 ← 发起时快照 ← 分析卡覆盖)、复算比较;run / recompute 的签名(第 1 波实现);测试用的假结论

This commit is contained in:
ThearchyHelios
2026-10-21 11:53:43 +08:00
parent 9e5debfb33
commit 844f5571f0
4 changed files with 487 additions and 0 deletions
+38
View File
@@ -0,0 +1,38 @@
"""确定性分析算子(docs/design/analysis-round/):变化拆解、波动基线、趋势、差异检验、可比性检查。
入口只有两个函数,agent 分析工具(tools/builtin)和分析卡(engine/nodes/analysis.py)都只经它们调用:
- run:按请求算一份分析结论。resolve 按调用里写的名字(Q3、来源节点 id)取回输入快照;单元格写成「名字.r行.列」
(Q1.r0.gmv),由这里拆开再经 resolve 取快照。parents 按名字(A3、分析卡里的项 id)取回上一层拆解,下钻用。
调用写错抛 contract.CallError;算不了返回 status refused 的结论。唯一的异步来自 context["probe"](现查表的日期范围)。
- recompute:按结论里记下的 params、settings、probes 和输入快照(按 sha256 取回)重算一遍,不现查;和封存的用
contract.same 比。
契约(名字、形状、缺省阈值)在 contract.py。
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from typing import Any
from app.analysis.contract import AnalysisRequest, AnalysisResult, Context, InputSnapshot
#: 按调用里写的名字取回输入快照;没有这个名字返回 None
Resolver = Callable[[str], InputSnapshot | None]
#: 按名字取回上一层的分析结论(下钻);没有返回 None
ParentResolver = Callable[[str], AnalysisResult | None]
async def run(request: AnalysisRequest, resolve: Resolver, *, settings: Mapping[str, Any], context: Context,
parents: ParentResolver | None = None) -> AnalysisResult:
"""算一份分析结论。settings 是 contract.effective_settings 算好的阈值。"""
raise NotImplementedError("analysis.run:第 1 波 W1-a 实现")
def recompute(result: AnalysisResult, load: Callable[[str], InputSnapshot | None], *,
load_result: Callable[[str], AnalysisResult | None] | None = None) -> AnalysisResult:
"""复算:load 按 sha256 取回输入快照,load_result 按 sha256 取回 parent 那份结论。"""
raise NotImplementedError("analysis.recompute:第 1 波 W1-a 实现")
__all__ = ["ParentResolver", "Resolver", "recompute", "run"]
+332
View File
@@ -0,0 +1,332 @@
"""分析结论的契约(docs/design/analysis-round/00-总方案.md §3.1,细案 01 §1)。
一次分析(agent 调一次分析工具,或分析卡里的一项)产出一份**分析结论**:内容寻址存成工件(ARTIFACT_KIND),证据台账里记一条
LEDGER_KIND,写作目录编号 A1、A2……。这里只定形状、名字、缺省值和几个纯函数;算子在 app/analysis 的各模块里,入口(agent
工具、分析卡)、证据接入、说法核对都按这里写,不各自另起一套。
- **纯函数**:结论只由输入快照里的格、probes(现查的表范围)、params、settings、method 决定;证据面板、导出、可验证报告用
recompute 按同一个函数复算,逐字段相同(same)才算核对通过。
- **模型只选算什么**:AnalysisRequest 里没有阈值;阈值只来自设置快照(run.options["analysis"])和分析卡上写的覆盖。
- **算不了是一条结论**(status refused + reason_code),调用写错是 CallError(不记结论,把 message 交回 agent 改)。
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Mapping
from datetime import date
from typing import Any, Literal, TypedDict
# --------------------------------------------------------------------------
# 名字(一种说法:别处一律引用这里,不另写字面量)
# --------------------------------------------------------------------------
#: 分析结论的工件种类
ARTIFACT_KIND = "analysis_result"
#: 证据台账里的条目种类(engine/evidence.EvidenceEntry.kind)
LEDGER_KIND = "analysis"
#: 写作目录编号的前缀:A1、A2……(和 Q、K 同一套按台账先后编号的规则)
CATALOG_PREFIX = "A"
#: 引用标记的种类:[[a:A3]]、[[a:A3.top_share|pct]]
MARKER_KIND = "a"
#: 分析卡的节点类型
NODE_TYPE = "analysis"
#: 设置分组的键:settings 里的 analysis 一组、run.options["analysis"]
SETTINGS_KEY = "analysis"
Op = Literal["decompose", "baseline", "trend", "differ", "comparable"]
OPS: tuple[str, ...] = ("decompose", "baseline", "trend", "differ", "comparable")
#: 有模式的算子和它们的模式;其余算子 mode 为 None
MODES: dict[str, tuple[str, ...]] = {
"decompose": ("additive", "ratio", "factor"),
"differ": ("rate", "mean", "rows"),
}
#: 每个算子(和模式)的方法名与版本:改算法就升版本号,结论里记下实际用的那一版
METHODS: dict[tuple[str, str | None], str] = {
("decompose", "additive"): "decompose.additive@1",
("decompose", "ratio"): "decompose.ratio@1",
("decompose", "factor"): "decompose.factor@1",
("baseline", None): "baseline.robust@1",
("trend", None): "trend.theil_sen_mk@1",
("differ", "rate"): "differ.rate@1",
("differ", "mean"): "differ.welch@1",
("differ", "rows"): "differ.welch_rows@1",
("comparable", None): "comparable@1",
}
#: agent 分析工具的名字(tools/builtin 里注册),一个算子一个工具:参数各自聚焦,模型少写错
TOOL_NAMES: dict[str, str] = {op: f"analysis_{op}" for op in OPS}
#: 每个算子可能给出的判定(细案 01 §2–§6)
VERDICTS: dict[str, tuple[str, ...]] = {
"decompose": ("flat", "driver", "concentrated", "dispersed", "unclosed", "mix_driven", "rate_driven", "both"),
"baseline": ("within_band", "outside_band", "far_outside_band"),
"trend": ("rising", "falling", "no_trend"),
"differ": ("differs", "no_evidence"),
"comparable": ("comparable", "conditional", "incomparable"),
}
#: 算不了的原因(细案 01 §8.2)。算不了也记成结论:写作者可以照实写「历史只有 5 周,判断不了是否异常」
REASONS: tuple[str, ...] = (
"truncated", "too_few_segments", "history_short", "partial_target", "target_gap", "series_gap",
"period_unreadable", "zero_base", "non_positive", "incomparable_sides",
)
#: 附带说明的种类(caveats[].code)
CAVEATS: tuple[str, ...] = (
"offset", "simpson", "small_denominator", "record_high", "record_low", "weekday_effect",
"multiple_comparisons", "outlier_sensitive", "small_sample", "measure_unlabeled", "null_values",
"headline_basis", "comparability",
)
# --------------------------------------------------------------------------
# 设置(细案 01 §10):键、缺省、取值范围。api/settings.py 的 analysis 一组、发起时的快照、分析卡上的覆盖都按这张表
# --------------------------------------------------------------------------
DEFAULTS: dict[str, Any] = {
"main_share": 0.5,
"concentration": 0.8,
"offset_ratio": 0.25,
"flat_pct": 0.005,
"k_warn": 2.0,
"k_alert": 3.0,
"min_history": 8,
"max_history": 52,
"trend_min_points": 6,
"trend_alpha": 0.05,
"differ_alpha": 0.05,
"min_n": 30,
"min_expected": 5,
"min_denominator": 30,
"max_calls": 12,
"max_drill_depth": 3,
"table_top_k": 10,
}
#: 每个键的类型和闭区间。整数键只收整数(布尔不算)
RANGES: dict[str, tuple[type, float, float]] = {
"main_share": (float, 0.3, 0.9),
"concentration": (float, 0.5, 0.99),
"offset_ratio": (float, 0.05, 1.0),
"flat_pct": (float, 0.0, 0.1),
"k_warn": (float, 1.0, 5.0),
"k_alert": (float, 1.5, 8.0),
"min_history": (int, 4, 104),
"max_history": (int, 8, 260),
"trend_min_points": (int, 4, 104),
"trend_alpha": (float, 0.001, 0.2),
"differ_alpha": (float, 0.001, 0.2),
"min_n": (int, 2, 10_000),
"min_expected": (int, 1, 50),
"min_denominator": (int, 1, 100_000),
"max_calls": (int, 1, 50),
"max_drill_depth": (int, 1, 6),
"table_top_k": (int, 3, 30),
}
#: 每个键给人看的叫法(设置页、报错都用它,句子里不裸写键名)
LABELS: dict[str, str] = {
"main_share": "「主要来源」的占比",
"concentration": "「集中」的合计占比",
"offset_ratio": "记为「抵消」的比例",
"flat_pct": "「基本持平」的幅度",
"k_warn": "「超出常规波动」的倍数",
"k_alert": "「明显超出」的倍数",
"min_history": "至少几期历史",
"max_history": "最多用几期历史",
"trend_min_points": "趋势至少几期",
"trend_alpha": "趋势的显著性水平",
"differ_alpha": "差异检验的显著性水平",
"min_n": "均值比较的最小样本数",
"min_expected": "比率比较的最小期望频数",
"min_denominator": "比率拆解的最小分母",
"max_calls": "一次运行最多分析几次",
"max_drill_depth": "最多下钻几层",
"table_top_k": "列出前几段",
}
def settings_problem(values: Mapping[str, Any]) -> str | None:
"""一组分析设置(或分析卡上的覆盖)哪里不对:不认识的键、类型不对、超出范围、k_alert 不大于 k_warn、
max_history 小于 min_history。都对返回 None。说法给人看。"""
for key, value in values.items():
if key not in RANGES:
return f"不认识的分析设置:{key}"
kind, low, high = RANGES[key]
label = LABELS[key]
if isinstance(value, bool) or not isinstance(value, (int, float)):
return f"{label}要填数字"
if kind is int and not isinstance(value, int):
return f"{label}要填整数"
if not low <= value <= high:
return f"{label}要在 {low:g} 到 {high:g} 之间"
merged = {**DEFAULTS, **values}
if merged["k_alert"] <= merged["k_warn"]:
return f"{LABELS['k_alert']}要大于{LABELS['k_warn']}"
if merged["max_history"] < merged["min_history"]:
return f"{LABELS['max_history']}不能小于{LABELS['min_history']}"
return None
def effective_settings(snapshot: Mapping[str, Any] | None, overrides: Mapping[str, Any] | None = None) -> dict[str, Any]:
"""这次分析实际用的阈值:缺省 ← 发起时的快照 ← 分析卡上的覆盖(agent 工具没有覆盖)。"""
out = dict(DEFAULTS)
out.update({k: v for k, v in (snapshot or {}).items() if k in DEFAULTS})
out.update({k: v for k, v in (overrides or {}).items() if k in DEFAULTS})
return out
# --------------------------------------------------------------------------
# 形状。都是普通 dict,TypedDict 只用来写清形状(同 engine/evidence.py 的写法)
# --------------------------------------------------------------------------
class InputSnapshot(TypedDict, total=False):
"""算子的一份输入:本次运行里交回过的一份查询快照(tools/datasource 的 to_payload 加上出处)。"""
ref: str # 调用里写的名字:agent 用本节点的编号(Q3),分析卡用来源节点 id
artifact: str # 查询快照工件的 sha256(台账里 kind query 那条的 artifact)
node_id: str # 交回这份快照的节点
source: str # 数据源名
sql: str
columns: list[str]
column_types: list[str | None]
rows: list[list[Any]]
truncated: bool
masked: list[str] # 不能当维度、度量的列(遮罩列、按 SQL 认出改了名字的遮罩列),比较时不分大小写
class AnalysisInput(TypedDict, total=False):
"""结论里记下的一份输入:用了哪份快照的哪些列、哪些行,或者哪一格。复算按 artifact 取回快照。"""
ref: str # 调用里写的名字(Q3、Q1.r0.gmv、A3)
artifact: str
role: str # data(主输入)、base / target(两份快照)、headline_base / headline_target、parent
columns: list[str]
rows: Literal["all"] | list[int]
row: int # 一格时:第几行、哪一列、值(十进制字符串)
column: str
value: str | None
class Probe(TypedDict, total=False):
"""可比性检查现查的一张表的一列的实际范围(data/coverage 的探查;失败退回剖析、取值观察)。复算用记下的值。"""
source: str
table: str
column: str
min: str | None # ISO 日期
max: str | None
basis: Literal["live", "profile", "observed"]
at: str # 现查的时刻(ISO,服务端时区)
class Verdict(TypedDict, total=False):
code: str # VERDICTS[op] 之一
subjects: list[list[str]] # 点名的段(维度键,一段一个列表,多维时几个值)或因子名;没有点名时为空
direction: Literal["up", "down"] | None
class Caveat(TypedDict):
code: str # CAVEATS 之一
text: str # 渲染好的一句(一种说法)
class ResultTable(TypedDict):
"""结果表:形状同查询快照(columns / rows),[[v:A3.r0.share]]、[[table:A3 …]] 按它取格。"""
columns: list[str] # 稳定的列 id(维度列照输入的列名,其余是 base、target、delta、share……)
labels: dict[str, str] # 列 id → 给人看的列名(基期、本期、变化、占变化的比重……)
rows: list[list[Any]] # 数一律十进制字符串或 None,维度值照原样
class AnalysisRequest(TypedDict, total=False):
"""一次分析的请求:只有「算什么」。agent 工具的参数、分析卡的一项都规范成这个形状。"""
op: str
mode: str | None
params: dict[str, Any] # 各算子的参数(细案 01 §2–§6):query / by / period / base / target / measure / numerator /
# denominator / factors / product / headline / parent / target / basis / grain / lag / label……
hypothesis: str # agent 写的这次在验证什么,只在证据面板显示,不进任何系统渲染的正文
class AnalysisResult(TypedDict, total=False):
op: str
mode: str | None
method: str # METHODS 里的一项
params: dict[str, Any] # 规范化后的参数
settings: dict[str, Any] # 实际用的阈值(effective_settings)
inputs: list[AnalysisInput]
probes: list[Probe]
status: Literal["ok", "refused"]
reason_code: str | None # REASONS 之一(refused 时)
reason: str | None
table: ResultTable | None
values: dict[str, Any] # 具名标量:[[a:A3.top_share]];数一律十进制字符串
verdict: Verdict | None
criterion: str # 判定标准,一句,照实际阈值写
phrase: str # 判定语(app/analysis/phrases 渲染),[[a:A3]] 渲染成它
caveats: list[Caveat]
hypothesis: str | None
class CallError(Exception):
"""调用错误(细案 01 §8.1):列不存在、列被遮罩、角色和目录度量类型冲突、同一键同一期多行、期次不在列里、乘积
不闭合、parent 不对、下钻超深、本运行分析次数用完。不记结论,message 交回 agent(一种说法,告诉它怎么改)。"""
def __init__(self, code: str, message: str) -> None:
super().__init__(message)
self.code = code
self.message = message
#: 现查一列的实际范围:(数据源名, 表, 列) → Probe;查不了返回 None(算子写「核对不了」,不判不可比)
Prober = Callable[[str, str, str], Awaitable[Probe | None]]
class Context(TypedDict, total=False):
"""入口交给算子的运行期依赖。复算不用它(probes 已记在结论里)。"""
today: date # 运行当天(服务端时区):判本期完不完整
probe: Prober | None # None 时只用 facts
catalogs: Mapping[str, Any] # 数据源名 → 数据目录(app/data/catalog 的 notes):认度量类型、业务日期列、列的叫法
facts: Mapping[str, Any] # 数据源名 → 剖析事实(探查失败时兜底)
# --------------------------------------------------------------------------
# 台账条目
# --------------------------------------------------------------------------
def ledger_entry(result: AnalysisResult, *, node_id: str, exec_no: int, artifact: str,
call_id: str | None = None, item: str | None = None) -> dict[str, Any]:
"""证据台账里的一条分析结论(engine/evidence.EvidenceEntry 的 kind analysis)。agent 工具带 call_id,分析卡带 item
(那一项的 id);inputs 是输入快照的 sha256(去重、保序),证据图据此画「派生自」的边,写作目录据此写「来自 Q3」。"""
entry: dict[str, Any] = {
"kind": LEDGER_KIND,
"node_id": node_id,
"exec": exec_no,
"artifact": artifact,
"op": result.get("op"),
"mode": result.get("mode"),
"status": result.get("status"),
"verdict": (result.get("verdict") or {}).get("code"),
"inputs": list(dict.fromkeys(i["artifact"] for i in result.get("inputs") or [] if i.get("artifact"))),
}
if call_id is not None:
entry["call_id"] = call_id
if item is not None:
entry["item"] = item
return entry
#: 复算核对时不比的字段:hypothesis 是 agent 写的话,不是算出来的
NOT_RECOMPUTED = frozenset({"hypothesis"})
def same(a: Mapping[str, Any], b: Mapping[str, Any]) -> bool:
"""两份结论(封存的、复算的)是否逐字段相同(hypothesis 除外)。"""
keys = (set(a) | set(b)) - NOT_RECOMPUTED
return all(a.get(k) == b.get(k) for k in keys)
__all__ = [
"ARTIFACT_KIND", "CATALOG_PREFIX", "CAVEATS", "DEFAULTS", "LABELS", "LEDGER_KIND", "MARKER_KIND", "METHODS", "MODES",
"NODE_TYPE", "NOT_RECOMPUTED", "OPS", "RANGES", "REASONS", "SETTINGS_KEY", "TOOL_NAMES", "VERDICTS",
"AnalysisInput", "AnalysisRequest", "AnalysisResult", "CallError", "Caveat", "Context", "InputSnapshot", "Probe",
"Prober", "ResultTable", "Verdict", "effective_settings", "ledger_entry", "same", "settings_problem",
]
+66
View File
@@ -0,0 +1,66 @@
"""测试用的分析结论(只给测试用):第 1 波各包并行开发时,证据接入、入口、说法核对的测试不等算子写完,按契约的形状
拿一份假的结论。算子落地后,端到端的测试改用 analysis.run 真算。"""
from __future__ import annotations
from typing import Any
from app.analysis.contract import METHODS, AnalysisResult, effective_settings
_PHRASES = {
"driver": "销售额下降 194 万(-8.7%),主要来自华东(-120 万,占 62%)",
"concentrated": "销售额下降 194 万(-8.7%),集中在华东、华北(合计占 85%)",
"dispersed": "销售额下降 194 万(-8.7%),分散在各地区,没有哪一段占到一半",
"within_band": "销售额本期变化 -3.1%,在过去 12 周的常规波动范围内(-6.1% 至 +5.3%)",
"outside_band": "销售额本期变化 -8.7%,超出过去 12 周的常规波动范围(-6.1% 至 +5.3%)",
"rising": "过去 10 周线上销售额呈上升趋势(每周约 +3.0%)",
"no_evidence": "西南和华东的退款率差异可能只是样本波动(样本量 40、230)",
"comparable": "两期可比",
"incomparable": "两期不可比:本期只有 3 天数据",
}
def fake_result(op: str = "decompose", mode: str | None = "additive", verdict: str | None = "driver", *,
subjects: list[list[str]] | None = None, direction: str | None = "down", status: str = "ok",
reason_code: str | None = None, inputs: list[dict[str, Any]] | None = None,
**overrides: Any) -> AnalysisResult:
"""一份形状完整的假结论。缺省是「销售额下降主要来自华东」的可加拆解。"""
if op not in ("decompose", "differ"):
mode = None
table = {
"columns": ["region", "base", "target", "delta", "pct", "share", "flag"],
"labels": {"region": "地区", "base": "基期", "target": "本期", "delta": "变化", "pct": "变化率",
"share": "占变化的比重", "flag": "说明"},
"rows": [["华东", "820", "700", "-120", "-0.1463", "0.6186", None],
["华北", "600", "560", "-40", "-0.0667", "0.2062", None],
["华南", "700", "666", "-34", "-0.0486", "0.1753", None],
["西南", "100", "100", "0", "0", "0", None]],
}
result: AnalysisResult = {
"op": op,
"mode": mode,
"method": METHODS.get((op, mode), f"{op}@1"),
"params": {"query": "Q2", "by": ["region"], "period": "week", "base": "W41", "target": "W42",
"measure": "gmv", "label": "销售额"},
"settings": effective_settings(None),
"inputs": inputs if inputs is not None else [
{"ref": "Q2", "artifact": "0" * 64, "role": "data", "columns": ["region", "week", "gmv"], "rows": "all"}],
"probes": [],
"status": status, # type: ignore[typeddict-item]
"reason_code": reason_code,
"reason": None if status == "ok" else "历史只有 5 期,至少要 8 期",
"table": table if status == "ok" else None,
"values": {"base_total": "2220", "target_total": "2026", "delta_total": "-194", "pct_total": "-0.0874",
"top_key": "华东", "top_share": "0.6186", "segments": "4"} if status == "ok" else {},
"verdict": ({"code": verdict, "subjects": subjects if subjects is not None else
([["华东"]] if verdict == "driver" else []), "direction": direction} # type: ignore[typeddict-item]
if status == "ok" and verdict else None),
"criterion": "一段占总变化一半以上算主要来源",
"phrase": _PHRASES.get(verdict or "", "") if status == "ok" else "变化拆解没有做成:历史只有 5 期,至少要 8 期",
"caveats": [],
"hypothesis": "地区拖累",
}
result.update(overrides) # type: ignore[typeddict-item]
return result
__all__ = ["fake_result"]
+51
View File
@@ -0,0 +1,51 @@
"""分析结论的契约(app/analysis/contract.py):设置的校验与合并、台账条目、复算比较。"""
from __future__ import annotations
from app.analysis import contract as c
from app.analysis.testing import fake_result
def test_every_setting_has_a_default_a_range_and_a_label():
assert set(c.DEFAULTS) == set(c.RANGES) == set(c.LABELS)
assert c.settings_problem(c.DEFAULTS) is None
for key, (kind, low, high) in c.RANGES.items():
assert low <= c.DEFAULTS[key] <= high, key
assert isinstance(c.DEFAULTS[key], kind), key
def test_settings_problems_are_spoken_with_labels_not_keys():
assert c.settings_problem({"k_warn": 3.0, "k_alert": 2.5}) == "「明显超出」的倍数要大于「超出常规波动」的倍数"
assert c.settings_problem({"min_history": 8.5}) == "至少几期历史要填整数"
assert c.settings_problem({"main_share": True}) == "「主要来源」的占比要填数字"
assert c.settings_problem({"main_share": 0.95}) == "「主要来源」的占比要在 0.3 到 0.9 之间"
assert c.settings_problem({"max_history": 8, "min_history": 9}) == "最多用几期历史不能小于至少几期历史"
assert c.settings_problem({"nope": 1}) == "不认识的分析设置:nope"
def test_effective_settings_layers_defaults_snapshot_and_card_overrides():
out = c.effective_settings({"k_warn": 2.5, "unknown": 1}, {"k_warn": 1.5, "main_share": 0.6})
assert out["k_warn"] == 1.5 and out["main_share"] == 0.6 and "unknown" not in out
assert c.effective_settings(None) == c.DEFAULTS
def test_the_ledger_entry_names_its_input_snapshots_once():
twice = [{"ref": "Q2", "artifact": "a" * 64, "role": "data"},
{"ref": "Q2.r0.gmv", "artifact": "a" * 64, "role": "headline_target"},
{"ref": "Q1", "artifact": "b" * 64, "role": "base"}]
entry = c.ledger_entry(fake_result(inputs=twice), node_id="n1", exec_no=2, artifact="c" * 64, item="drop")
assert entry == {"kind": "analysis", "node_id": "n1", "exec": 2, "artifact": "c" * 64, "op": "decompose",
"mode": "additive", "status": "ok", "verdict": "driver", "inputs": ["a" * 64, "b" * 64],
"item": "drop"}
def test_recompute_comparison_ignores_only_the_hypothesis():
sealed = fake_result()
assert c.same(sealed, {**sealed, "hypothesis": "别的说法"})
assert not c.same(sealed, {**sealed, "phrase": "改过的判定语"})
def test_every_op_has_a_method_and_a_tool_name():
for op in c.OPS:
modes = c.MODES.get(op, (None,))
assert all((op, m) in c.METHODS for m in modes), op
assert c.TOOL_NAMES[op] == f"analysis_{op}"