复核套件骨架(后端):先把后续几个工作包要用的位置占好,各包只改自己的文件。节点类型加双盲复算 dual_check、反方审稿 challenge,执行器是占位(记一条 info 日志后跳过,产出 skipped 和原因),校验按普通节点走,叫法进 labels,暂不写进助手的节点说明。出具契约留一个通用口子:成果节点上游任何节点产出里的 issuance_gaps(字符串列表)收进缺口,前面加上「节点类型「节点名」:」,出具随之降档、不会升档;只认这个成果的上游,写坏的条目跳过。路由先登记 schedules、compare、bundle、inquiry、whatif 五个空前缀,挂在 /api/runs 下的四个排在 runs.router 之前,以后加固定路径不会被 /{run_id} 吃掉。定时任务的调度先放空的 start / stop,AGENTLAB_SCHEDULER 默认开、0 / false / off 关,写进设置;启动时开着才 start,测试环境默认关

This commit is contained in:
ThearchyHelios
2026-10-04 23:40:20 +08:00
parent ec7f6a573d
commit a3058e7df8
17 changed files with 340 additions and 3 deletions
+10
View File
@@ -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"])
+10
View File
@@ -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"])
+9
View File
@@ -0,0 +1,9 @@
"""审批质询:审批人在批准前就待审的报告提出质询。
目前只登记了路由前缀,接口由审批质询的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。
"""
from __future__ import annotations
from fastapi import APIRouter
router = APIRouter(prefix="/api/runs", tags=["inquiry"])
+9
View File
@@ -0,0 +1,9 @@
"""定时任务:按周期自动发起工作流运行。
目前只登记了路由前缀,接口由定时任务的工作包补上;后台的调度循环在 app/scheduler 里。
"""
from __future__ import annotations
from fastapi import APIRouter
router = APIRouter(prefix="/api/schedules", tags=["schedules"])
+9
View File
@@ -0,0 +1,9 @@
"""数字推演:在一次运行的基础上改动输入,推演指标会怎样变化,不改原运行。
目前只登记了路由前缀,接口由数字推演的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。
"""
from __future__ import annotations
from fastapi import APIRouter
router = APIRouter(prefix="/api/runs", tags=["whatif"])
+15
View File
@@ -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:
+5 -1
View File
@@ -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,
}
# 这些节点的出边由"决定"路由,不能直接连死
+1
View File
@@ -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
+21
View File
@@ -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}}}
+21
View File
@@ -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}}}
+31
View File
@@ -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
# --------------------------------------------------------------------------
# 引用模式:按报告文档逐段复核
# --------------------------------------------------------------------------
+3
View File
@@ -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 导入:报错和门禁文案用它,
+17 -2
View File
@@ -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,
+23
View File
@@ -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 过也要能安全调用。"""
+2
View File
@@ -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
@@ -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"] == []
+44
View File
@@ -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()