diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index c34e943b..5895b195 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -44,6 +44,7 @@ from typing import Any from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage from sqlalchemy import select +from app.analysis.contract import CATALOG_PREFIX, LEDGER_KIND, catalog_result, is_result, method_label from app.core.artifact_store import canonical_json, content_hash from app.core.artifact_store import load as load_artifact from app.core.errors import describe_exception, not_configured @@ -160,6 +161,15 @@ JUDGE_RULES = f"""判定（verdict）五选一： 7. used 写你实际用到的证据编号（摘录里【】中的那个，比如 m:gmv、Q1），没用到就写空数组。 8. 每句都要给判定，unit 照抄句子前面方括号里的编号。""" +#: 摘录里有分析结论（【A1】）时接在裁判规则后面的一条。只在这时才加：没有分析结论的报告，裁判看到的规则一字不差， +#: 已判过的句子照旧复用（复用键里只有摘录，没有规则） +ANALYSIS_JUDGE_PROMPT = ( + "9. 【A1】这类分析结论的判定、判定标准和关键数是系统按平台的方法算出来的，不要质疑也不要复算；只判断句子的说法和" + "判定是否一致：「主要来自」「集中在」「超出常规波动」「呈上升趋势」「差异站得住」这类判断要和判定对得上，句子点名的段" + "要在判定里，方向要一致；判定写着算不了的，不能支持任何判断。分析结论只说明变化来自哪里，说明不了原因：句子写了" + "原因、证据里又没有原因的，判 partial 或 insufficient。" +) + #: 报告节点 claims 为 judge 时拼在写作规则后面：写作者要知道还有一道按证据的核对 JUDGE_RULE = ("这份报告的结论句还会由另一个模型逐句核对：它只看你挂的依据，判断这句话站不站得住。" "说法不要超出证据——证据里没有的原因、范围、趋势，不要写成结论。") @@ -299,8 +309,8 @@ def _candidate(unit: dict[str, Any], priority: int, order: int, text: str, secti rows.append((alias, int(locator["row"]))) if kind == "cell" and locator.get("column") not in (None, ""): cols.append((alias, str(locator["column"]))) - elif kind in ("query", "retrieval"): - whole.append(alias) + elif kind in ("query", "retrieval", LEDGER_KIND): + whole.append(alias) # 分析结论：判定语、具名标量、依据 [[see:A1]] 都整份摘 elif kind == "quote" and isinstance(locator.get("hit"), int): hits.append((alias, int(locator["hit"]))) return Candidate(unit=str(unit.get("id")), priority=priority, order=order, text=text, section=section, @@ -664,6 +674,102 @@ def _retrieval_excerpt(alias: str, entry: dict[str, Any], picked: list[int], who return "\n".join(lines) +def _analysis_alias(alias: Any) -> bool: + """写作目录里分析结论的编号（A1、A2……）。""" + return isinstance(alias, str) and alias.startswith(CATALOG_PREFIX) and alias[len(CATALOG_PREFIX):].isdigit() + + +#: 分析结论的摘录最多列几个具名的数 +ANALYSIS_VALUES = 16 + + +def _analysis_excerpt(alias: str, entry: dict[str, Any], rows: set[int], fetch: _Fetch, + masked: dict[str, Any] | None, catalog: dict[str, Any], mask: Any = None) -> str | None: + """一条分析结论的摘录：方法、判定语、判定标准、关键数（具名标量）、附带说明、结果表（被引用的行先给，再从头补满 + EXCERPT_ROWS 行，最多 EXCERPT_COLS 列，表头写给人看的列名）、输入取自哪几次查询、用了哪几列，以及那几次查询的 SQL + （截到 SQL_CHARS，最多 ENTITY_QUERIES 次）。判定是系统算的，裁判只判句子的说法和判定是否一致（ANALYSIS_JUDGE_PROMPT）。 + + 结论按工件取（经 fetch：按需裁判时只认封存范围内的工件，取不到、哈希不符的不给这段摘录）；目录条目带着整份结论的 + 直接用。agent 写的假设不给：它不是结论。mask 同 _query_excerpt（核查 agent 的摘录才给）：SQL 先遮再截，结果表里的文字、 + 判定语按值换，数照原样。""" + result = catalog_result(entry) + if result is None: + content = fetch(entry.get("artifact")) + result = content if is_result(content) else None + if result is None: + return None + scrub = mask.text if mask is not None else (lambda t: t) + lines = [f"【{alias}】分析结论（{method_label(result.get('op'), result.get('mode'))}，方法 {result.get('method') or '—'}）"] + if result.get("status") == "refused": + lines.append(f"判定：算不了（{_clean(str(result.get('reason') or '没有写原因'))}）") + else: + lines.append(f"判定：{scrub(_clean(str(result.get('phrase') or ''))) or '—'}") + if criterion := _clean(str(result.get("criterion") or "")): + lines.append(f"判定标准：{criterion}") + values = result.get("values") if isinstance(result.get("values"), dict) else {} + if values: + shown = [] + for key, raw in list(values.items())[:ANALYSIS_VALUES]: + try: + text = render_cell(raw) + except RenderError: + text = MISSING + shown.append(f"{key}={text if isinstance(raw, (int, float)) else scrub(text)}") + more = f" | …另有 {len(values) - ANALYSIS_VALUES} 个" if len(values) > ANALYSIS_VALUES else "" + lines.append("关键数：" + " | ".join(shown) + more) + caveats = [_clean(str(c.get("text") or "")) for c in result.get("caveats") or [] if isinstance(c, dict)] + if caveats := [c for c in caveats if c]: + lines.append("附带说明：" + "；".join(scrub(c) for c in caveats)) + table = result.get("table") if isinstance(result.get("table"), dict) else None + if table is not None and isinstance(table.get("rows"), list) and table["rows"]: + columns = [str(c) for c in table.get("columns") or []][:EXCERPT_COLS] + labels = table.get("labels") if isinstance(table.get("labels"), dict) else {} + total = len(table["rows"]) + picked = sorted(r for r in rows if 0 <= r < total)[:EXCERPT_ROWS] + picked += [r for r in range(total) if r not in picked][:EXCERPT_ROWS - len(picked)] + lines.append(f"结果表（共 {total} 行）列：" + " | ".join(_clean(str(labels.get(c) or c)) for c in columns)) + for r in sorted(picked): + cells = [] + for c in columns: + try: + raw, _ = locate_cell(table, r, c) + text = render_cell(raw) + cells.append(text if isinstance(raw, (int, float)) else scrub(text)) + except (CellError, RenderError): + cells.append(MISSING) + lines.append(f"r{r}：" + " | ".join(cells)) + sqls: list[str] = [] + seen: set[str] = set() + for item in result.get("inputs") or []: + if not isinstance(item, dict) or not item.get("artifact") or item.get("role") == "parent": + continue + artifact = str(item["artifact"]) + if artifact in seen: + continue + seen.add(artifact) + query = _query_alias(catalog, artifact) + snap = fetch(artifact) + snap = snap if isinstance(snap, dict) else {} + source = (catalog.get(query) or {}).get("source") if query else None + hidden, names = _derived(snap, _source_hidden(source or snap.get("source"), catalog, fetch, masked) + | _mask_names(snap.get("mask_columns")), fetch, mask) + used = [str(c) for c in item.get("columns") or [] if str(c).lower() not in hidden] + label = f"查询 {query}" if query else "一次查询（不在这份报告的证据目录中）" + lines.append(f"输入：取自{label}" + (f"，用到 {'、'.join(used)} 列" if used else "")) + sql = _clean(str(snap.get("sql") or "")) + if sql and mask is not None: + sql = mask.sql(sql, snap, names) + if sql and len(sqls) < ENTITY_QUERIES: + sqls.append(_sql_line(f"查询 {query} 的 SQL" if query else "这次查询的 SQL", sql)) + parents = [str(i.get("artifact")) for i in result.get("inputs") or [] + if isinstance(i, dict) and i.get("role") == "parent"] + for artifact in parents: + found = next((a for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") == LEDGER_KIND + and e.get("artifact") == artifact), None) + lines.append(f"下钻自：{found or '一条分析结论（不在这份报告的证据目录中）'}") + return "\n".join(lines + sqls) + + def _excerpt(alias: str, entry: dict[str, Any], cands: list[Candidate], fetch: _Fetch, masked: dict[str, Any] | None, catalog: dict[str, Any], *, provenance_on: bool = False, mask: Any = None) -> str | None: @@ -683,6 +789,9 @@ def _excerpt(alias: str, entry: dict[str, Any], cands: list[Candidate], fetch: _ if kind == "retrieval": picked = [h for c in cands for a, h in c.hits if a == alias] return _retrieval_excerpt(alias, entry, picked, any(alias in c.whole for c in cands), fetch, mask) + if kind == LEDGER_KIND: + rows = {r for c in cands for a, r in c.rows if a == alias} + return _analysis_excerpt(alias, entry, rows, fetch, masked, catalog, mask) if kind == "table": head = f"【{alias}】表 {entry.get('name')}" + (f"（数据源 {entry['source']}）" if entry.get("source") else "") return "\n".join([head, *_table_lines(entry, catalog, fetch, masked), @@ -754,8 +863,10 @@ class JudgeRequest: aliases = list(dict.fromkeys(a for c in units for a in c.aliases if a in self.excerpts)) evidence = "\n\n".join(self.excerpts[a] for a in aliases) or "（这些句子都没有挂依据）" ordered = sorted(units, key=lambda c: c.order) + # 摘录里有分析结论时另加一条（ANALYSIS_JUDGE_PROMPT）；没有的和以前一字不差 + rules = f"{JUDGE_RULES}\n{ANALYSIS_JUDGE_PROMPT}" if any(_analysis_alias(a) for a in aliases) else JUDGE_RULES return [ - SystemMessage(content=f"{JUDGE_ROLE}\n\n{JUDGE_RULES}"), + SystemMessage(content=f"{JUDGE_ROLE}\n\n{rules}"), HumanMessage(content=( f"证据摘录（【】里是编号，判定时 used 写它）：\n{evidence}\n\n" "要判断的句子（每句都要给判定）：\n" + "\n".join(self.unit_line(c) for c in ordered) + "\n\n" diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py index f53861e4..71e1b447 100644 --- a/backend/app/ledger/labels.py +++ b/backend/app/ledger/labels.py @@ -20,6 +20,9 @@ 整题（whole_of）：主报告里所有带标签的数都来自同一口径的同一版、绑定一致，就记 equivalent，并写出还缺的入口字段（口径入口 节点声明了、这次的绑定里没有的；required 且没有缺省值的另列）。 +分析结论里的数（[[a:A3]]、[[a:A3.键]]、[[v:A3.r0.列]]、整表 [[table:A3]] 展开的格）不标：它们是从查询快照派生出来的， +不是口径（docs/design/analysis-round/00-总方案.md D11），不和口径比对、不进整题判断的计数，也不算成「没比对」。 + 这里只是纯函数：标签在比对时算好、存进比对工件（ledger/match.py）；接口读的时候按这次运行查到过的遮罩值遮一遍（view_label）： 绑定的值是遮罩值的不给值，差异条目里的 SQL 片段走证据面板同一套遮罩（api/evidence._Shown）。 """ @@ -28,6 +31,8 @@ from __future__ import annotations from collections.abc import Callable, Mapping from typing import Any +from app.analysis.contract import LEDGER_KIND + #: 标签的四种、认法的两种 KINDS = ("caliber", "related", "reasoning", "unknown") HOWS = ("independent", "composed") @@ -259,6 +264,8 @@ def referenced_columns(docs: list[Mapping[str, Any]], if cite.get("status") != "resolved": continue entry = catalog.get(cite.get("alias")) if isinstance(catalog.get(cite.get("alias")), dict) else {} + if entry.get("kind") == LEDGER_KIND: + continue # 分析结论的结果表不是查询，不比对 locator = cite.get("locator") if isinstance(cite.get("locator"), dict) else {} artifact = str(entry.get("artifact") or "") if cite.get("kind") == "cell" and artifact and isinstance(locator.get("column"), str): @@ -375,7 +382,7 @@ def label_doc(doc: Mapping[str, Any], queries: Mapping[str, Mapping[str, Any]], calibers: Mapping[str, Mapping[str, Any]], load_card: Callable[[str], Mapping[str, Any] | None]) -> dict[str, dict[str, Any]]: """一份报告文档里每个数的标签：{片段 id: 标签}。只标引用解析了的数字、取值片段（单元格、指标）；运行输入、表名、引文、 - 没有引用的数不标。""" + 没有引用的数不标，分析结论里的数（具名标量、结果表的格）也不标（D11：派生的数，不是口径）。""" from app.engine.evidence import iter_units catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} @@ -389,6 +396,8 @@ def label_doc(doc: Mapping[str, Any], queries: Mapping[str, Mapping[str, Any]], if cite.get("status") != "resolved": continue entry = catalog.get(cite.get("alias")) if isinstance(catalog.get(cite.get("alias")), dict) else {} + if entry.get("kind") == LEDGER_KIND: + continue locator = cite.get("locator") if isinstance(cite.get("locator"), dict) else {} artifact = str(entry.get("artifact") or "") if cite.get("kind") == "cell" and artifact: diff --git a/backend/tests/test_analysis_judge_ledger.py b/backend/tests/test_analysis_judge_ledger.py new file mode 100644 index 00000000..c3db6bc5 --- /dev/null +++ b/backend/tests/test_analysis_judge_ledger.py @@ -0,0 +1,99 @@ +"""分析结论进证据链（第 1 波 W1-b）：结论句裁判的摘录、口径账本。 + +- 裁判：引用了分析结论的句子，摘录里有方法、判定、判定标准、关键数、结果表和输入查询的 SQL；判定是系统算的，规则里另加 + 一条（只判句子说法和判定一致不一致）——没有分析结论的报告，裁判看到的规则一字不差 +- 口径账本：分析结论里的数（判定语、具名标量、结果表的格）不标口径、不比对、不进整题判断的计数（D11） +""" +from __future__ import annotations + +from typing import Any + +from app.analysis.testing import fake_result +from app.engine import judge as judging +from app.engine.evidence import build_catalog, compose_doc, iter_segments +from app.ledger import labels as ledger_labels +from tests.test_analysis_evidence import setup + + +def composed(text: str, **kw: Any) -> tuple[Any, dict[str, Any], dict[str, Any]]: + store, ledger, _, _ = setup(**kw) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + return store, catalog, compose_doc(text, catalog, loader=store.load) + + +# -------------------------------------------------------------------------- +# 裁判 +# -------------------------------------------------------------------------- + + +def test_the_judge_gets_the_method_the_judgment_and_the_input_sql(): + store, catalog, doc = composed("[[a:A1]]，华东变化 [[v:A1.r0.delta]]。[[see:A1]]") + request = judging.prepare(doc, catalog, loader=store.load) + [cand] = request.cands + assert cand.aliases == ("A1",) and "A1" in cand.whole and ("A1", 0) in cand.rows and ("A1", "delta") in cand.cols + excerpt = request.excerpts["A1"] + assert excerpt.startswith("【A1】分析结论（变化拆解（可加），方法 decompose.additive@1）") + assert "判定：销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）" in excerpt + assert "判定标准：一段占总变化一半以上算主要来源" in excerpt + assert "关键数：base_total=2,220 | target_total=2,026 | delta_total=-194" in excerpt + assert "结果表（共 4 行）列：地区 | 基期 | 本期 | 变化" in excerpt and "r0：华东 | 820 | 700 | -120" in excerpt + assert "输入：取自查询 Q1，用到 region、week、gmv 列" in excerpt + assert "查询 Q1 的 SQL：SELECT region, week, SUM(gmv) AS gmv FROM orders" in excerpt + assert "地区拖累" not in excerpt # agent 写的假设不进摘录 + + system = request.messages(request.cands)[0].content + assert judging.JUDGE_RULES in system and judging.ANALYSIS_JUDGE_PROMPT in system + + +def test_reports_without_analyses_see_exactly_the_old_rules(): + store, catalog, doc = composed("华东本周 [[v:Q1.r1.gmv]]。[[see:Q1]]") + request = judging.prepare(doc, catalog, loader=store.load) + system = request.messages(request.cands)[0].content + assert system == f"{judging.JUDGE_ROLE}\n\n{judging.JUDGE_RULES}" + + +def test_excerpts_reload_the_result_from_the_stored_summary_and_say_refusals(): + store, catalog, doc = composed("下降来自哪里还说不清。[[see:A1]]", + result=fake_result(status="refused", reason_code="too_few_segments")) + request = judging.prepare(doc, doc["catalog"], loader=store.load) # 文档里的目录只留摘要：按工件取回 + assert "判定：算不了（历史只有 5 期，至少要 8 期）" in request.excerpts["A1"] + gone = judging.prepare(doc, doc["catalog"], loader=lambda _: None) + assert "A1" not in gone.excerpts + + +def test_drill_downs_name_the_parent_in_the_excerpt(): + store, ledger, q, a = setup() + from app.analysis.contract import ledger_entry + + child = fake_result(inputs=[{"ref": "Q1", "artifact": q, "role": "data", "columns": ["region"]}, + {"ref": "A3", "artifact": a, "role": "parent"}]) + ledger.append(ledger_entry(child, node_id="an", exec_no=2, artifact=store.put(child), item="drill")) + catalog = build_catalog(nodes={}, ledger=ledger, loader=store.load) + doc = compose_doc("华东里主要来自上海。[[see:A2]]", catalog, loader=store.load) + excerpt = judging.prepare(doc, catalog, loader=store.load).excerpts["A2"] + assert "下钻自：A1" in excerpt + + +# -------------------------------------------------------------------------- +# 口径账本：分析结论的数不标口径 +# -------------------------------------------------------------------------- + + +def test_analysis_numbers_get_no_caliber_label_and_are_not_counted(): + store, catalog, doc = composed("[[a:A1]]，华东占 [[a:A1.top_share|pct]]，变化 [[v:A1.r0.delta]]；" + "华东上周 [[v:Q1.r0.gmv]]。") + q = catalog["Q1"]["artifact"] + queries = {q: {"artifact": q, "status": "none", "candidates": []}} + labels = ledger_labels.label_doc(doc, queries, {}, lambda _: None) + labelled = {s["id"]: s.get("ref") for s in iter_segments(doc) if s["id"] in labels} + assert list(labelled.values()) == ["v:Q1.r0.gmv"] + whole = ledger_labels.whole_of(labels, {}) + assert whole["counts"] == {"caliber": 0, "related": 0, "reasoning": 1, "unknown": 0} + assert ledger_labels.referenced_columns([doc], lambda _: None) == {q: {"gmv"}} + + +def test_a_report_citing_only_analyses_has_nothing_to_label(): + store, catalog, doc = composed("[[a:A1]]。[[see:A1]]") + assert ledger_labels.label_doc(doc, {}, {}, lambda _: None) == {} + assert ledger_labels.referenced_columns([doc], lambda _: None) == {} + assert ledger_labels.whole_of({}, {})["reason"] == "报告里没有可标口径的数"