diff --git a/backend/app/api/bundle.py b/backend/app/api/bundle.py new file mode 100644 index 00000000..27a3ef13 --- /dev/null +++ b/backend/app/api/bundle.py @@ -0,0 +1,10 @@ +"""可验证报告：把一次运行的报告和证据导出成可独立核对的 HTML 包。 + +目前只登记了路由前缀，接口由可验证报告的工作包补上。路径挂在 /api/runs/{run_id}/… 下面； +注意 runs.py 里已经有 /{run_id}/verify（封存复验），别撞名。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/runs", tags=["bundle"]) diff --git a/backend/app/api/compare.py b/backend/app/api/compare.py new file mode 100644 index 00000000..582d926c --- /dev/null +++ b/backend/app/api/compare.py @@ -0,0 +1,10 @@ +"""运行对比：两次运行按「变化三分法」对照。 + +目前只登记了路由前缀，接口由运行对比的工作包补上。路径挂在 /api/runs/{run_id}/… 下面； +这个路由登记在 runs.router 之前，以后加 /api/runs/compare 这类固定路径也不会被 /{run_id} 吃掉。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/runs", tags=["compare"]) diff --git a/backend/app/api/inquiry.py b/backend/app/api/inquiry.py new file mode 100644 index 00000000..829b228c --- /dev/null +++ b/backend/app/api/inquiry.py @@ -0,0 +1,9 @@ +"""审批质询：审批人在批准前就待审的报告提出质询。 + +目前只登记了路由前缀，接口由审批质询的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/runs", tags=["inquiry"]) diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py new file mode 100644 index 00000000..718eb8b8 --- /dev/null +++ b/backend/app/api/schedules.py @@ -0,0 +1,9 @@ +"""定时任务：按周期自动发起工作流运行。 + +目前只登记了路由前缀，接口由定时任务的工作包补上；后台的调度循环在 app/scheduler 里。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/schedules", tags=["schedules"]) diff --git a/backend/app/api/whatif.py b/backend/app/api/whatif.py new file mode 100644 index 00000000..364ee3e1 --- /dev/null +++ b/backend/app/api/whatif.py @@ -0,0 +1,9 @@ +"""数字推演：在一次运行的基础上改动输入，推演指标会怎样变化，不改原运行。 + +目前只登记了路由前缀，接口由数字推演的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/api/runs", tags=["whatif"]) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 7055d754..d32b1157 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -87,6 +87,12 @@ class Settings(BaseSettings): max_upload_mb: int = 50 max_concurrent_runs: int = 20 + # --- 定时任务 --- + # 后台调度循环开不开（AGENTLAB_SCHEDULER=0 / false / off 关掉）。拷一份数据目录另起的服务 + # （检查脚本的 --stacks、沙箱）必须关：库里的定时任务会照常到点发起运行、发出 webhook， + # 拷贝里的任务和原库的是同一批，等于同一件事做两遍 + scheduler: bool = True + # --- 工具 --- http_tool_timeout: int = 20 http_tool_max_bytes: int = 2_000_000 @@ -117,6 +123,15 @@ class Settings(BaseSettings): return [part.strip() for part in text.split(",") if part.strip()] return value + @field_validator("scheduler", mode="before") + @classmethod + def _blank_is_default(cls, value: object) -> object: + # AGENTLAB_SCHEDULER= 写成空值等于没设（默认开）。其余写法照 pydantic 的布尔解析： + # 0 / false / off / no 关，1 / true / on / yes 开，认不出的值启动时就报错，不猜 + if isinstance(value, str) and not value.strip(): + return True + return value + @field_validator("web_dist", mode="before") @classmethod def _blank_is_unset(cls, value: object) -> object: diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 05ef4472..491d59af 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -11,7 +11,9 @@ from langgraph.graph import END, START, StateGraph from app.core.errors import describe_exception, raw_detail from app.core.events import EventType from app.engine.context import NodeContext, NodeError, RunContext -from app.engine.nodes import control, human, io, knowledge, llm, merge, metrics, multi, report, tools +from app.engine.nodes import ( + challenge, control, dual_check, human, io, knowledge, llm, merge, metrics, multi, report, tools, +) from app.engine.schema import GraphNode, GraphSpec, NodeType, back_edges, innermost_loops from app.engine.state import GraphState @@ -37,6 +39,8 @@ RUNNERS: dict[NodeType, NodeRunner] = { NodeType.METRICS: metrics.run_metrics, NodeType.RETRIEVE: knowledge.run_retrieve, NodeType.REPORT: report.run_report, + NodeType.DUAL_CHECK: dual_check.run_dual_check, + NodeType.CHALLENGE: challenge.run_challenge, } # 这些节点的出边由"决定"路由，不能直接连死 diff --git a/backend/app/engine/labels.py b/backend/app/engine/labels.py index 073934e2..721bb032 100644 --- a/backend/app/engine/labels.py +++ b/backend/app/engine/labels.py @@ -22,6 +22,7 @@ TYPE_LABEL: dict[str, str] = { "branch": "条件分支", "loop": "循环", "subgraph": "子工作流", "memory": "长期记忆", "retrieve": "知识检索", "transform": "数据整形", "human": "人工审批", "validate": "结构校验", "metrics": "口径卡", "report": "报告撰写", + "dual_check": "双盲复算", "challenge": "反方审稿", } #: 各类节点共用的字段名（nodeDefs.ts）。同一个键在不同节点上叫法不同的，见 _FIELD_BY_TYPE diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py new file mode 100644 index 00000000..a2348aad --- /dev/null +++ b/backend/app/engine/nodes/challenge.py @@ -0,0 +1,21 @@ +"""反方审稿节点：让另一个模型站在反方，专挑上游报告的毛病，逐条提出反驳意见。 + +目前只是占位：图里放了这个节点也能运行，执行器原样跳过、记一条日志，不改任何状态。 +真正的审稿由后续工作包补上；没能回应的反驳通过节点产出里的 issuance_gaps 交给成果节点的 +出具契约降档（见 nodes/io.py 的 _upstream_issuance_gaps）。 +""" +from __future__ import annotations + +from typing import Any + +from app.core.events import EventType +from app.engine.context import NodeContext +from app.engine.state import GraphState + +#: 占位期间的跳过原因：节点产出和日志里都写这一句 +PENDING_REASON = "反方审稿功能开发中，本节点暂未生效，已跳过" + + +async def run_challenge(state: GraphState, ctx: NodeContext) -> dict[str, Any]: + ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON) + return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}} diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py new file mode 100644 index 00000000..5174ffd4 --- /dev/null +++ b/backend/app/engine/nodes/dual_check.py @@ -0,0 +1,21 @@ +"""双盲复算节点：另起一路、不看原结果独立算一遍，再和原结果逐项比对。 + +目前只是占位：图里放了这个节点也能运行，执行器原样跳过、记一条日志，不改任何状态。 +真正的复算和比对由后续工作包补上；比不上的项通过节点产出里的 issuance_gaps 交给成果节点的 +出具契约降档（见 nodes/io.py 的 _upstream_issuance_gaps）。 +""" +from __future__ import annotations + +from typing import Any + +from app.core.events import EventType +from app.engine.context import NodeContext +from app.engine.state import GraphState + +#: 占位期间的跳过原因：节点产出和日志里都写这一句 +PENDING_REASON = "双盲复算功能开发中，本节点暂未生效，已跳过" + + +async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]: + ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON) + return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}} diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index e289aa84..80102660 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -258,6 +258,8 @@ def _apply_contract( if isinstance(payload, dict) and payload.get("exhausted"): gaps.append(f"协作团队「{titles.get(node_id, node_id)}」用完 {payload.get('rounds', '?')} " "轮仍未完成，交付的是成员最后的回复") + # 上游节点自己报上来的缺口（双盲复算比不上、反方审稿的反驳没回应……）：只会降档，不会升档 + gaps.extend(_upstream_issuance_gaps(state, ctx)) tier = decide_tier( missing_required=missing_required, @@ -314,6 +316,35 @@ def _apply_contract( return issuance +def _upstream_issuance_gaps(state: GraphState, ctx: NodeContext) -> list[str]: + """上游节点产出里的 issuance_gaps（字符串列表），每条前面加上是哪个节点说的。 + + 这是给「把关类」节点留的通用口子：节点觉得这次的结论不该完整出具（双盲复算两路对不上、 + 反方审稿的反驳没有回应），在自己的产出里写 issuance_gaps，成果节点的出具契约照单收进缺口， + 出具随之降档。写的时候不用带节点名，这里统一加上「双盲复算「复核销售额」：」这样的前缀， + 和别的缺口一样读得出是谁报的。 + + 只认这个成果节点的上游：另一条支路上的节点管不到这份成果。不是字符串、空白的条目跳过—— + 这个键谁都能写（数据整形的 JSON 模板也行），写坏了只当没写，不能让出具崩掉。 + """ + from app.engine.labels import type_label + + spec = ctx.run.spec + upstream = _ancestors(spec, ctx.node.id) + nodes = state.get("nodes") or {} + out: list[str] = [] + for node in spec.nodes: + payload = nodes.get(node.id) + if node.id not in upstream or not isinstance(payload, dict): + continue + reported = payload.get("issuance_gaps") + if not isinstance(reported, list): + continue + prefix = f"{type_label(node.type)}「{node.title}」：" + out.extend(prefix + gap.strip() for gap in reported if isinstance(gap, str) and gap.strip()) + return out + + # -------------------------------------------------------------------------- # 引用模式：按报告文档逐段复核 # -------------------------------------------------------------------------- diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 43a28688..d3ae6325 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -37,6 +37,9 @@ class NodeType(StrEnum): VALIDATE = "validate" # 结构化校验 + 自动重试 METRICS = "metrics" # 口径卡：受控指标集，叙述层唯一合法的数字来源 REPORT = "report" # 报告撰写：模型只写引用标记，数字由系统从口径卡取出来渲染 + # 下面两种目前只是占位（执行器原样放行），各自的工作包补上真正的逻辑 + DUAL_CHECK = "dual_check" # 双盲复算：另起一路独立算一遍，和原结果比对 + CHALLENGE = "challenge" # 反方审稿：专挑报告的毛病，提出反驳意见 # 节点类型在画布上的叫法（TYPE_LABEL / type_label）从 labels.py 导入：报错和门禁文案用它， diff --git a/backend/app/main.py b/backend/app/main.py index 0faed210..8a1feb7a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,8 +9,8 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from app.api import ( - artifacts, catalog, conversations, copilot, datasources, evidence, governance, knowledge, runs, - sandbox, settings as settings_api, source_versions, table_imports, tools, workflows, + artifacts, bundle, catalog, compare, conversations, copilot, datasources, evidence, governance, inquiry, knowledge, + runs, sandbox, schedules, settings as settings_api, source_versions, table_imports, tools, whatif, workflows, ) from app.api.coded import CodedHTTPException, coded_handler from app.core.config import settings @@ -74,10 +74,18 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: from app.sandbox.manager import sandbox_manager as _sbm await _sbm.start_reaper() + # 定时任务的调度循环放在 run_manager 之后起：它到点要发起运行。拷来的数据目录（检查脚本的 + # --stacks）用 AGENTLAB_SCHEDULER=0 关掉，免得替原库把定时任务再执行一遍 + from app import scheduler + + if scheduler.scheduler_enabled(): + await scheduler.start() logger.info("AgentLab 就绪 · 数据目录 %s", settings.data_dir) try: yield finally: + # 先停调度，再停运行：停运行的途中不该再有新运行被发起 + await scheduler.stop() await run_manager.shutdown() from app.sandbox.manager import sandbox_manager @@ -102,6 +110,12 @@ app.add_middleware( for router in ( workflows.router, + # 这几个也挂在 /api/runs 下面，要在 runs.router 之前登记：以后加的固定路径（/api/runs/compare 之类） + # 才不会被 runs 的 /{run_id} 当成运行 id 吃掉。/{run_id}/… 的子路径放在哪边都不冲突 + compare.router, + bundle.router, + inquiry.router, + whatif.router, runs.router, runs.approvals_router, evidence.router, @@ -117,6 +131,7 @@ for router in ( conversations.router, artifacts.router, governance.router, + schedules.router, # 按配方导入的 /api/datasources/imports/… 要在 datasources 的通配 /{source_id} 之前登记 table_imports.router, source_versions.router, diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py new file mode 100644 index 00000000..e6da00b8 --- /dev/null +++ b/backend/app/scheduler/__init__.py @@ -0,0 +1,23 @@ +"""定时任务的后台调度：到点按定时任务的设置发起工作流运行。 + +目前 start / stop 都是空的，调度循环由定时任务的工作包补上。服务启动时（main.py 的 lifespan） +只在 scheduler_enabled() 时调 start，关闭时总是调 stop——stop 要能在没 start 过时安全返回。 +""" +from __future__ import annotations + +from app.core.config import Settings, settings + + +def scheduler_enabled(config: Settings | None = None) -> bool: + """调度循环开不开：AGENTLAB_SCHEDULER，默认开，0 / false / off 关（见 core/config.py）。 + + config 只给测试用：settings 是模块级单例，改环境变量不会让它重读。""" + return bool((config or settings).scheduler) + + +async def start() -> None: + """起调度循环。调度循环在后续工作包里实现，现在什么都不做。""" + + +async def stop() -> None: + """停调度循环，等手上的那一轮收尾。现在什么都不做；没 start 过也要能安全调用。""" diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 42722887..c8f7d71d 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -20,6 +20,8 @@ _TMP = Path(tempfile.mkdtemp(prefix="agentlab-tests-")) os.environ["AGENTLAB_DATA_DIR"] = str(_TMP) # 别把真实的 key 带进测试环境 os.environ.setdefault("AGENTLAB_SECRET_KEY", "test-only-not-a-real-key") +# 定时任务的调度循环不在测试里起：走 lifespan 的测试不该到点替人发起运行。要测调度的直接调它 +os.environ.setdefault("AGENTLAB_SCHEDULER", "0") import pytest # noqa: E402 diff --git a/backend/tests/test_issuance_upstream_gaps.py b/backend/tests/test_issuance_upstream_gaps.py new file mode 100644 index 00000000..3427336d --- /dev/null +++ b/backend/tests/test_issuance_upstream_gaps.py @@ -0,0 +1,110 @@ +"""上游节点报上来的出具缺口（产出里的 issuance_gaps）：成果节点的出具契约照单收下，出具降档。 + +这是给双盲复算、反方审稿这类把关节点留的通用口子。节点只管在产出里写一句缺口， +不带节点名；契约统一加上「节点类型「节点名」：」的前缀，和别的缺口一样读得出是谁报的。 +没写这个键的图，出具和以前一模一样。 +""" +from __future__ import annotations + +import asyncio +import json + +import pytest +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine.runner import run_manager + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, + "data": {"label": label or nid, "config": config}} + + +def graph(gate: dict, *, side: dict | None = None) -> dict: + """口径卡 → 叙述 → 把关节点 → 成果。叙述里的数都回指得到口径卡，不加把关节点的话是完整出具。 + side 是挂在另一条支路上、不通向成果的节点。""" + nodes = [ + node("start", "input"), + node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=[{"id": "orders", "expression": "120"}]), + node("draft", "transform", mode="template", assign_to="summary", template="本周订单 120 单"), + gate, + node("out", "output", fields=[{"name": "结论", "value": "{{ vars.summary }}"}], + contract={"metrics_from": "card", "narrative": "{{ vars.summary }}"}), + ] + edges = [{"source": "start", "target": "card"}, {"source": "card", "target": "draft"}, + {"source": "draft", "target": gate["id"]}, {"source": gate["id"], "target": "out"}] + if side: + nodes.append(side) + edges.append({"source": "start", "target": side["id"]}) + return {"nodes": nodes, "edges": edges} + + +def gaps_node(nid: str, label: str, gaps: list) -> dict: + """用数据整形的 JSON 模板产出 issuance_gaps：和真正的把关节点产出同一个形状""" + return node(nid, "transform", label, mode="json", template=json.dumps({"issuance_gaps": gaps}, ensure_ascii=False)) + + +async def finish(spec: dict) -> Run: + run = await run_manager.start(graph=spec, input_payload={}) + for _ in range(200): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed"): + break + assert row.status == "succeeded", row.error + return row + + +async def issuance_event(run_id: str) -> dict: + async with SessionLocal() as session: + event = (await session.execute( + select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == "issuance") + )).scalars().one() + return event.data + + +async def test_upstream_gap_degrades_issuance_and_names_the_node(): + row = await finish(graph(gaps_node("recheck", "复核销售额", ["两路结果不一致：订单数 120 对 118", " ", 42]))) + + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded" + # 空白和不是字符串的条目跳过，只收那一句；前缀是节点类型的叫法加节点名 + assert issuance["gaps"] == ["数据整形「复核销售额」：两路结果不一致：订单数 120 对 118"] + # 数字照常回指：降档只因为这条缺口 + assert issuance["unmatched_numbers"] == [] and issuance["matched_numbers"] == 1 + assert (await issuance_event(row.id))["gaps"] == issuance["gaps"] + + +async def test_stub_gate_nodes_pass_through_without_touching_issuance(): + """双盲复算、反方审稿目前是占位：放在图里照样能运行，出具不受影响""" + for kind in ("dual_check", "challenge"): + row = await finish(graph(node("gate", kind))) + issuance = row.output["_issuance"] + assert issuance["tier"] == "formal", (kind, issuance) + assert issuance["gaps"] == [] + async with SessionLocal() as session: + logs = (await session.execute( + select(RunEvent).where(RunEvent.run_id == row.id, RunEvent.type == "log", RunEvent.node_id == "gate") + )).scalars().all() + assert [e.data.get("level") for e in logs] == ["info"] + assert "开发中" in logs[0].data["message"] + + +async def test_without_issuance_gaps_nothing_changes(): + """没写 issuance_gaps 的节点，和别的支路上写了的节点，都不影响这份成果""" + plain = node("recheck", "transform", "复核销售额", mode="json", template='{"ok": true}') + row = await finish(graph(plain, side=gaps_node("elsewhere", "别处的检查", ["和这份成果无关"]))) + + issuance = row.output["_issuance"] + assert issuance["tier"] == "formal" + assert issuance["gaps"] == [] diff --git a/backend/tests/test_scheduler_flag.py b/backend/tests/test_scheduler_flag.py new file mode 100644 index 00000000..c0ebb94c --- /dev/null +++ b/backend/tests/test_scheduler_flag.py @@ -0,0 +1,44 @@ +"""AGENTLAB_SCHEDULER：定时任务的调度循环开不开。 + +默认开；0 / false / off 关。检查脚本的 --stacks 拷一份数据目录另起服务时靠它关掉， +否则拷贝里的定时任务会替原库再发起一遍运行、再发一遍 webhook。 +""" +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from app import scheduler +from app.core.config import Settings + + +def flag(monkeypatch, value: str | None) -> bool: + if value is None: + monkeypatch.delenv("AGENTLAB_SCHEDULER", raising=False) + else: + monkeypatch.setenv("AGENTLAB_SCHEDULER", value) + # 不读仓库里的 .env：测的是环境变量本身的解析 + return scheduler.scheduler_enabled(Settings(_env_file=None)) + + +@pytest.mark.parametrize("value", [None, "", " ", "1", "true", "on", "TRUE", "yes"]) +def test_enabled_by_default_and_when_switched_on(monkeypatch, value): + assert flag(monkeypatch, value) is True + + +@pytest.mark.parametrize("value", ["0", "false", "off", "False", "OFF", "no"]) +def test_switched_off(monkeypatch, value): + assert flag(monkeypatch, value) is False + + +def test_unrecognized_value_fails_at_startup(monkeypatch): + """认不出的写法不猜开还是关：启动时就报错""" + with pytest.raises(ValidationError): + flag(monkeypatch, "maybe") + + +async def test_start_and_stop_are_safe_noops(): + """调度循环还没实现：start / stop 什么都不做，没 start 过直接 stop 也不出错""" + await scheduler.stop() + await scheduler.start() + await scheduler.stop()