助手建图认分析能力：节点手册加分析卡一段（用途是模板里固定要做的分析，items / overrides，输入只认调用数据库查询工具的调用工具节点和合并查询，产出 nodes.<id>.results.<项> 可按判定分支，结果不进口径卡）、agent 段补五个分析工具（分析类问题绑，绑了平台注入分析方法，只绑给查库的 agent）、报告段补分析结论的引用写法（a:An、a:An.键、v:An.r0.列、table:An、句末 see:An）和分析说法核对（check 缺省、strict、off，只能往严里改）、分析范围，受管指引和发布前修复的硬性约定写明分析说法核对要严格；搭图规则加一条。问数据的包装加「分析类问题」的形状（input → agent 绑查询、表结构和分析工具 → report 按分析结论写、引用判定语 → output）。报告的证据来源认分析卡。自查：GenerateIn 加 question（问数据的一轮把原问题带上，重来时 instruction 是带着原因的那段话），问题是分析类、查库的 Agent 没绑分析工具、图里也没有分析卡时提一条问题（analysis_tools_missing，带 for_model），照 coverage 只交回模型改一轮，改过一轮仍在的不挡运行、收尾时留作提醒。测试：手册里的引用写法交给真解析器、分析卡与工具的叫法取契约、包装的分析形状、自查只挡一轮与几种不提的情形、重来按原问题判

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.<agent 的 assign_to>.字段
+- 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.<id>.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.<assign_to>.字段。出口字段写 {{ 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 也一样；

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:

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 是当前目录。"""

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")

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.<id>.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"(?<!analysis_)claims\s*[:=]\s*['\"]?(require_citation|judge)", text)
+    assert "analysis_claims " in shape and "不要关" in shape
+
+
+def test_the_question_for_the_self_check():
+    """自查按用户原本的问题判意图：问数据有原问题时用它（重来时 instruction 是带着原因的那段话），没给用 instruction；
+    画布上用户说的是流程，不判。"""
+    retry = "这个工作流上一次运行没有成功：用满了步数。\n\n请针对这个原因改它，改完继续回答原来的问题：上周销售额多少"
+    assert _question_of(GenerateIn(instruction=retry, intent="answer", question="上周销售额多少")) == "上周销售额多少"
+    assert _question_of(GenerateIn(instruction="上周销售额为什么下降", intent="answer")) == "上周销售额为什么下降"
+    assert _question_of(GenerateIn(instruction="做一个分析销售额为什么下降的流程")) is None
+
+
+def test_the_chat_drivers_pass_the_original_question():
+    from app.chat.drive import TurnDriver
+
+    driver = TurnDriver("上周销售额为什么下降", datasource_ids=[])
+    payload = driver.generate_in("这个工作流上一次运行没有成功：…改完继续回答原来的问题：上周销售额为什么下降", None)
+    assert payload.question == "上周销售额为什么下降" and payload.intent == "answer"
+
+
+# --------------------------------------------------------------------------
+# 报告的证据来源认分析卡
+# --------------------------------------------------------------------------
+
+
+def test_an_analysis_card_counts_as_an_evidence_source():
+    card = GraphNode.model_validate({"id": "check", "type": "analysis", "data": {"label": "异动分析卡", "config": {}}})
+    assert card.type == NodeType.ANALYSIS and _evidence_source(card)
+
+
+# --------------------------------------------------------------------------
+# 自查：分析类问题，查库的 Agent 要绑分析工具（只挡一轮）
+# --------------------------------------------------------------------------
+
+
+def _n(nid: str, ntype: str, **config) -> 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]