diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index fb8b7ade..1f25038b 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -137,6 +137,12 @@ NODE_REFERENCE = """\ 加标识符引号：MySQL / MariaDB 用反引号（AS `月份`），PostgreSQL、Oracle、SQLite 用双引号（AS "月份"），不要用 单引号（那是字符串）。只给最外层 SELECT 的结果列起别名，WHERE、JOIN、GROUP BY 里照旧写原列名；Oracle 旧版本 的标识符最长 30 字节，别名不超过 10 个字 + **问题问的是为什么、怎么变了、哪里拖累、谁贡献最大、是否异常、波动、趋势、是不是真的更高这类分析问题时**，查库的 agent + 除了 db_query__ / db_schema__ 再绑上五个分析工具（analysis_comparable、analysis_baseline、analysis_trend、 + analysis_decompose、analysis_differ）：变化从哪来、算不算异常、是不是持续在变、差异站不站得住由平台按固定的方法和阈值算， + 判定语由系统写好，报告引用它；agent 只决定算什么（比哪两期、拆哪个维度、往哪一段下钻）。绑了任一个，平台就把分析方法 + 注入 agent 的 system，prompt 里不用再教方法，也不要写阈值。分析工具只分析这个 agent 自己查过的结果，所以只绑给同时绑了 + db_query__ 的 agent；不能放进 tool 节点（模板里每次都要做的固定分析用 analysis 分析卡） **默认不要写 max_steps、budget_tokens、budget_usd**。平台用的是护栏而不是固定步数： 步数只是很高的兜底（跟随设置，默认 100），重复调用、连续几步没有新信息、预算用完、 上下文快满时，它会按查到的部分收尾。写死一个小数字（见过 8）会让它查完表结构就没额度回答了。 @@ -215,6 +221,21 @@ NODE_REFERENCE = """\ 写 2 只剩「2%」，0.004 这样的小比率还会显示不出来、报告引用不了。on_missing 默认 fail（缺输入整张卡失败）；写 null 则缺输入的指标记为空值，交给出具契约判档 取 tool 节点查库的结果用 cell(nodes.取数节点id, 行, '列名')（行号从 0 数，列按列名）：cell() 直接认查询工具 交回的结果，出处精确到快照里的那一格，不用再接 transform 解析。取 agent 的字段写 vars..字段 +- analysis：分析卡。对上游查询节点的结果固定做几项确定性分析，和口径卡平级：用于模板里每次都要做的分析（周报的异动： + 这周的变化在不在常规波动范围内、主要来自哪个地区）。问数据里要看了数才知道拆哪个维度的不用它，给查库的 agent 绑分析工具。 + config: {items:[{id, op, mode, from, params, hypothesis}], overrides, assign_to} + · op：decompose（变化拆解，mode 选 additive 可加 / ratio 比率 / factor 乘积）、baseline（波动基线）、trend（趋势）、 + differ（差异检验，mode 选 rate / mean / rows）、comparable（可比性检查）；hypothesis 一句话写这一项在验证什么 + · from 是要分析的上游查询节点 id：只能是调用 db_query__ 的 tool 节点或 merge 节点，而且连在分析卡之前；agent 不能当 + 输入（它可能查了好几次）。基期、本期是两份查询时不写 from，在 params 里写 base_query、target_query 两个节点 id + · params 和同名分析工具的参数一样（period、measure、by、base、target、headline……，不写 query），可以用 {{ input.x }} + 按运行输入选期次；headline 的两格写成「节点id.r行号.列名」（行号从 0 数，如 weekly.r0.销售额）；下钻的 parent 写 + {"of": 前面一项的 id, "key": [那一段的维度值]}。params 里不写阈值：阈值来自设置，模板要固定阈值时写在 overrides + （键同设置里分析一组的键，如 k_warn、main_share） + · 产出 nodes..results.<项 id> = {status, verdict, direction, values, phrase}：下游 branch 可以按判定分支，如 + nodes.check.results.wow.verdict in ['outside_band', 'far_outside_band'] 时转人工。报告引用它的结论（见 report 里分析 + 结论的写法）；不要把结果交给口径卡：口径卡读分析卡的结果追不到原始数据。一项算不了（历史不够、本期没过完）不让卡 + 失败，那一项照实记成算不了 - dual_check：双盲复算。正式出具前，口径卡上输入来自查询的指标另起一个看不到原 SQL 的只读 agent 独立写 SQL 再算一遍： 一致就放行；不一致时停下来，审批卡并排列出两边的 SQL、差异说明和两个值，由人逐个指标选采用原流程、采用盲算还是 两边都不对。选盲算或都不对的计入缺口、出具降档；两边只差一个方面（业务日期按哪一列、有效记录条件、去重）时，选择 @@ -228,10 +249,11 @@ NODE_REFERENCE = """\ 出具降档）/ fail（审核人没采用原流程的结果就判为失败）；enabled 不写跟随设置里的开关，true / false 强制开关。 不要给它配 retries：它的失败是确定的，重试只会把盲算再做一遍、再请审核人选一次（画布校验会报错） - report：报告撰写（带引用）。config: {instructions, system, metrics_from:[口径卡id], numbers, on_violation, max_repairs, - claims, entities, assign_to} + claims, analysis_claims, analysis_scope, entities, assign_to} **凡是给人看的、带数字的报告、分析、结论，都用 report，不要用 llm**。它自动收集上游的口径卡指标、查询结果 - （tool 节点和 agent 查过的库、merge 的合并结果，按查询先后编成 Q1、Q2…）、查库当时的表结构、知识库检索（按先后编成 K1、K2…） - 和运行输入，模型只能写引用标记，数值由系统从证据里取出来渲染；模型自己写的数字会被打回重写： + （tool 节点和 agent 查过的库、merge 的合并结果，按查询先后编成 Q1、Q2…）、查库当时的表结构、知识库检索（按先后编成 K1、K2…）、 + 分析结论（agent 调分析工具、分析卡做的，按先后编成 A1、A2…）和运行输入，模型只能写引用标记，数值由系统从证据里取出来渲染； + 模型自己写的数字会被打回重写： - [[m:指标id]]：口径卡指标。比率、增幅、占比、差值这类派生计算只能先在口径卡里登记再引用。渲染出来自带单位（口径卡里的 unit，「8条」），instructions 里不要写成「覆盖 [[m:line_count]] 条产线」这样后面再跟一个单位，写「覆盖 [[m:line_count]] 产线」 - [[v:Q1.r0.列名]]：第 1 次查询结果里第 0 行那一列的原值（行号从 0 数） @@ -244,6 +266,12 @@ NODE_REFERENCE = """\ 没有出处；只是拿检索当依据时写 [[see:K1]]。要引原话，在 report 前面接 retrieve 节点 - 句末的 [[see:m:指标id,Q1,K1]]：这句结论的依据，不显示，所以只能放在句末；instructions 里不要写「以 [[see:…]] 为依据」 「根据 [[see:…]]」这样把它当成句子成分的示例，模型照着写会留下「以 为依据」这样的断句 + - 分析结论：[[a:A1]] 渲染成 A1 的判定语（系统按判定写好的一句，连同里面的数）；[[a:A1.top_share|pct]] 是它的一个具名的数 + （换算后缀同 [[m:]]）；[[v:A1.r0.share]]、[[table:A1 cols=地区,delta,share rows=0-4]] 取它的结果表；句末的 [[see:A1]] + 把它当这句的依据。「主要来自」「集中在」「拖累」「异常」「大幅」「持续上升」「趋势」「显著高于」这类判断只能照分析结论的 + 判定写、挂它当依据（最好直接引用判定语），没有对应的分析结论就只陈述数字和变化；拆解只说明变化来自哪里、说明不了为什么， + 写原因要有知识库原话或运行输入作依据，否则写成待核实的推测。这些规矩系统会告诉写作者，instructions 里写「按分析结论写、 + 判断引用判定语」就够了，不要替它把判断写好 - 涉及码值（状态、类型这类用数字编码的取值）时，instructions 里让模型写它的含义（「作废记录」），不要写码值本身： 码值是没有出处的数字，会被数字核对打回重写 metrics_from 不写就取所有上游口径卡；on_violation 不写时探索运行是 flag（标出来、照常产出），正式运行是 fail； @@ -260,6 +288,10 @@ NODE_REFERENCE = """\ 不写 judge 的预算（judge 整个不写，金额、句数、时长上限都取设置里的默认）：正式运行在节点里自动判，探索运行由读的人 点开哪句判哪句，不在节点里花钱；报告要交给人工审批的，探索运行也在节点里判。用户明确说不要核对结论句的不加；只回答一个问题、一次查询的不必加。要按受管级别 发布的模板不适用这条默认，按下面受管级别的规则写预算 + analysis_claims 是分析说法核对：check（默认）判断词没挂对分析结论、原因没有依据的句子交回写作者重写一次，仍有的标在 + 报告里，不影响出具档位；strict 按 max_repairs 重写，仍有的计入出具缺口、出具降档；off 不核。只能往严里改 + （off < check < strict），不要为了少重写一次写成 off。analysis_scope：auto（默认：用到分析结论时，系统在报告末尾附一段 + 「分析范围」，写明拆过什么、排除了什么、哪些候选维度没拆）/ off，一般不写 entities 默认 link （核对表名、字段名），报告里根本不提表名时才写 off - output 的出具契约：config.contract = {report_from: 报告节点id, metrics_from:[口径卡id], @@ -268,10 +300,13 @@ NODE_REFERENCE = """\ （[[v:]] / [[table:]]）的，契约里写 cells: true，否则这些引用按解析不了算；只引单元格、图里没有口径卡的 （问数据的典型图），契约不写 metrics_from，也不写 required / expected 要按受管级别发布的模板（发布前检查不过就发不出去）：报告节点写明 numbers: "strict"、on_violation: "fail"、 + analysis_claims: "strict"、 claims: "require_citation"（或者更严的 claims: "judge"，这时 judge 里必须写 judge.max_cost_usd——写一个金额，或者 写 null 表示不限，预算要用户定，拿不准就问；只有受管级别要这样，普通草稿里默认开的 judge 不写预算、不为预算问人）；契约用 report_from（受管级别不收 narrative），成果字段只取报告节点的正文；给口径卡 供数的 agent 配 output_schema 和 cite_fields: true，code 节点喂口径卡的必须是真取数（evidence_role: "source"）， - 否则把计算写进口径卡；llm、agent、协作团队写的文字不能绕过报告节点直接进出口，llm 也不能喂口径卡 + 否则把计算写进口径卡；llm、agent、协作团队写的文字不能绕过报告节点直接进出口，llm 也不能喂口径卡。分析卡的输入要是 + 调用 db_query__ 的 tool 节点或 merge；分析卡的结论总能引用，agent 用分析工具现场做的分析和它查的格同一个规矩（契约 + 写 cells: true 才引用得了） - challenge：反方审稿。报告写完后挑出几句关键结论（挂了依据的、带比较或因果的优先），每句派一个只读 agent 站在反方 自己写查询找反例，逐句判经受住、被削弱、被推翻或无法判断，理由引用它自己查到的数据；不改报告。 config: {report, max_claims, max_queries, timeout_s, provider, model, on_refuted, enabled} @@ -299,6 +334,9 @@ NODE_REFERENCE = """\ 把它当成上周来写。想让报告说清数据最新到哪天，另取一列 MAX(日期) AS 数据最新日期。 写按周期取数的 SQL 前先看表的日期范围（look 结果里的剖析、取值观察）：只有一天的数据、或者最新日期早于要查的区间的表， 不能用来算这段时间（查出来全是 0）——换有历史的表；没有这样的表，就在说明里照实告诉用户这段时间没有数据 +7. 问题问为什么、怎么变了、哪里拖累、谁贡献最大、是否异常、波动、趋势、是不是真的更高这类分析问题时，查库的 agent 绑上 + 分析工具（判定由平台算，报告引用判定语）；模板里每次都要做的固定分析（周报的异动）用分析卡：取数 → 分析卡 → report → output。 + 分析卡的结果不进口径卡，报告直接引用 模型字段（llm / agent / supervisor / report 的 config.model）： - **不要填**。留空表示跟随当前供应商的默认模型，这几乎总是对的。 @@ -365,6 +403,16 @@ class GenerateIn(BaseModel): conversation_id: str | None = None #: 这一轮只查这几个数据源（id 或名字）。不传或空表示不限 datasource_ids: list[str] | None = None + #: 问数据里用户原本问的那句话。自动重来、重试本轮时 instruction 是带着原因的那段话（「上一次运行没有成功：…」）， + #: 自查按问题判分析意图（app/analysis/intent）时用这一句；不传就用 instruction + question: str | None = None + + +def _question_of(payload: GenerateIn) -> str | None: + """自查判分析意图用的那句问题：只在问数据（intent answer）里有，画布上用户说的是一个流程，不是问题。""" + if payload.intent != "answer": + return None + return (payload.question or "").strip() or payload.instruction def _user_message(payload: GenerateIn, *, patch: bool) -> str: @@ -404,6 +452,15 @@ def _user_message(payload: GenerateIn, *, patch: bool) -> str: "「回答：{{ input.question }}」（可以附上 {{ nodes.agent的id.text }} 当分析思路），数用 [[v:Q1.r0.列名]] " "直接引用 agent 查过的单元格，不必经过口径卡；要比率、增幅再在 agent 和 report 之间加口径卡，agent 配 " "output_schema + cite_fields，口径卡读 vars..字段。出口字段写 {{ nodes.report的id.text }}\n" + # 分析能力一轮（00-总方案 §3.7）：问原因、来源、异常、趋势、差异的，判定由平台的分析工具算、系统写判定语，报告引用它。 + # 问题意图由自查按 app/analysis/intent 的词表再核一遍：分析类问题而查库的 agent 没绑分析工具时交回改一轮 + "- 问的是为什么、怎么变了、哪里拖累、谁贡献最大、是否异常、正常吗、波动、趋势、是不是真的更高这类分析问题时，搭 " + "input → agent → report → output：agent 绑 db_query__、db_schema__ 和五个分析工具（analysis_comparable、" + "analysis_baseline、analysis_trend、analysis_decompose、analysis_differ），prompt 里写清要回答的问题、比哪两期" + "（问题没说就比上一个整期），不用教分析方法（平台会注入）、不写阈值；report 的 instructions 写「回答：{{ input.question }}。" + "按分析结论写，判断直接引用判定语；没有分析结论的判断不下，原因没有依据的写成待核实的推测」（可以附上 " + "{{ nodes.agent的id.text }} 当分析思路）：数引用 agent 查过的单元格，判定引用它做过的分析结论。analysis_claims " + "用默认，不要关\n" "- 只问表结构、清单（有哪些表、有哪些字段、列出名单）的，不加 report，agent → output 就行\n" # 三期起报告会核对写出来的表名、字段名：instructions 里让它「列出用到的表」时，编出来的名字会被标出来 "- report 提到表名、字段名时写 [[t:表名]] / [[c:表名.字段名]]（英文名也可以放进反引号，中文名不行：" @@ -932,7 +989,8 @@ async def generate( baseline = copy.deepcopy((payload.base_graph or {}).get("nodes") or (prev_graph or {}).get("nodes") or []) applier = GraphApplier(nodes, edges, patches=False) gen = Generation(model, system=system, applier=applier, looks=looks, - checker=GraphChecker(applier, scope=scope, sources=sources, baseline=baseline, checkers=checkers)) + checker=GraphChecker(applier, scope=scope, sources=sources, baseline=baseline, checkers=checkers, + question=_question_of(payload))) human = _with_history(_graph_request(payload), await _history_section(session, payload.conversation_id)) try: async for _op in gen.draft(human): @@ -956,6 +1014,7 @@ async def generate( async for _op in gen.repair(): pass covered = await gen.checker.coverage(level="warning") if isinstance(gen.checker, GraphChecker) else [] + covered += gen.checker.analysis_gaps(level="warning") if isinstance(gen.checker, GraphChecker) else [] try: spec = GraphSpec.model_validate({"nodes": list(nodes.values()), "edges": edges}) @@ -1237,9 +1296,9 @@ def _numeric_input(field: Any) -> bool: def _evidence_source(node: Any) -> bool: - """报告撰写能从这个节点拿到可引用的证据：口径卡、调用工具、配了工具的 agent、知识检索，以及声明了数值字段的 - 入口（运行输入进证据目录，报告用 [[i:字段]] 引用）。子工作流、协作团队里面是什么看不见，一律算有——这条会把图 - 打回去让模型改，宁可漏判不可误判。 + """报告撰写能从这个节点拿到可引用的证据：口径卡、分析卡（结论总能引用，判定语、具名的数、结果表，00-总方案 D10）、 + 调用工具、配了工具的 agent、知识检索，以及声明了数值字段的入口（运行输入进证据目录，报告用 [[i:字段]] 引用）。 + 子工作流、协作团队里面是什么看不见，一律算有——这条会把图打回去让模型改，宁可漏判不可误判。 入口只声明了文字字段的（问题、主题）不算：报告能照引的只有问题本身，要写的数照样没有出处。画布上的入口字段 不写类型，没写默认值的字段看不出是不是数，也不算——提示里说明了怎么声明（写数值默认值）。""" @@ -1248,8 +1307,8 @@ def _evidence_source(node: Any) -> bool: if node.type == NodeType.INPUT: fields = node.config.get("fields") return isinstance(fields, list) and any(_numeric_input(f) for f in fields) - return node.type in (NodeType.METRICS, NodeType.TOOL, NodeType.MERGE, NodeType.RETRIEVE, NodeType.SUBGRAPH, - NodeType.SUPERVISOR) + return node.type in (NodeType.METRICS, NodeType.ANALYSIS, NodeType.TOOL, NodeType.MERGE, NodeType.RETRIEVE, + NodeType.SUBGRAPH, NodeType.SUPERVISOR) def _unsourced_reports(spec: GraphSpec, changed: set[str]) -> list[dict[str, Any]]: @@ -1743,7 +1802,7 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None): started = time.monotonic() gen = Generation(model, system=system, applier=applier, looks=looks, started=started, checker=GraphChecker(applier, scope=scope, sources=sources, baseline=baseline, checkers=checkers, - chat_user=chat_user)) + chat_user=chat_user, question=_question_of(payload))) async def steps(): yield {"op": "model", "model": model_id} @@ -1818,6 +1877,8 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None): # 工具节点写死的 SQL 查的区间里表里没有数据（data/coverage.py）：自查时交回模型改过一轮，仍在的不挡运行，留在检查 # 结果里当提醒（模型照要求在说明里告诉用户） warnings += await gen.checker.coverage(level="warning") if isinstance(gen.checker, GraphChecker) else [] + # 分析类问题、查库的 Agent 没绑分析工具：同上，交回模型改过一轮仍在的不挡运行，留作提醒 + warnings += gen.checker.analysis_gaps(level="warning") if isinstance(gen.checker, GraphChecker) else [] # 建完要不要跑，由模型在 done.run 里表态；问了用户的不跑：要等他回答 autorun = applier.autorun and gen.asked is None if remaining: @@ -1893,8 +1954,9 @@ _NO_LOWERING = """\ 不删掉或清空出具契约，不删 required 里的指标，不把 strict 改成 false，不往 allow_numbers 里加数，不替人打开 cells； 审批策略只能往严里改（每次调用都审批 > 仅危险工具需要审批 > 全部自动放行），不放宽、不删掉，更不改成 never； 不取消子工作流钉住的版本；发布级别不是你能改的。结构要动（比如换掉协作团队），输出 question 说明怎么拆，不要自己动手； - 报告撰写节点的 numbers、on_violation、claims 只能往严里改（numbers: strict、on_violation: fail、 - claims: require_citation），不放宽、不删掉；claims 从宽到严是 off < require_citation < judge，已经是 judge 的不改回 + 报告撰写节点的 numbers、on_violation、claims、analysis_claims 只能往严里改（numbers: strict、on_violation: fail、 + claims: require_citation、analysis_claims: strict），不放宽、不删掉；analysis_claims 从宽到严是 off < check < strict， + 已经是 strict 的不改回 check；claims 从宽到严是 off < require_citation < judge，已经是 judge 的不改回 require_citation；judge 缺预算（judge.max_cost_usd）时输出 question 问写多少、还是写 null 不限，不替人定； judge.on_unsupported 是 withhold 的不改回 degrade、不删掉；update_node 会把 judge 整个换掉，补预算时把原来的 judge（包括从全图默认继承来的）整份带上，只加 max_cost_usd；契约里的 claims 也一样； diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py index 68c7a888..6d646d51 100644 --- a/backend/app/api/copilot_gen.py +++ b/backend/app/api/copilot_gen.py @@ -707,12 +707,17 @@ class GraphChecker: """图的自查：按运行时同一套规则（copilot._blocking_issues）挑出挡路的问题，交回模型改。 known 给了（升级证据）：只把这一轮冒出来的问题交回去（图上原来就有的不归这一轮管），一个修改都没落地时不修。 + + question 给了（问数据：用户问的那句话）：问题是分析类（app/analysis/intent）、查库的 Agent 却没绑分析工具时另提一条 + （analysis_gaps），同 coverage 只交回模型改一轮。 """ def __init__(self, applier: GraphApplier, *, scope: set[str] | None = None, sources: list[Any] | None = None, baseline: list[dict[str, Any]] | None = None, level: str | None = None, defaults: dict[str, Any] | None = None, checkers: dict[str, Any] | None = None, - known: set[Any] | None = None, chat_user: bool = False) -> None: + known: set[Any] | None = None, chat_user: bool = False, question: str | None = None) -> None: + from app.analysis.intent import analysis_intent + self.applier = applier #: 图要以开登录后 user 角色的身份在问数据里发起（D-c）：使用者不能用的节点、工具也算挡路 self.chat_user = chat_user @@ -728,6 +733,10 @@ class GraphChecker: self._coverage_cache: Any = None #: 这类问题已经交回模型改过一轮：之后不再挡（修正后仍在的留在检查结果里当提醒，由模型在说明里照实告诉用户） self._coverage_given = False + #: 用户问题的分析意图（问为什么、怎么变了、是否异常、趋势……），不是分析类问题为 None + self.intent = analysis_intent(question) if question else None + #: 「分析类问题、Agent 没绑分析工具」已经交回模型改过一轮：之后不再挡，同 _coverage_given + self._analysis_given = False def blocking(self) -> list[dict[str, Any]] | None: from app.api.copilot import _blocking_issues @@ -743,7 +752,10 @@ class GraphChecker: # 查的区间里表里没有数据：交回模型改一轮（换有这段数据的表，或者在说明里照实告诉用户）；改过一轮仍在的不挡 covered = [] if self._coverage_given or errors is None else await self.coverage() self._coverage_given = self._coverage_given or bool(covered) - return [*refused, *(errors or []), *covered] + # 分析类问题、查库的 Agent 没绑分析工具：同样只交回模型改一轮（用不上分析的，模型在说明里讲一句为什么就行） + analysis = [] if self._analysis_given or errors is None else self.analysis_gaps() + self._analysis_given = self._analysis_given or bool(analysis) + return [*refused, *(errors or []), *covered, *analysis] from app.engine.autofix import _sig if not self.applier.effective: @@ -793,9 +805,49 @@ class GraphChecker: return await coverage.sql_gaps(source, sql, facts=_profile_facts(row), masked=masked_columns(row.options), cache=self._coverage_cache, budget=COVERAGE_BUDGET_S) + def analysis_gaps(self, *, level: str = "error") -> list[dict[str, Any]]: + """问题是分析类（self.intent），图里没有分析卡，查库的 Agent 一个分析工具都没绑：一条问题（形状同自查的问题，code 是 + ANALYSIS_TOOLS_MISSING），指向第一个查库的 Agent。没有查库的 Agent 的图（只用调用工具节点取数、选用了口径）不提：这条 + 说的是「查库的 Agent 要绑分析工具」，别的结构由助手按节点手册定。level 给 warning 时是收尾那一步的提醒。""" + if self.intent is None: + return [] + from app.analysis.contract import TOOL_NAMES + from app.tools.datasource import QUERY_PREFIX + + bound = set(TOOL_NAMES.values()) + querying: tuple[str, str] | None = None + for nid, node in self.applier.nodes.items(): + ntype = (node or {}).get("type") + if ntype == NodeType.ANALYSIS: + return [] + if ntype != NodeType.AGENT: + continue + data = (node or {}).get("data") or {} + tools = {t.strip() for t in (data.get("config") or {}).get("tools") or [] if isinstance(t, str)} + if tools & bound: + return [] + if querying is None and any(t.startswith(QUERY_PREFIX) for t in tools): + querying = (nid, str(data.get("label") or "").strip() or nid) + if querying is None: + return [] + nid, title = querying + asked = f"问题问的是{self.intent.label}（「{self.intent.text}」）" + names = "、".join(TOOL_NAMES[op] for op in ANALYSIS_TOOL_ORDER) + return [{"level": level, "node_id": nid, "edge_id": None, "field": "tools", "code": ANALYSIS_TOOLS_MISSING, + "message": f"{asked}，查库的「{title}」没有绑定分析工具：变化从哪来、是否异常、是不是趋势只能由模型自己判断，" + "报告里的这类判断没有分析结论可依据", + "for_model": f"{asked}。按「分析类问题」的形状改：给查库的 agent「{nid}」的 tools 加上 {names}" + "（db_query__、db_schema__ 照留），平台会把分析方法注入它的 system，prompt 里写清要回答的问题、" + "比哪两期就行；report 的 instructions 让它按分析结论写、判断直接引用判定语。问的其实不是原因、" + "异常、趋势这类、确实用不上分析的，不改这一处，在 done 的 explanation 里说一句为什么"}] + #: 自查时一条 SQL 的覆盖检查最多花多少秒（data/coverage.py）：搭图要等它，比查询时的 BUDGET_S 短 COVERAGE_BUDGET_S = 8.0 +#: 分析类问题、查库的 Agent 没绑分析工具（GraphChecker.analysis_gaps）那条问题的 code +ANALYSIS_TOOLS_MISSING = "analysis_tools_missing" +#: 自查让助手绑分析工具时列出的先后：先定现象（可比性、基线、趋势），再拆、再检验（同分析方法 tools/builtin/analysis.py） +ANALYSIS_TOOL_ORDER = ("comparable", "baseline", "trend", "decompose", "differ") async def _probe_source(row: Any) -> Any: diff --git a/backend/app/chat/drive.py b/backend/app/chat/drive.py index d406b27f..5295d9d4 100644 --- a/backend/app/chat/drive.py +++ b/backend/app/chat/drive.py @@ -150,11 +150,12 @@ class TurnDriver: # ------------------------------------------------------------------ 子类给的差异 def generate_in(self, instruction: str, base_graph: dict[str, Any] | None) -> Any: - """这一次生成的请求（api/copilot.GenerateIn）。""" + """这一次生成的请求（api/copilot.GenerateIn）。question 是这一轮原本的问题：重来时 instruction 是带着原因的那段话， + 自查按原本的问题判分析意图。""" from app.api.copilot import GenerateIn return GenerateIn(instruction=instruction, base_graph=base_graph, intent="answer", provider=self.provider, - model=self.model, datasource_ids=self.datasource_ids or None) + model=self.model, datasource_ids=self.datasource_ids or None, question=self.question) def catalog_view(self) -> Any: """这一轮读哪一份数据目录（data/catalog_view.py），None 是当前目录。""" diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index 6bfdfae9..d45e7365 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -153,7 +153,8 @@ class ChatDriver(TurnDriver): from app.api.copilot import GenerateIn return GenerateIn(instruction=instruction, base_graph=base_graph, intent="answer", - conversation_id=self.execution.conversation_id, datasource_ids=self.datasource_ids or None) + conversation_id=self.execution.conversation_id, datasource_ids=self.datasource_ids or None, + question=self.question) async def on_op(self, op: dict[str, Any]) -> None: kind = op.get("op") diff --git a/backend/tests/test_copilot_analysis.py b/backend/tests/test_copilot_analysis.py new file mode 100644 index 00000000..b98e89ee --- /dev/null +++ b/backend/tests/test_copilot_analysis.py @@ -0,0 +1,248 @@ +"""助手搭图认分析能力（00-总方案 §3.7）：节点手册教分析卡和分析工具、报告的分析结论引用写法；问数据的包装多一种「分析类 +问题」的形状；自查时问题是分析类（app/analysis/intent）、查库的 Agent 却没绑分析工具，交回模型改一轮（只挡一轮，照 coverage +的样板），改过一轮仍在的不挡运行、留作提醒；报告的上游只有分析卡也算有可引用的来源。 + +手册里举的引用标记交给真正的解析器，确认写法和系统认的是同一套（同 tests/test_copilot_entity_guide.py）。问法、节点、 +表都是虚构的零售场景。 +""" +from __future__ import annotations + +import json +import re + +import pytest +from langchain_core.messages import AIMessageChunk + +from app.analysis import contract +from app.api.copilot import _ASSIST_RULES, NODE_REFERENCE, GenerateIn, _evidence_source, _question_of, _user_message +from app.api.copilot_gen import ANALYSIS_TOOLS_MISSING, GraphApplier, GraphChecker +from app.engine.evidence import _CELL_REF, _ROW_REF, SUPPORTED_KINDS, parse_markers +from app.engine.schema import ANALYSIS_CLAIMS_VALUES, ANALYSIS_SCOPE_VALUES, GraphNode, NodeType +from tests.test_copilot_answer_ops import _sse, _use, client # noqa: F401 + +TOOLS = list(contract.TOOL_NAMES.values()) + + +def _section(start: str, end: str) -> str: + i = NODE_REFERENCE.index(start) + return NODE_REFERENCE[i:NODE_REFERENCE.index(end, i)] + + +def _report() -> str: + return _section("- report：", "\n- output") + + +# -------------------------------------------------------------------------- +# 节点手册 +# -------------------------------------------------------------------------- + + +def test_the_card_entry_teaches_the_analysis_card(): + card = _section("- analysis：分析卡", "\n- dual_check") + for words in (["items", "overrides", "from", "params", "hypothesis"], # 配置 + ["模板里每次都要做的分析", "周报的异动"], # 用途 + ["nodes..results.<项 id>", "verdict", "branch"], # 产出、按判定分支 + ["tool 节点或 merge", "agent 不能当", "headline", "parent"], # 输入只认查询节点 + ["不要把结果交给口径卡", "追不到原始数据"]): # D16 + assert all(w in card for w in words), [w for w in words if w not in card] + assert all(op in card for op in contract.OPS) + assert all(mode in card for modes in contract.MODES.values() for mode in modes) + # 例子里的阈值键、判定码都是契约里的 + assert all(key in contract.DEFAULTS for key in re.findall(r"\b(k_warn|main_share|k_alert|flat_pct)\b", card)) + branch = re.search(r"verdict in \[([^\]]+)\]", card).group(1) + assert {v.strip(" '") for v in branch.split(",")} <= set(contract.VERDICTS["baseline"]) + + +def test_the_agent_entry_names_the_five_analysis_tools(): + agent = _section("- agent：", "\n- supervisor") + assert all(name in agent for name in TOOLS) + for words in ("分析方法", "注入", "不要写阈值", "只绑给同时绑了", "db_query__", "不能放进 tool 节点"): + assert words in agent, words + + +def test_the_markers_of_analysis_conclusions_parse_as_the_system_reads_them(): + markers = [m for m in parse_markers(_report()) if "A1" in m["body"]] + kinds = {m["kind"] for m in markers} + assert {contract.MARKER_KIND, "v", "table", "see"} <= kinds <= SUPPORTED_KINDS + by_kind = {m["kind"]: m for m in markers} + plain = [m for m in markers if m["kind"] == contract.MARKER_KIND and m["ref"] == "A1"] + named = [m for m in markers if m["kind"] == contract.MARKER_KIND and m["ref"] != "A1"] + assert plain and named and named[0]["ref"] == "A1.top_share" and named[0]["conv"] == "pct" + cell = _CELL_REF.match(by_kind["v"]["ref"]) + assert cell and cell.group("alias") == "A1" and cell.group("column") == "share" + assert _ROW_REF.match(by_kind["table"]["ref"]).group("alias") == "A1" + assert by_kind["see"]["refs"] == ["A1"] + # 拆解结果表的列 id（判定写法里举的 delta、share）是真结果表的列 + from app.analysis.decompose import _FIXED_ADDITIVE + + cols = by_kind["table"]["options"]["cols"].split(",") + assert set(cols[1:]) <= set(_FIXED_ADDITIVE), cols + + +def test_the_report_entry_spells_out_analysis_claims_and_scope(): + report = _report() + assert "analysis_claims" in report and "analysis_scope" in report + assert all(v in report for v in ANALYSIS_CLAIMS_VALUES) and all(v in report for v in ANALYSIS_SCOPE_VALUES) + assert "off < check < strict" in report and "只能往严里改" in report + for words in ("主要来自", "异常", "持续上升", "待核实的推测", "说明不了为什么"): + assert words in report, words + + +def test_governed_templates_and_the_publish_assist_keep_analysis_claims_strict(): + governed = NODE_REFERENCE[NODE_REFERENCE.index("要按受管级别发布的模板"):] + assert 'analysis_claims: "strict"' in governed[:400] + rules = " ".join(_ASSIST_RULES.split()) + assert re.search(r"analysis_claims.{0,40}只能往严里改|只能往严里改.{0,80}analysis_claims: strict", rules), rules + assert "off < check < strict" in rules and "不改回 check" in rules + + +def test_the_build_rules_bind_analysis_tools_for_analysis_questions(): + rules = NODE_REFERENCE[NODE_REFERENCE.index("搭图规则"):NODE_REFERENCE.index("模型字段")] + seventh = rules[rules.index("\n7."):] + assert "分析工具" in seventh and "分析卡" in seventh and "不进口径卡" in seventh + + +# -------------------------------------------------------------------------- +# 问数据的包装 +# -------------------------------------------------------------------------- + + +def test_the_answer_request_has_an_analysis_shape(): + text = _user_message(GenerateIn(instruction="上周销售额为什么下降", intent="answer"), patch=False) + shape = next(line for line in text.splitlines() if line.startswith("- 问的是为什么")) + assert "input → agent → report → output" in shape + assert all(name in shape for name in TOOLS) and "db_query__" in shape and "db_schema__" in shape + assert "按分析结论写" in shape and "判定语" in shape and "不写阈值" in shape + # 问数是探索运行：不替人打开结论句策略（test_copilot_entity_guide 守着），说法核对用默认、不关 + assert not re.search(r"(? dict: + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def _graph(*, tools=("db_query__shop", "db_schema__shop"), card=False): + nodes = {"start": _n("start", "input"), + "bot": _n("bot", "agent", prompt="{{ input.question }}", tools=list(tools)), + "write": _n("write", "report", instructions="回答：{{ input.question }}"), + "out": _n("out", "output", fields=[{"name": "回答", "value": "{{ nodes.write.text }}"}])} + if card: + nodes["check"] = _n("check", "analysis", items=[]) + order = ["start", "bot", "write", "out"] + return nodes, [{"source": a, "target": b} for a, b in zip(order, order[1:])] + + +async def test_an_analysis_question_without_analysis_tools_is_returned_once(): + nodes, edges = _graph() + checker = GraphChecker(GraphApplier(nodes, edges), question="上周销售额为什么下降") + [issue] = [i for i in await checker.issues() if i["code"] == ANALYSIS_TOOLS_MISSING] + assert (issue["level"], issue["node_id"], issue["field"]) == ("error", "bot", "tools") + assert issue["message"].startswith("问题问的是原因（「为什么下降」），查库的「bot」没有绑定分析工具") + assert all(name in issue["for_model"] for name in TOOLS) and "explanation" in issue["for_model"] + # 交回模型改过一轮：之后不再挡；收尾时作为提醒 + assert not [i for i in await checker.issues() if i["code"] == ANALYSIS_TOOLS_MISSING] + [warning] = checker.analysis_gaps(level="warning") + assert (warning["level"], warning["code"]) == ("warning", ANALYSIS_TOOLS_MISSING) + + +@pytest.mark.parametrize("question, graph", [ + ("上周销售额为什么下降", _graph(tools=("db_query__shop", "db_schema__shop", "analysis_decompose"))), # 绑了一个就算 + ("上周销售额为什么下降", _graph(card=True)), # 有分析卡 + ("上周销售额为什么下降", _graph(tools=("current_time",))), # 没有查库的 Agent + ("上周销售额是多少", _graph()), # 不是分析类问题 + (None, _graph()), # 画布：没有问题 +]) +async def test_the_rule_stays_quiet_otherwise(question, graph): + nodes, edges = graph + checker = GraphChecker(GraphApplier(nodes, edges), question=question) + assert not [i for i in await checker.issues() if i["code"] == ANALYSIS_TOOLS_MISSING] + assert checker.analysis_gaps(level="warning") == [] + + +def _ops(tools: list[str]) -> list[dict]: + nodes, edges = _graph(tools=tools) + return [{"op": "plan", "summary": "查销售额"}, + *({"op": "add_node", "node": {"id": n["id"], "type": n["type"], "label": n["id"], + "config": n["data"]["config"]}} for n in nodes.values()), + *({"op": "add_edge", "edge": e} for e in edges), {"op": "done", "explanation": "查一下", "run": True}] + + +def _scripted(first: list[dict], fix: list[dict], seen: list[str]): + class Scripted: + async def astream(self, messages): + human = next(text for role, text in messages if role == "human") + seen.append(human) + for op in (fix if "上线前自查" in human else first): + yield AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n") + + return Scripted() + + +async def test_the_stream_asks_for_one_repair_and_the_model_binds_the_tools(client, monkeypatch): # noqa: F811 + seen: list[str] = [] + fix = [{"op": "update_node", "id": "bot", "config": {"tools": ["db_query__shop", "db_schema__shop", *TOOLS]}}, + {"op": "done", "explanation": "绑上了分析工具"}] + _use(monkeypatch, _scripted(_ops(["db_query__shop", "db_schema__shop"]), fix, seen)) + got = await _sse(client, {"instruction": "上周销售额为什么下降", "intent": "answer"}) + [repairing] = [op for op in got if op["op"] == "check" and op["status"] == "repairing"] + assert ANALYSIS_TOOLS_MISSING in [i["code"] for i in repairing["issues"]] + assert any("上线前自查" in h and "analysis_decompose" in h for h in seen), "交回模型改一轮" + [final] = [op for op in got if op["op"] == "final"] + bot = next(n for n in final["graph"]["nodes"] if n["id"] == "bot") + assert set(TOOLS) <= set(bot["data"]["config"]["tools"]) + assert not [i for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_MISSING] + + +async def test_a_model_that_keeps_its_graph_still_runs_with_a_warning(client, monkeypatch): # noqa: F811 + seen: list[str] = [] + keep = [{"op": "done", "explanation": "这个问题其实只要一个数，不用分析"}] + _use(monkeypatch, _scripted(_ops(["db_query__shop", "db_schema__shop"]), keep, seen)) + got = await _sse(client, {"instruction": "上周销售额为什么下降", "intent": "answer"}) + assert sum("上线前自查" in h for h in seen) == 1, "只挡一轮" + [final] = [op for op in got if op["op"] == "final"] + assert final["next"] == "run", "改过一轮仍在的不挡运行" + assert [i["level"] for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_MISSING] == ["warning"] + + +async def test_a_retry_reads_the_original_question(client, monkeypatch): # noqa: F811 + """重来时 instruction 是带着原因的那段话（里面有「原因」两个字）：按原问题判，原问题不是分析类就不提。""" + seen: list[str] = [] + _use(monkeypatch, _scripted(_ops(["db_query__shop", "db_schema__shop"]), [], seen)) + retry = "这个工作流上一次运行没有成功：用满了步数。\n\n请针对这个原因改它，改完继续回答原来的问题：上周销售额多少" + got = await _sse(client, {"instruction": retry, "intent": "answer", "question": "上周销售额多少"}) + assert not [op for op in got if op["op"] == "check" and op["status"] == "repairing"] + [final] = [op for op in got if op["op"] == "final"] + assert not [i for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_MISSING]