diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py index 9bdbf3fa..4ba3897e 100644 --- a/backend/app/engine/launch.py +++ b/backend/app/engine/launch.py @@ -87,7 +87,8 @@ async def launch_run( 参数和 POST /api/runs 的请求体一一对应(input 叫 input_payload);actor 是记在运行上的发起人。workflow_name 是没挂在 工作流上的探索运行记在运行上的名字(蓝图的试运行写蓝图名),不给就是「未保存的工作流」;挂在工作流上的用工作流的名字。 schedule_id 是发起它的定时任务,options 是整次运行沿用的选项快照,都原样交给 run_manager.start;图里有双盲复算、 - 反方审稿节点时 options 另加核查设置的快照(_with_verify_snapshot)。 + 反方审稿节点时 options 另加核查设置的快照(_with_verify_snapshot),有分析卡、或 Agent / 协作成员绑了分析工具时另加 + 分析设置的快照(_with_analysis_snapshot)。 """ # 没挂在工作流上(画布没保存就发起、问数据当场搭的)。老数据里是「临时图」,前端两种都认 name = workflow_name or "未保存的工作流" @@ -182,6 +183,7 @@ async def launch_run( raise LaunchRefused(400, "没有要运行的内容:请求中既没有工作流 ID,也没有工作流内容") await _refuse_missing_tools(session, graph) options = await _with_verify_snapshot(session, graph, options) + options = await _with_analysis_snapshot(session, graph, options) options, capability_changes = await _with_capability_snapshot( session, graph, options, published=published_capabilities) @@ -311,7 +313,43 @@ async def _with_verify_snapshot(session: AsyncSession, graph: dict[str, Any], async def _has_verify_nodes(session: AsyncSession, spec: GraphSpec) -> bool: - """图里有没有双盲复算、反方审稿节点,子工作流里的也算。 + """图里有没有双盲复算、反方审稿节点,子工作流里的也算。""" + return await _any_graph(session, spec, lambda graph: any(node.type in _VERIFY_NODES for node in graph.nodes)) + + +async def _with_analysis_snapshot(session: AsyncSession, graph: dict[str, Any], + options: dict[str, Any] | None) -> dict[str, Any] | None: + """图里(含子工作流)有分析卡、或 Agent / 协作成员绑了分析工具时,把分析设置(api/settings.py 的 analysis 一组: + 判定用的阈值、次数上限)快照进 options["analysis"](docs/design/analysis-round/00-总方案.md §3.10)。 + + 道理同 _with_verify_snapshot:分析在运行中途读阈值,审批停下期间有人改了设置,恢复后的那一段就按另一套阈值判,正式 + 运行也说不清当时按什么规则判的「主要来自」。分析工具、分析卡只读这一份(app/tools/analysis.py 的 run_snapshot), + 没有快照的现读设置。别的图 options 原样不动,run.started 和以前逐字相同;调用方已经给了 options["analysis"] 的照它的。 + """ + from app.api.settings import ANALYSIS_GROUP, analysis_settings + + if isinstance(options, dict) and ANALYSIS_GROUP in options: + return options + try: + spec = GraphSpec.model_validate(graph) + except Exception: # noqa: BLE001 - 图本身不成形交给 run_manager.start 报 + return options + if not await _any_graph(session, spec, _analyzes): + return options + return {**(options or {}), ANALYSIS_GROUP: await analysis_settings(session)} + + +def _analyzes(graph: GraphSpec) -> bool: + """这一张图(不含子工作流)会用到分析设置:有分析卡,或者有绑了分析工具的 Agent、协作成员(tool_bindings 的取法, + 节点上没写就取图级 defaults)。""" + from app.tools.analysis import is_analysis_tool + + return any(node.type == NodeType.ANALYSIS for node in graph.nodes) \ + or any(is_analysis_tool(name) for _, name in tool_bindings(graph)) + + +async def _any_graph(session: AsyncSession, spec: GraphSpec, test: Any) -> bool: + """这张图或它(一层层)套着的子工作流里,有没有一张让 test(图) 为真。 子工作流照运行时的取法读(同 runner._bound_sources):固定了版本的读那一版,没固定的读工作流现在的图。读不到、 图不成形的跳过——那个节点运行时自己会报错。同一个工作流(同一版)只读一次,互相引用也不会转圈。 @@ -320,7 +358,7 @@ async def _has_verify_nodes(session: AsyncSession, spec: GraphSpec) -> bool: queue: list[GraphSpec] = [spec] while queue: graph = queue.pop(0) - if any(node.type in _VERIFY_NODES for node in graph.nodes): + if test(graph): return True for node in graph.nodes: if node.type != NodeType.SUBGRAPH: diff --git a/backend/tests/test_analysis_settings.py b/backend/tests/test_analysis_settings.py index 229dddd3..0b782c5a 100644 --- a/backend/tests/test_analysis_settings.py +++ b/backend/tests/test_analysis_settings.py @@ -1,20 +1,31 @@ -"""分析设置(设置 → 偏好设置 → 分析,docs/design/analysis-round/01-算子与判定.md §10)。 +"""分析设置(设置 → 偏好设置 → 分析,docs/design/analysis-round/01-算子与判定.md §10)和发起时的快照。 - 设置接口的 analysis 一组:键、缺省、范围、叫法全取 app/analysis/contract.py;GET 按生效的样子给(存坏了的项按缺省), PUT 先查后写(不认识的键、类型、范围、两两之间的大小) +- 发起运行时快照进 run.options["analysis"]:只在图(含子工作流)里有分析卡、或 Agent / 协作成员绑了分析工具时写; + 别的图的 run.started 一字不变 """ from __future__ import annotations +import asyncio + import pytest from httpx import ASGITransport, AsyncClient +from sqlalchemy import select from app.analysis import contract from app.api.settings import ANALYSIS_GROUP, analysis_problem, analysis_values from app.db.base import SessionLocal -from app.db.models import Setting +from app.db.models import Run, RunEvent, Setting, Workflow +from app.engine.launch import launch_run from app.engine.runner import run_manager from app.main import app +#: 加 options 之前 run.started 就有的键(同 tests/test_run_options.py) +LEGACY_STARTED_KEYS = {"nodes", "resumed", "memory_scope", "collection", "approval_default", "tool_trust", + "agent_limits", "data_versions", "replay_protocol"} + + @pytest.fixture(autouse=True) async def engine_up(): await run_manager.setup() @@ -38,6 +49,14 @@ async def clean_group(): await session.commit() +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + # -------------------------------------------------------------------------- # 设置接口 # -------------------------------------------------------------------------- @@ -83,3 +102,78 @@ async def test_put_refuses_bad_values_and_stores_nothing(client): r = await client.put("/api/settings", json={"values": {ANALYSIS_GROUP: {"main_share": 0.6, "k_alert": 1.5}}}) assert r.status_code == 422 and contract.LABELS["k_alert"] in r.json()["detail"] assert (await client.get("/api/settings")).json()[ANALYSIS_GROUP] == contract.DEFAULTS + + +# -------------------------------------------------------------------------- +# 发起时的快照 +# -------------------------------------------------------------------------- + + +async def _started(run_id: str) -> dict: + for _ in range(200): + async with SessionLocal() as session: + found = (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "run.started"))).scalars().first() + if found is not None: + await run_manager.wait_idle(run_id) + return found.data + await asyncio.sleep(0.05) + raise AssertionError("没有 run.started") + + +async def _launch(graph, **kw) -> tuple[Run, dict]: + async with SessionLocal() as session: + run = await launch_run(session, graph=graph, actor="测试", **kw) + return run, await _started(run.id) + + +CARD = node("card", "analysis", items=[{"id": "wow", "op": "baseline", "from": "q", + "params": {"period": "week", "measure": "gmv"}}]) +QUERY = node("q", "tool", tool="calculator", args={"expression": "1 + 1"}) +OUT = node("out", "output", fields=[{"name": "r", "value": "完成"}]) + + +async def test_graphs_without_analysis_start_exactly_as_before(): + run, started = await _launch(chain(node("start", "input"), node("t", "transform", mode="template", template="x"), + OUT)) + assert set(started) == LEGACY_STARTED_KEYS, "没有分析的图,run.started 一个键都不多" + async with SessionLocal() as session: + assert (await session.get(Run, run.id)).options in (None, {}) + # 绑了别的工具的 Agent:有能力治理的快照(以前就有),没有分析这一组 + _, started = await _launch(chain(node("start", "input"), + node("bot", "agent", prompt="说一句", tools=["calculator"]), OUT)) + assert ANALYSIS_GROUP not in (started.get("options") or {}) + + +async def test_an_analysis_card_snapshots_the_settings(): + async with SessionLocal() as session: + session.add(Setting(key=ANALYSIS_GROUP, value={"main_share": 0.6})) + await session.commit() + run, started = await _launch(chain(node("start", "input"), QUERY, CARD, OUT)) + assert started["options"][ANALYSIS_GROUP] == {**contract.DEFAULTS, "main_share": 0.6} + async with SessionLocal() as session: + assert (await session.get(Run, run.id)).options[ANALYSIS_GROUP]["main_share"] == 0.6 + + +@pytest.mark.parametrize("graph", [ + chain(node("start", "input"), node("bot", "agent", prompt="为什么", tools=[contract.TOOL_NAMES["trend"]]), OUT), + chain(node("start", "input"), node("team", "supervisor", goal="为什么", agents=[ + {"name": "分析员", "tools": [contract.TOOL_NAMES["differ"]]}]), OUT), +]) +async def test_agents_and_members_with_analysis_tools_snapshot_the_settings(graph): + _, started = await _launch(graph) + assert started["options"][ANALYSIS_GROUP] == contract.DEFAULTS + + +async def test_a_card_inside_a_subgraph_counts_and_given_options_win(): + async with SessionLocal() as session: + inner = Workflow(name="异动分析(子流程)", graph=chain(node("start", "input"), QUERY, CARD, OUT)) + session.add(inner) + await session.commit() + inner_id = inner.id + outer = chain(node("start", "input"), node("sub", "subgraph", workflow_id=inner_id), OUT) + _, started = await _launch(outer) + assert started["options"][ANALYSIS_GROUP] == contract.DEFAULTS + # 调用方已经给了这一组的,照它的 + _, started = await _launch(outer, options={ANALYSIS_GROUP: {"max_calls": 2}}) + assert started["options"][ANALYSIS_GROUP] == {"max_calls": 2}