diff --git a/backend/app/analysis/intent.py b/backend/app/analysis/intent.py index e5055bb6..3879c6a1 100644 --- a/backend/app/analysis/intent.py +++ b/backend/app/analysis/intent.py @@ -1,10 +1,12 @@ """问题意图：用户问的一句话是不是分析类问题（docs/design/analysis-round/00-总方案.md §3.7）。 -助手搭问数据的图时自查（api/copilot_gen.GraphChecker 的 analysis_gaps）用它：问题问的是为什么、怎么变了、哪里拖累、谁贡献 -最大、是否异常、正常吗、波动、趋势、持续、是不是真的更高这类，而查库的 Agent 没绑分析工具、图里也没有分析卡时，提一条只挡 +助手搭问数据的图时自查（api/copilot_gen.GraphChecker 的 analysis_gaps）用它：问题问的是为什么、怎么变了、哪里拖累、谁拉动 +了变化、是否异常、正常吗、波动、趋势、持续、是不是真的更高这类，而查库的 Agent 没绑分析工具、图里也没有分析卡时，提一条只挡 一轮的问题，请助手给 Agent 绑上分析工具——变化从哪来、算不算异常、是不是趋势由平台按固定的方法和阈值算，判定语由系统写， 报告引用它。 +问构成的不算（「哪个门店销售额最多」「谁贡献最大」）：没有变化要说，分析卡和分析工具都用不上。 + **确定的词表，带版本**（INTENT_VERSION）：不调模型。只收明确的问法，宁可漏判：漏判只是少一条自查提醒，误判要让助手多改一轮、 给用不上的 Agent 绑分析工具。按句子认（句号、问号、感叹号、分号、换行分开；逗号不分：「销售额下降了，为什么」是一句）， 一句里命中一种就算，先认先得（原因 → 来源 → 异常 → 波动 → 趋势 → 差异）。各类收什么、不收什么： @@ -12,11 +14,13 @@ - 原因（why）：「为什么 / 为何 / 为啥 / 怎么会」要和涨跌、变化、这么高、比…低这类说法在同一句（「为什么报错」「为什么查不到 数据」不算）；「原因」要是变化的原因（「下降的原因」「什么原因导致订单减少」「销售额下滑了，原因是什么」），「退款原因有哪些」 「按取消原因统计」问的是一个字段，不算 -- 来源（source）：「怎么变了 / 有什么变化」「哪里拖累 / 谁拉动 / 什么带动了增长」「贡献最大」「下降主要来自哪」「驱动因素」 +- 来源（source）：「怎么变了 / 有什么变化」「哪里拖累 / 谁拉动 / 什么带动了增长」「下降主要来自哪」「驱动因素」 「把变化拆开」「归因分析」。「订单来自哪些渠道」问的是构成，不算（要先说变化）；「表结构有什么变化」「口径有什么变化」问的 - 不是数，不算 + 不是数，不算。「谁贡献最大 / 哪个门店贡献的销售额最多」问的也是构成，同句里要有变化的说法（涨、跌、降、增、变化、降幅、 + 增幅这类）才算（「对销售额下降贡献最大」）；拖累、拉动这几个词本身就说明了变化，不另要求 - 异常（anomaly）：「是否异常 / 异常吗 / 正常吗 / 算不算异常 / 有没有异动 / 是不是暴跌了」这类问句，和「超出正常范围」。 - 「异常订单有多少」「有没有异常记录」「门店是否正常营业」问的是业务状态，「系统正常吗」「连接正常吗」问的是服务，都不算 + 「异常订单有多少」「有没有异常记录」「门店是否正常营业」问的是业务状态，「系统正常吗」「连接正常吗」问的是服务，都不算； + 「数据是否正常 / 今天的数据正常吗 / 上周的数据有没有异常」问的是这批数据本身（「数据」作主语，中间没有别的名词），也不算 - 波动（fluctuation）：「波动大吗 / 变化大不大 / 是不是正常波动 / 只是波动」。「列出价格波动超过 10% 的商品」是筛选，「网络波动大吗」问的 是服务，都不算 - 趋势（trend）：「趋势 / 走势」「持续上升 / 下降会持续吗 / 一直在降 / 连续三周下降 / 越来越少 / 是不是在涨 / 稳步增长」。 @@ -35,8 +39,9 @@ from __future__ import annotations import re from dataclasses import dataclass -#: 词表的版本：加减词、改规则就升版本（评测比两次的提示词时据此认得出自查规则变了） -INTENT_VERSION = 1 +#: 词表的版本：加减词、改规则就升版本（评测比两次的提示词时据此认得出自查规则变了；app/eval/runner.py 的 +#: prompt_fingerprint 带着它） +INTENT_VERSION = 2 #: 意图的种类 → 给人看的叫法（自查的那条问题里说「问题问的是……」） KINDS: dict[str, str] = { @@ -87,6 +92,35 @@ _WHY_WHAT = re.compile(rf"{_CHANGE}|(?:这么|那么|如此|这样|偏|特别| _REASON_ASK = re.compile(r"原因(?:是什么|是啥|有哪些|在哪|何在|为何)") _CHANGE_RE = re.compile(_CHANGE) +#: 来源里问「贡献」的这一条：还要同一句里有变化的说法（_CHANGE 里的涨跌、变化、降幅、增幅这类）才认。「哪个门店贡献的 +#: 销售额最多」「谁贡献最大」问的是构成，不算；「对销售额下降贡献最大」才算。拖累、拉动、带动、推动本身就在说变化，不另要求 +_SOURCE_CONTRIBUTION = re.compile(r"(?:哪|谁|什么).{0,10}?贡献|贡献了?.{0,6}?(?:最大|最多|最明显)") +#: 问异常的两条：「数据」作主语时（前面出现「数据」、中间没有别的名词）问的是这批数据本身，不算（_DATA_SUBJECT） +_ANOMALY_ASK = re.compile(rf"(?:是否|是不是|有没有|算不算|算是|属于)(?:出现)?了?(?:异常|反常|不正常)(?!{_STATUS_NOUN})") +_ANOMALY_NORMAL = re.compile(rf"(?:异常|反常)(?!{_STATUS_NOUN})(?:吗|么|与否)|异常不异常" + rf"|(?:正常|不正常)(?!{_STATUS_VERB})(?:吗|么|与否)|正不正常" + rf"|(?:是否|是不是|算不算|属于)正常(?!{_STATUS_VERB}|的?波动)") +#: 「数据」当主语：命中点前面不远处是「数据」，中间没有别的名词（「今天的数据正常吗」「数据是否正常」）。中间隔着「销售额」 +#: 「订单」这类名词的是在问那一项的数（「这周的数据里销售额是否异常」），照认 +_DATA_SUBJECT = re.compile(r"数据[^，,、]{0,4}$") +#: 中间这些字不算「别的名词」：助词、代词、判断词、量词、时间词 +_NOT_A_NOUN = set("的了吗呢么啊呀吧是有没有这那该其本整批些个条张期天周月年日昨今上前下去来到在也都就还又很更太最" + "显示看起来觉得认为说查看算统计") + + +def _data_subject(sentence: str, at: int) -> bool: + """命中点前面是不是「…的数据」（中间没有别的名词）：问的是这批数据本身正常不正常，不是那一项的数变没变。""" + head = sentence[:at] + found = _DATA_SUBJECT.search(head) + if found is None: + return False + middle = head[found.start() + len("数据"):] + return all(ch in _NOT_A_NOUN or not ("一" <= ch <= "鿿") for ch in middle) + + +#: 命中点在「…的数据」后面时不算的规则（问这批数据本身，不是在问那一项的数） +_DATA_SUBJECT_RULES = (_ANOMALY_ASK, _ANOMALY_NORMAL) + #: (种类, 规则, data_only：这一句有 _NOT_DATA 时不认) _RULES: list[tuple[str, re.Pattern[str], bool]] = [ ("why", re.compile(rf"(?:{_CHANGE}).{{0,6}}?的?(?:主要)?原因"), False), @@ -94,19 +128,17 @@ _RULES: list[tuple[str, re.Pattern[str], bool]] = [ ("source", re.compile(r"归因分析|做(?:个|一下|下)?归因|归因一下|(?:怎么|如何)归因|驱动(?:因素|力)" r"|(?:由|靠)(?:什么|哪些|哪个).{0,4}?驱动"), False), ("source", re.compile(r"怎么变(?:了|的|化)|如何变化|有(?:什么|啥|哪些)变化|变化(?:在哪|如何|怎么样|情况)"), True), - ("source", re.compile(r"(?:哪|谁|什么).{0,10}?(?:拖累|拉低|拉高|拉动|带动|推动|贡献)" - r"|(?:拖累|拉动|带动|推动|贡献)了?.{0,6}?(?:最大|最多|最明显)"), False), + ("source", re.compile(r"(?:哪|谁|什么).{0,10}?(?:拖累|拉低|拉高|拉动|带动|推动)" + r"|(?:拖累|拉动|带动|推动)了?.{0,6}?(?:最大|最多|最明显)"), False), + ("source", _SOURCE_CONTRIBUTION, False), ("source", re.compile(rf"对.{{0,12}}?(?:{_CHANGE}|增量|增幅|降幅).{{0,2}}?贡献"), False), ("source", re.compile(rf"(?:{_CHANGE}).{{0,12}}?(?:主要)?(?:来自|来源于|出在|集中在|源于|发生在)(?:哪|什么|何)"), False), ("source", re.compile(rf"(?:哪|什么).{{0,8}}?(?:导致|造成|引起|带来)了?.{{0,12}}?(?:{_CHANGE})"), False), ("source", re.compile(rf"(?:拆解|拆分|分解|拆开).{{0,12}}?(?:{_CHANGE})|(?:{_CHANGE}).{{0,12}}?(?:拆解|拆分|分解|拆开)"), False), - ("anomaly", re.compile(rf"(?:是否|是不是|有没有|算不算|算是|属于)(?:出现)?了?(?:异常|反常|不正常)(?!{_STATUS_NOUN})"), - True), - ("anomaly", re.compile(rf"(?:异常|反常)(?!{_STATUS_NOUN})(?:吗|么|与否)|异常不异常" - rf"|(?:正常|不正常)(?!{_STATUS_VERB})(?:吗|么|与否)|正不正常" - rf"|(?:是否|是不是|算不算|属于)正常(?!{_STATUS_VERB}|的?波动)"), True), + ("anomaly", _ANOMALY_ASK, True), + ("anomaly", _ANOMALY_NORMAL, True), ("anomaly", re.compile(r"(?:超出|超过|偏离)了?(?:正常|常规|历史)的?(?:范围|水平|波动|区间)" r"|(?:有没有|是否有|是不是)异动|异动分析|异动吗"), True), ("anomaly", re.compile(r"(?:是否|是不是|算不算|有没有).{0,6}?(?:大幅|骤降|暴跌|激增|飙升|突增|突降|大涨|大跌)"), True), @@ -185,8 +217,16 @@ def analysis_intent(question: str | None) -> Intent | None: for kind, rule, data_only in _RULES: if data_only and not_data: continue - if (m := rule.search(sentence)) is not None: - return Intent(kind, _quote(m.group(0))) + m = rule.search(sentence) + if m is None: + continue + # 问「贡献」的：同句要有变化的说法，问构成（谁最多）不算 + if rule is _SOURCE_CONTRIBUTION and _CHANGE_RE.search(sentence) is None: + continue + # 问异常的：「…的数据是否正常」「今天的数据正常吗」问的是这批数据本身，不算 + if rule in _DATA_SUBJECT_RULES and _data_subject(sentence, m.start()): + continue + return Intent(kind, _quote(m.group(0))) for kind, rule in _EN_RULES: if (m := rule.search(sentence)) is not None: return Intent(kind, _quote(m.group(0))) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 34191d7a..b33c72a8 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -404,14 +404,33 @@ _ANALYSIS_PROMPT_SPANS = ( ) +#: 关掉分析工具时节点手册里另外改写的几处：(原文, 换成)。报告段里提到分析结论的来处、分析结论的引用写法、分析说法核对和分析 +#: 范围两个配置，受管规则里分析卡和分析工具的那两句——关的那一臂没有分析结论，也不让助手知道有分析卡、分析工具 +_ANALYSIS_PROMPT_EDITS = ( + ("K2…）、\n 分析结论（agent 调分析工具、分析卡做的，按先后编成 A1、A2…）和运行输入", "K2…）\n 和运行输入"), + (" claims, analysis_claims, analysis_scope, entities, assign_to}", " claims, entities, assign_to}"), + (' analysis_claims: "strict"、\n', ""), + ("。分析卡的输入要是\n 调用 db_query__ 的 tool 节点或 merge；分析卡的结论总能引用，agent 用分析工具现场做的分析和它查的格同一个规矩" + "（契约\n 写 cells: true 才引用得了）", "。"), +) +#: 同上，整段去掉的：(从这一行起, 到这一行之前) +_ANALYSIS_PROMPT_CUTS = ( + (" - 分析结论：[[a:A1]]", " - 涉及码值"), + (" analysis_claims 是分析说法核对", " entities 默认 link"), +) + + def node_reference(*, analysis_tools: bool = True) -> str: - """给助手的节点手册。analysis_tools 为假时去掉只讲分析能力的那几段（_ANALYSIS_PROMPT_SPANS）。""" + """给助手的节点手册。analysis_tools 为假时去掉讲分析能力的那几段（_ANALYSIS_PROMPT_SPANS、_ANALYSIS_PROMPT_CUTS）， + 改写顺带提到它的几句（_ANALYSIS_PROMPT_EDITS）：手册里不再出现分析卡、分析工具和分析相关的配置键。""" if analysis_tools: return NODE_REFERENCE text = NODE_REFERENCE - for start, end in _ANALYSIS_PROMPT_SPANS: + for start, end in (*_ANALYSIS_PROMPT_SPANS, *_ANALYSIS_PROMPT_CUTS): i = text.index(start) text = text[:i] + text[text.index(end, i):] + for old, new in _ANALYSIS_PROMPT_EDITS: + text = text.replace(old, new, 1) return text @@ -971,19 +990,62 @@ def _chat_user_issues(nodes: dict[str, dict[str, Any]], edges: list[dict[str, An "去掉不允许的节点、工具和记忆范围 / 知识库配置"}] -#: 关掉分析工具时（answer_ops 的 analysis_tools 为假）图里还绑着分析工具的那条问题的 code +#: 关掉分析工具时（answer_ops 的 analysis_tools 为假）图里还用着分析能力的那条问题的 code ANALYSIS_TOOLS_OFF = "analysis_tools_off" +#: 那条问题交回模型的改法（几种节点共用的后半句） +_OFF_FIX = "这一轮没有分析能力可用：数据要怎么比、怎么拆，写进查询，报告直接引用查询结果" -def _analysis_off_issues(nodes: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: - """评测「关掉分析工具」那一臂：这一轮的工具清单里没有分析工具，图里照样绑了的（Agent、协作成员、调用工具节点）按 - 「绑定的工具不存在」挡（同 engine/launch 发起前那一道的说法），交回模型去掉；改不好就不运行。一个节点一条。""" +async def analysis_off_subgraphs(nodes: dict[str, dict[str, Any]]) -> set[str]: + """评测「关掉分析工具」那一臂：图里哪些子工作流节点（照运行时的取法：固定了版本的读那一版，没固定的读现在的图， + 一层层往下）用了分析能力——有分析卡，或者有绑了分析工具的 Agent、协作成员（engine/launch 快照分析设置时的同一个认法）。 + 读不到、不成形的不算（子工作流节点运行时自己会报）。""" + from app.db.base import SessionLocal + from app.engine.launch import _analyzes, _any_graph + + out: set[str] = set() + subs = [(nid, n) for nid, n in nodes.items() if (n or {}).get("type") == NodeType.SUBGRAPH] + if not subs: + return out + async with SessionLocal() as session: + for nid, node in subs: + try: + spec = GraphSpec.model_validate({"nodes": [node], "edges": []}) + except Exception: # noqa: BLE001 - 节点本身不成形，校验另报 + continue + if await _any_graph(session, spec, _analyzes): + out.add(nid) + return out + + +def _analysis_off_issues(nodes: dict[str, dict[str, Any]], subgraphs: set[str] | frozenset[str] = frozenset() + ) -> list[dict[str, Any]]: + """评测「关掉分析工具」那一臂：这一轮的工具清单、节点手册里都没有分析能力，图里照样用了的挡回去改（改不好就不运行）， + 一个节点一条，说法照「绑定的工具不存在」那一套（engine/launch 发起前那一道）： + - Agent、协作成员、调用工具节点绑了分析工具：绑定的工具不存在，去掉它们 + - 分析卡：这次不能用分析卡，去掉它，连线接回去 + - 子工作流里用了分析能力（subgraphs：analysis_off_subgraphs 算好的节点 id）：这次不能用这个子工作流""" from app.tools.analysis import ANALYSIS_TOOLS out: list[dict[str, Any]] = [] for nid, node in nodes.items(): cfg = ((node or {}).get("data") or {}).get("config") or {} ntype = (node or {}).get("type") + title = str(((node or {}).get("data") or {}).get("label") or "").strip() or nid + if ntype == NodeType.ANALYSIS: + out.append({"level": "error", "node_id": nid, "edge_id": None, "field": None, "code": ANALYSIS_TOOLS_OFF, + "message": f"这次不能用分析卡：「{title}」是分析卡，这一轮没有分析能力可用，请去掉它", + "for_model": f"「{nid}」是 analysis 节点，这一轮不能用：用 remove_node 去掉它，再把它上游的节点直接" + f"连到它下游；{_OFF_FIX}"}) + continue + if ntype == NodeType.SUBGRAPH: + if nid in subgraphs: + out.append({"level": "error", "node_id": nid, "edge_id": None, "field": "workflow_id", + "code": ANALYSIS_TOOLS_OFF, + "message": f"这次不能用这个子工作流：「{title}」嵌套的工作流里有分析卡或分析工具，这一轮没有分析能力可用", + "for_model": f"「{nid}」嵌套的工作流里有分析卡或分析工具，这一轮不能用：去掉这个 subgraph 节点，" + f"取数直接用 tool 节点或 agent 查；{_OFF_FIX}"}) + continue if ntype == NodeType.TOOL: names = [cfg.get("tool")] elif ntype == NodeType.AGENT: @@ -995,13 +1057,11 @@ def _analysis_off_issues(nodes: dict[str, dict[str, Any]]) -> list[dict[str, Any bound = list(dict.fromkeys(n.strip() for n in names if isinstance(n, str) and n.strip() in ANALYSIS_TOOLS)) if not bound: continue - title = str(((node or {}).get("data") or {}).get("label") or "").strip() or nid field = "tool" if ntype == NodeType.TOOL else "agents" if ntype == NodeType.SUPERVISOR else "tools" out.append({"level": "error", "node_id": nid, "edge_id": None, "field": field, "code": ANALYSIS_TOOLS_OFF, "message": f"绑定的工具不存在：「{title}」绑定的 {'、'.join(bound)}。这一轮没有分析工具可用，请在节点中去掉它们", - "for_model": f"「{nid}」绑定的 {'、'.join(bound)} 不在这一轮的工具清单里，这一轮没有分析工具可用：从它的" - "工具里去掉这几个，只用上面列出的工具（db_query__、db_schema__ 照留）；数据要怎么比、怎么拆，" - "写进查询"}) + "for_model": f"「{nid}」绑定的 {'、'.join(bound)} 不在这一轮的工具清单里：从它的工具里去掉这几个，只用上面" + f"列出的工具（db_query__、db_schema__ 照留）；{_OFF_FIX}"}) return out @@ -1277,12 +1337,14 @@ def _blocking_issues( *, 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, chat_user: bool = False, analysis_tools: bool = True, + off_subgraphs: set[str] | frozenset[str] = frozenset(), ) -> list[dict[str, Any]] | None: """按运行时同一套校验，挑出会挡住运行的问题。图本身不成形时返回 None。 chat_user：图要以开登录后 user 角色的身份在问数据里发起（D-c），使用者不能用的节点、工具、配置也挡（_chat_user_issues）。 - analysis_tools 为假（评测「关掉分析工具」那一臂）：图里还绑着分析工具的也挡（_analysis_off_issues）。 + analysis_tools 为假（评测「关掉分析工具」那一臂）：图里还用着分析能力的也挡（_analysis_off_issues：绑了分析工具、分析卡， + off_subgraphs 里那几个用了分析能力的子工作流节点，由调用方先用 analysis_off_subgraphs 读好）。 限定了数据源时，用了范围外的库也算：用户点了「只查这个库」，拿别的库的数 回答他，比跑不起来更糟——答案看起来是对的。 @@ -1313,7 +1375,7 @@ def _blocking_issues( + _issue_dicts(scope_issues(list(nodes.values()), scope), "datasource_out_of_scope") + [i for i in authored if i["level"] == "error"] + (_chat_user_issues(nodes, edges) if chat_user else []) - + ([] if analysis_tools else _analysis_off_issues(nodes))) + + ([] if analysis_tools else _analysis_off_issues(nodes, off_subgraphs))) if level: from app.engine.governance import lint_for_publish @@ -1822,7 +1884,8 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None, analysis_ analysis_tools 为假是评测「关掉分析工具」那一臂（分析工具开、关对照，chat/drive.TurnDriver 的 analysis_tools 透传过来）： 工具清单不列五个分析工具，节点手册、问数据的包装里只讲分析能力的那几段不给，问题意图那条自查不提；图里照样绑了分析工具 - 的按「绑定的工具不存在」挡（_analysis_off_issues）。缺省为真，和以前一字不差。 + 的、有分析卡的、嵌套的子工作流里用了分析能力的，照「绑定的工具不存在」那一套挡回去改（_analysis_off_issues）。缺省为真， + 和以前一字不差。 final 带 next（chat/turn.next_step：run 运行、ready 只建不跑、fail 判失败），问数据页照它走。最后多交一条 usage （providers/usage.py）：{op: usage, by_model, input_tokens, output_tokens, calls, cost_usd, @@ -1943,8 +2006,9 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None, analysis_ # 循环 / 分支条件里套了 {{ }}），前面的步骤白跑，报错还停在半路 async for op in _with_heartbeat(gen.repair(), phase="repairing"): yield op + off_subgraphs = set() if analysis_tools else await analysis_off_subgraphs(nodes) remaining = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline, checkers=checkers, - chat_user=chat_user, analysis_tools=analysis_tools) + chat_user=chat_user, analysis_tools=analysis_tools, off_subgraphs=off_subgraphs) # 工具绑定变化不挡运行，但要在自查这一步就说出来：工具被改没了的图照样 # 能跑，只是跑出来的是模型「假设」查过库的答案 changes = tool_changes(baseline, list(nodes.values())) @@ -1981,7 +2045,7 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None, analysis_ # 使用者发起不了的图（D-c）：自查没改好的照实列出，界面据此不运行、说清原因 issues += _chat_user_issues(nodes, edges) if chat_user else [] # 关掉分析工具的那一臂里还绑着分析工具的（自查没改好）：同上，照实列出，不运行 - issues += [] if analysis_tools else _analysis_off_issues(nodes) + issues += [] if analysis_tools else _analysis_off_issues(nodes, off_subgraphs) # 跳过的节点要说出来，不能安静地少一步。code 给前端认：这一类要单独 # 提示「少了一步」，不能和普通校验警告混在一起 issues += [ diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py index 8e207c4d..5d19c362 100644 --- a/backend/app/api/copilot_gen.py +++ b/backend/app/api/copilot_gen.py @@ -737,6 +737,8 @@ class GraphChecker: self._coverage_given = False #: 这一轮给不给分析工具（评测「关掉分析工具」那一臂为假） self.analysis_tools = analysis_tools + #: 关掉分析工具时，用了分析能力的子工作流节点（issues 每次先读好） + self._off_subgraphs: set[str] = set() #: 用户问题的分析意图（问为什么、怎么变了、是否异常、趋势……），不是分析类问题、或者这一轮没有分析工具时为 None self.intent = analysis_intent(question) if question and analysis_tools else None #: 「分析类问题、Agent 没绑分析工具」已经交回模型改过一轮：之后不再挡，同 _coverage_given @@ -747,9 +749,15 @@ class GraphChecker: return _blocking_issues(self.applier.nodes, self.applier.edges, self.scope, sources=self.sources, baseline=self.baseline, level=self.level, defaults=self.defaults, - checkers=self.checkers, chat_user=self.chat_user, analysis_tools=self.analysis_tools) + checkers=self.checkers, chat_user=self.chat_user, analysis_tools=self.analysis_tools, + off_subgraphs=self._off_subgraphs) async def issues(self) -> list[dict[str, Any]]: + if not self.analysis_tools: + # 关掉分析工具的那一臂：嵌套的子工作流里用没用分析能力要读库，先读好，blocking 照它挡 + from app.api.copilot import analysis_off_subgraphs + + self._off_subgraphs = await analysis_off_subgraphs(self.applier.nodes) errors = self.blocking() refused = _refused_issues(self.applier) if self.known is None: @@ -836,14 +844,24 @@ class GraphChecker: 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 里说一句为什么"}] + "for_model": analysis_gaps_fix(asked=asked, nid=nid)}] + + +def analysis_gaps_fix(*, asked: str = "问题问的是变化从哪来（「把上周销售额的变化拆开」）", nid: str = "查库的 agent", + names: str | None = None) -> str: + """「分析类问题、查库的 Agent 没绑分析工具」交回模型的那一句（analysis_gaps 的 for_model）。缺省的一套只是拿去算 + 提示词指纹（app/eval/runner.py 的 prompt_fingerprint）：改了这句话，评测比两次时就标「提示词变了」。""" + if names is None: + from app.analysis.contract import TOOL_NAMES + + names = "、".join(TOOL_NAMES[op] for op in ANALYSIS_TOOL_ORDER) + return (f"{asked}。按「分析类问题」的形状改：给查库的 agent「{nid}」的 tools 加上 {names}" + "（db_query__、db_schema__ 照留），平台会把分析方法注入它的 system，prompt 里写清要回答的问题、" + "比哪两期就行；report 的 instructions 让它按分析结论写、判断直接引用判定语。问的其实不是原因、" + "异常、趋势这类、确实用不上分析的，不改这一处，在 done 的 explanation 里说一句为什么") #: 自查时一条 SQL 的覆盖检查最多花多少秒（data/coverage.py）：搭图要等它，比查询时的 BUDGET_S 短 diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py index 6d688e14..fe5b959f 100644 --- a/backend/app/eval/runner.py +++ b/backend/app/eval/runner.py @@ -100,8 +100,11 @@ def leader_of(run: Any) -> str: def prompt_fingerprint() -> str: """助手提示词的指纹：节点手册、输出协议、问数据的请求包法、重来的两种措辞、复核的规则；平台的分析方法（分析工具的 guide：id 带版本，连同全文）、写作者的分析规则（写作目录里分析结论那一段、判断词的写法、重写时补的那句）和说法表的 - 版本。改了任何一段，和上一次比时就标「提示词变了」。""" + 版本；自查判问题意图的词表（版本，改词表要升）和「分析类问题、查库的 Agent 没绑分析工具」时交回模型的那一句。 + 改了任何一段，和上一次比时就标「提示词变了」。""" + from app.analysis.intent import INTENT_VERSION from app.api import copilot + from app.api.copilot_gen import analysis_gaps_fix from app.chat import turn from app.core.artifact_store import canonical_json, content_hash from app.engine import analysis_claims, evidence, review @@ -113,7 +116,8 @@ def prompt_fingerprint() -> str: str(getattr(review, "REVIEW_SYSTEM", "") or getattr(review, "SYSTEM_PROMPT", "")), METHOD_GUIDE.id, METHOD_GUIDE.text, evidence.ANALYSIS_PROMPT, evidence.ANALYSIS_SEE_ONLY_PROMPT, analysis_claims.ANALYSIS_RULES_PROMPT, analysis_claims.ANALYSIS_NONE_PROMPT, - analysis_claims.ANALYSIS_REPAIR_PROMPT, analysis_claims.LEXICON_VERSION] + analysis_claims.ANALYSIS_REPAIR_PROMPT, analysis_claims.LEXICON_VERSION, + INTENT_VERSION, analysis_gaps_fix()] return content_hash(canonical_json(parts))[:16] diff --git a/backend/tests/test_analysis_intent.py b/backend/tests/test_analysis_intent.py index 811ef234..d906d87e 100644 --- a/backend/tests/test_analysis_intent.py +++ b/backend/tests/test_analysis_intent.py @@ -1,7 +1,8 @@ """问题意图（app/analysis/intent.py）：一句问题是不是分析类问题，给助手搭问数据的图时自查用。 只收明确的问法、宁可漏判：漏判只是少一条自查提醒，误判要让助手多改一轮。所以反例和正例一样多——问一个字段（退款原因）、 -问业务状态（异常订单）、问服务（系统正常吗）、筛选（波动超过 10%）、问时长（持续几天）都不算。问法都是虚构的零售场景。 +问业务状态（异常订单）、问服务（系统正常吗）、问构成（谁贡献最大、哪个门店销售额最多）、问这批数据本身（今天的数据正常 +吗）、筛选（波动超过 10%）、问时长（持续几天）都不算。问法都是虚构的零售场景。 """ from __future__ import annotations @@ -23,7 +24,7 @@ POSITIVE = [ ("这周和上周比，销售额怎么变了", "source"), ("上周各渠道的销售额有什么变化", "source"), ("哪个地区拖累了整体销售额", "source"), - ("谁贡献最大", "source"), + ("哪个地区对销售额下降贡献最大", "source"), ("增长主要来自哪个地区", "source"), ("各地区对销售额增长的贡献是多少", "source"), ("销售额增长的驱动因素有哪些", "source"), @@ -70,6 +71,15 @@ NEGATIVE = [ "退款原因有哪些", "按取消原因统计上周的订单数", "订单主要来自哪些渠道", + # 问构成（哪个最多）：同句没有变化的说法，不算来源 + "谁贡献最大", + "哪个门店贡献的销售额最多", + "哪个门店贡献最大", + # 「数据」当主语：问的是这批数据本身，不是那一项的数变没变 + "数据是否正常", + "今天的数据正常吗", + "上周的数据有没有异常", + "数据有没有异常", "有多少异常订单", "有没有异常记录", "列出状态为异常的订单", @@ -109,7 +119,7 @@ def test_other_questions_are_left_alone(question): def test_the_lexicon_is_versioned_and_every_kind_has_a_label(): - assert INTENT_VERSION == 1 + assert INTENT_VERSION == 2 assert {kind for _, kind in POSITIVE} == set(KINDS) found = analysis_intent("上周销售额为什么下降了") assert isinstance(found, Intent) and found.version == INTENT_VERSION and found.label == "原因" diff --git a/backend/tests/test_copilot_analysis.py b/backend/tests/test_copilot_analysis.py index 6f4aae53..89ffa4c0 100644 --- a/backend/tests/test_copilot_analysis.py +++ b/backend/tests/test_copilot_analysis.py @@ -195,14 +195,32 @@ async def test_the_rule_stays_quiet_otherwise(question, graph): assert checker.analysis_gaps(level="warning") == [] -def _ops(tools: list[str]) -> list[dict]: - nodes, edges = _graph(tools=tools) +def _graph_ops(nodes: dict, edges: list[dict]) -> list[dict]: 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 _ops(tools: list[str]) -> list[dict]: + return _graph_ops(*_graph(tools=tools)) + + +def _card_ops(*, subgraph: str | None = None) -> list[dict]: + """带分析卡的图：入口 → 查库的 Agent → 分析卡 → 报告 → 出口。subgraph 给了就把分析卡换成一个嵌套它的子工作流节点。""" + middle = (_n("check", "analysis", items=[{"id": "wow", "op": "baseline", "from": "q", + "params": {"period": "week", "measure": "销售额"}}]) + if subgraph is None else _n("sub", "subgraph", workflow_id=subgraph)) + nodes = {"start": _n("start", "input"), + "q": _n("q", "tool", tool="db_query__shop", args={"sql": "SELECT 1"}), + "bot": _n("bot", "agent", prompt="{{ input.question }}", tools=["db_query__shop", "db_schema__shop"]), + middle["id"]: middle, + "write": _n("write", "report", instructions="回答：{{ input.question }}"), + "out": _n("out", "output", fields=[{"name": "回答", "value": "{{ nodes.write.text }}"}])} + order = ["start", "q", middle["id"], "write", "out"] + return _graph_ops(nodes, [{"source": a, "target": b} for a, b in zip(order, order[1:])]) + + def _scripted(first: list[dict], fix: list[dict], seen: list[str]): class Scripted: async def astream(self, messages): @@ -257,6 +275,8 @@ async def test_a_retry_reads_the_original_question(client, monkeypatch): # noqa def test_the_catalog_and_the_manual_leave_analysis_out_when_switched_off(): + """关掉分析工具的那一臂：工具清单里没有五个分析工具，节点手册里一处都不提分析卡、分析工具、analysis_ 开头的配置键和 + 标记——助手不知道有这回事，也不会被教着去用。开着时一字不差（node_reference() 就是 NODE_REFERENCE）。""" assert all(f"- {name}（" in _tool_catalog([]) for name in TOOLS) off = _tool_catalog([], analysis_tools=False) assert not [line for line in off.splitlines() if line.startswith("- analysis_")] @@ -264,10 +284,15 @@ def test_the_catalog_and_the_manual_leave_analysis_out_when_switched_off(): assert node_reference() is NODE_REFERENCE manual = node_reference(analysis_tools=False) assert not any(name in manual for name in TOOLS) - # 去掉的是 agent 段里「分析类问题绑分析工具」那一条、分析卡一整条、搭图规则第 7 条；报告段的引用写法、受管规则照留 + for gone in ("分析卡", "分析工具", "analysis_"): + assert gone not in manual, gone + # 去掉的是 agent 段里「分析类问题绑分析工具」那一条、分析卡一整条、搭图规则第 7 条；报告段里分析结论的来处和引用写法、 + # analysis_claims / analysis_scope 两个配置、受管规则里分析卡的那一句也不给（关的那一臂没有分析结论可引） assert "- analysis：" not in manual and "\n7. " not in manual and "**问题问的是为什么" not in manual - # 别的段落一字不少：只去掉那三段 - for kept in (" **默认不要写 max_steps", "- dual_check：", "\n模型字段", "- report：", "[[a:A1]]", "analysis_claims"): + assert "[[a:" not in manual and "判定语" not in manual + # 别的段落一字不少：报告、双盲复算、反方审稿、受管规则里和分析无关的几条照留 + for kept in (" **默认不要写 max_steps", "- dual_check：", "\n模型字段", "- report：", "- challenge：", + "claims: \"require_citation\"", "[[m:指标id]]", "[[see:m:指标id,Q1,K1]]", "受管级别"): assert kept in manual, kept request = GenerateIn(instruction="上周销售额为什么下降", intent="answer") on, off_text = _user_message(request, patch=False), _user_message(request, patch=False, analysis_tools=False) @@ -318,6 +343,86 @@ async def test_switched_off_the_copilot_sees_no_analysis_and_bound_tools_are_blo assert final["next"] == "fail" and [i["level"] for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] == ["error"] +async def test_switched_off_an_analysis_card_is_blocked_and_removed(monkeypatch): + """分析卡也挡：关掉分析工具的那一臂，图里有分析卡就交回模型去掉它、把上下游直接连起来。改好了照常运行。""" + seen: list[list[tuple[str, str]]] = [] + fix = [{"op": "remove_node", "id": "check"}, + {"op": "add_edge", "edge": {"source": "bot", "target": "write"}}, + {"op": "done", "explanation": "去掉了分析卡"}] + _use(monkeypatch, _recording(_card_ops(), fix, seen)) + got = await _ops_of("上周销售额为什么下降", analysis_tools=False) + rounds = [op for op in got if op["op"] == "check" and op["status"] == "repairing"] + [issue] = [i for r in rounds for i in r["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + assert (issue["level"], issue["node_id"], issue["field"]) == ("error", "check", None) + assert issue["message"] == "这次不能用分析卡：「check」是分析卡，这一轮没有分析能力可用，请去掉它" + # 交回模型的那一句守规矩：说清去掉它、上下游接回去、数怎么来；关的那一臂手册里没有「分析卡」这个说法， + # 所以指节点说它是 analysis 节点 + assert "「check」是 analysis 节点" in issue["for_model"] and "remove_node" in issue["for_model"] + assert "去掉它" in issue["for_model"] and "写进查询" in issue["for_model"] + [final] = [op for op in got if op["op"] == "final"] + assert final["next"] == "run", final["issues"] + assert "check" not in {n["id"] for n in final["graph"]["nodes"]} + assert not [i for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + + # 改不好：照实列出、不运行（图里还留着分析卡） + seen.clear() + _use(monkeypatch, _recording(_card_ops(), [{"op": "done", "explanation": "x"}], seen)) + got = await _ops_of("上周销售额为什么下降", analysis_tools=False) + [final] = [op for op in got if op["op"] == "final"] + assert final["next"] == "fail" + [left] = [i for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + assert (left["level"], left["node_id"]) == ("error", "check") + + +async def test_switched_off_a_subgraph_that_uses_analysis_is_blocked(monkeypatch, tmp_path): + """钉了版本、里面用了分析能力的子工作流（这里是里面有分析卡）也挡：这一轮用不了它。里面没有分析能力的子工作流光看外层 + 看不出来，照常运行。""" + from app.db.base import SessionLocal + from app.db.models import Workflow + + async def make(name: str, graph: dict) -> str: + async with SessionLocal() as session: + row = Workflow(name=name, graph=graph, version=1, published_version=1) + session.add(row) + await session.commit() + return row.id + + inner_card = _graph(tools=("db_query__shop", "db_schema__shop"), card=True)[0] + plain_nodes = _graph(tools=("db_query__shop", "db_schema__shop"))[0] + with_card = await make("异动分析（虚构）", {"nodes": list(inner_card.values()), + "edges": [{"source": a, "target": b} + for a, b in zip(inner_card, list(inner_card)[1:])]}) + without_card = await make("只查数（虚构）", {"nodes": list(plain_nodes.values()), + "edges": [{"source": a, "target": b} + for a, b in zip(plain_nodes, list(plain_nodes)[1:])]}) + seen: list[list[tuple[str, str]]] = [] + fix = [{"op": "remove_node", "id": "sub"}, + {"op": "add_edge", "edge": {"source": "q", "target": "write"}}, + {"op": "done", "explanation": "去掉了子工作流"}] + _use(monkeypatch, _recording(_card_ops(subgraph=with_card), fix, seen)) + got = await _ops_of("上周销售额为什么下降", analysis_tools=False) + [repairing] = [op for op in got if op["op"] == "check" and op["status"] == "repairing"] + [issue] = [i for i in repairing["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + assert (issue["node_id"], issue["field"]) == ("sub", "workflow_id") + assert "这次不能用这个子工作流" in issue["message"] and "分析卡或分析工具" in issue["message"] + assert "去掉这个 subgraph 节点" in issue["for_model"] + [final] = [op for op in got if op["op"] == "final"] + assert final["next"] == "run", final["issues"] + assert "sub" not in {n["id"] for n in final["graph"]["nodes"]} + assert not [i for i in final["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + + # 里面没分析能力的子工作流不挡；开着分析工具时，有分析卡的子工作流也不挡 + seen.clear() + _use(monkeypatch, _recording(_card_ops(subgraph=without_card), [], seen)) + got = await _ops_of("上周销售额为什么下降", analysis_tools=False) + assert not [i for op in got if op["op"] == "check" and op["status"] == "repairing" + for i in op["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + _use(monkeypatch, _recording(_card_ops(subgraph=with_card), [], seen)) + got = await _ops_of("上周销售额为什么下降", analysis_tools=True) + assert not [i for op in got if op["op"] == "check" and op["status"] == "repairing" + for i in op["issues"] if i["code"] == ANALYSIS_TOOLS_OFF] + + async def test_switched_on_bound_analysis_tools_are_fine(monkeypatch): """缺省开着：绑了分析工具不挡，和以前一样直接运行。""" seen: list[list[tuple[str, str]]] = [] diff --git a/backend/tests/test_eval_analysis_runs.py b/backend/tests/test_eval_analysis_runs.py index e34e3e04..f03a817b 100644 --- a/backend/tests/test_eval_analysis_runs.py +++ b/backend/tests/test_eval_analysis_runs.py @@ -342,21 +342,24 @@ def plain_graph() -> dict: class SwitchCopilot: - """助手剧本：给的工具清单里有分析工具（开着的那一臂）就搭「查库 → 分析卡 → 报告」，没有（关掉的那一臂）就搭「查库 → - 报告」——照给它看的用。systems 记下每次建图收到的 system。""" + """助手剧本：两臂都先照「查库 → 分析卡 → 报告」搭——关掉分析工具的那一臂靠真代码（answer_ops 的自查）挡回去， + 替身不替它判断。自查交回来时（只有关的那一臂会有）去掉分析卡、把查库直接接到报告。systems 记下每次建图收到的 system。""" def __init__(self) -> None: self.systems: list[str] = [] async def astream(self, messages): if "上线前自查" in messages[1][1]: - yield _line({"op": "done"}) + yield _line({"op": "remove_node", "id": "card"}) + yield _line({"op": "add_edge", "edge": {"source": "q", "target": "write"}}) + # 这一臂没有分析结论可引：报告只按查询结果写（报告撰写的剧本认这个记号，见 use_writers） + yield _line({"op": "update_node", "id": "write", "config": {"instructions": f"{PLAIN_MARK}上周的周报"}}) + yield _line({"op": "done", "explanation": "这一轮没有分析工具，去掉分析卡"}) return role, system = messages[0] assert role == "system" self.systems.append(system) - graph = answer_graph() if "analysis_decompose" in system else plain_graph() - for line in _graph_lines(graph, "看上周销售额"): + for line in _graph_lines(answer_graph(), "看上周销售额"): yield line diff --git a/backend/tests/test_eval_runner.py b/backend/tests/test_eval_runner.py index 2b074787..b809ce19 100644 --- a/backend/tests/test_eval_runner.py +++ b/backend/tests/test_eval_runner.py @@ -534,6 +534,30 @@ async def test_the_estimate_says_it_cannot_price_instead_of_zero(): assert got["total_usd"] == 0.0 and not got.get("at_least") and not got.get("unpriced_calls"), "目录价是 0 的才是真不花钱" +def test_the_prompt_fingerprint_covers_the_intent_lexicon_and_the_self_check_sentence(monkeypatch): + """提示词指纹认得出「问题意图词表变了」和「自查交回模型的那一句变了」：改词表要升 INTENT_VERSION，改那句话它自己 + 就带在里面。两者任一变了，两次评测比的时候标「提示词变了」。""" + from app.analysis import intent + from app.api import copilot_gen + + before = runner.prompt_fingerprint() + assert before == runner.prompt_fingerprint(), "同一份提示词指纹一样" + monkeypatch.setattr(intent, "INTENT_VERSION", intent.INTENT_VERSION + 1) + assert runner.prompt_fingerprint() != before, "词表升版本要认得出" + monkeypatch.undo() + + real = copilot_gen.analysis_gaps_fix + monkeypatch.setattr(copilot_gen, "analysis_gaps_fix", + lambda **kw: real(**kw).replace("按「分析类问题」的形状改", "换一种形状改")) + assert runner.prompt_fingerprint() != before, "自查那句话变了要认得出" + monkeypatch.undo() + + # 那句话里写着两条东西：分析类问题的形状、以及不改这一处时在说明里讲清为什么 + said = copilot_gen.analysis_gaps_fix(asked="问题问的是原因（「为什么下降」）", nid="bot") + assert said.startswith("问题问的是原因（「为什么下降」）。按「分析类问题」的形状改：给查库的 agent「bot」的 tools 加上") + assert "analysis_decompose" in said and "done 的 explanation" in said + + def test_the_judge_role_keeps_its_unpriced_calls(): """按角色的用量里裁判那一份以前只有金额和次数：裁判模型不在价格目录里时写成 $0。""" usage = runner.judge_usage({"claims": {"cost_usd": 0.0, "calls": 2, "priced": False},