mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 20:01:57 +00:00
ThearchyHelios
3c51c40e9d
多数节点（模型调用、Agent、报告撰写、多 Agent 协作、条件分支、结构校验、知识检索）、协作成员和全图默认的接入在检查器里没有下拉框，以前点名的接入坏了只能去设置里建一个同名接入。现在门禁接受接入表（governance.lint_for_publish 的 providers，接口层读好传进来，同数据目录 SQL 检查器），点名接入的问题成了门禁的一部分，自动修复给每一处一条要人选的修复：改用一个启用的接入，或不指定接入；选的接入不提供这一处写着的模型时模型一并去掉。一处一条（修复 id 带字段，复核只认那个字段），协作节点自己的接入和成员的接入各修各的。校验和发布检查的提示按位置写：有下拉框的说「在节点中换一个接入」，别处说「在发布前检查中改用其他接入」；剩下的只有接入问题时交给助手的说明也改成指向这些修复。 复核发现一并修：核查节点自己的开关关着、结构校验不许重试时不再查（执行时根本不取模型，以前会挡住发布）；报告没写裁判配置时查全图默认里的那一份（以前放行，正式运行却全部未裁判）；接入名不是字符串时查助手模型的接口不再 500，设置接口对门控、助手两组的这类值回 422 带字段；助手、表格导入、数据目录起草用的模型点名的接入不存在或已停用时照实说是哪个接入、去哪里改（以前说「尚未配置完成」，停用接入的下一步丢了），删掉两处已走不到的分支。 测试补齐：结构校验让模型修复、知识检索让模型重排、按需裁判各路径的不存在与已停用，助手、定时正式运行、裁判的两种情况，反方审稿节点上点名的接入；修复的候选、应用、拒收和图级默认。
549 lines
24 KiB
Python
549 lines
24 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.errors import graph_error
|
||
from app.api.runs import actor_of
|
||
from app.core.artifact_store import graph_hash
|
||
from app.data.catalog_impact import catalog_drift, catalog_versions_for, recorded_versions
|
||
from app.db.base import get_session
|
||
from app.db.models import Run, Workflow, WorkflowVersion
|
||
from app.engine.governance import publish_issues
|
||
from app.engine.provider_refs import CODES as PROVIDER_CODES, load_issues, load_providers
|
||
from app.engine.schema import GraphSpec, NodeType, validate_graph
|
||
|
||
router = APIRouter(prefix="/api/workflows", tags=["workflows"])
|
||
|
||
_GONE = "工作流不存在，可能已被删除"
|
||
_LEVELS = ("published", "governed")
|
||
_BAD_LEVEL = "发布级别只能是「已发布」或「受管」"
|
||
#: 剩下的挡住发布的只有点名接入的问题、人又要交给助手时的说明：换成哪个接入要人拿主意，每一处都有自己的修复
|
||
PROVIDER_LEFT = ("剩下的问题是指定的模型接入不存在或已停用，需要你决定改用哪个接入：请在每一处的修复中选择，"
|
||
"或在「设置 → 模型接入」中核对")
|
||
|
||
|
||
class WorkflowIn(BaseModel):
|
||
name: str = Field(min_length=1, max_length=200)
|
||
description: str = ""
|
||
graph: dict[str, Any] = Field(default_factory=dict)
|
||
tags: list[str] = Field(default_factory=list)
|
||
|
||
|
||
class WorkflowPatch(BaseModel):
|
||
name: str | None = None
|
||
description: str | None = None
|
||
graph: dict[str, Any] | None = None
|
||
tags: list[str] | None = None
|
||
note: str = ""
|
||
|
||
|
||
class WorkflowOut(BaseModel):
|
||
id: str
|
||
name: str
|
||
description: str
|
||
graph: dict[str, Any]
|
||
tags: list[str]
|
||
version: int
|
||
is_template: bool
|
||
status: str = "draft"
|
||
published_version: int | None = None
|
||
published_by: str | None = None
|
||
created_at: Any = None
|
||
updated_at: Any = None
|
||
run_count: int = 0
|
||
|
||
model_config = {"from_attributes": True}
|
||
|
||
|
||
@router.get("", response_model=list[WorkflowOut])
|
||
async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[WorkflowOut]:
|
||
rows = list(
|
||
(await session.execute(select(Workflow).order_by(Workflow.updated_at.desc()))).scalars()
|
||
)
|
||
counts = dict(
|
||
(await session.execute(select(Run.workflow_id, func.count(Run.id)).group_by(Run.workflow_id))).all()
|
||
)
|
||
out: list[WorkflowOut] = []
|
||
for row in rows:
|
||
item = WorkflowOut.model_validate(row)
|
||
item.run_count = counts.get(row.id, 0)
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
@router.post("", response_model=WorkflowOut, status_code=201)
|
||
async def create_workflow(
|
||
payload: WorkflowIn, session: AsyncSession = Depends(get_session)
|
||
) -> Workflow:
|
||
workflow = Workflow(
|
||
name=payload.name,
|
||
description=payload.description,
|
||
graph=payload.graph,
|
||
tags=payload.tags,
|
||
)
|
||
session.add(workflow)
|
||
await session.flush()
|
||
session.add(
|
||
WorkflowVersion(workflow_id=workflow.id, version=1, graph=payload.graph,
|
||
graph_hash=graph_hash(payload.graph), note="初始版本")
|
||
)
|
||
await session.commit()
|
||
await session.refresh(workflow)
|
||
return workflow
|
||
|
||
|
||
@router.get("/{workflow_id}", response_model=WorkflowOut)
|
||
async def get_workflow(
|
||
workflow_id: str, session: AsyncSession = Depends(get_session)
|
||
) -> Workflow:
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
return workflow
|
||
|
||
|
||
@router.patch("/{workflow_id}", response_model=WorkflowOut)
|
||
async def update_workflow(
|
||
workflow_id: str, payload: WorkflowPatch, session: AsyncSession = Depends(get_session)
|
||
) -> Workflow:
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
|
||
if payload.name is not None:
|
||
workflow.name = payload.name
|
||
if payload.description is not None:
|
||
workflow.description = payload.description
|
||
if payload.tags is not None:
|
||
workflow.tags = payload.tags
|
||
if payload.graph is not None and payload.graph != workflow.graph:
|
||
# 图有实际变化才留快照，避免改个名字就堆一堆版本
|
||
workflow.graph = payload.graph
|
||
workflow.version += 1
|
||
session.add(
|
||
WorkflowVersion(
|
||
workflow_id=workflow.id,
|
||
version=workflow.version,
|
||
graph=payload.graph,
|
||
graph_hash=graph_hash(payload.graph),
|
||
note=payload.note,
|
||
)
|
||
)
|
||
# status 说的是**当前画布**的状态，不是"这个工作流曾经发布过"。
|
||
# 改完图还挂着 governed，界面上就会显示一张从未过闸的图是受管模板——
|
||
# 已发布的那一版仍由 published_version 指着，formal 运行不受影响。
|
||
if workflow.status in ("published", "governed"):
|
||
workflow.status = "draft"
|
||
await session.commit()
|
||
await session.refresh(workflow)
|
||
return workflow
|
||
|
||
|
||
@router.delete("/{workflow_id}", status_code=204)
|
||
async def delete_workflow(
|
||
workflow_id: str, session: AsyncSession = Depends(get_session)
|
||
) -> None:
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
# 运行记录是 ON DELETE SET NULL：删掉工作流，已结束的运行留着、照样能核对封存；
|
||
# 还在跑的那次会失去归属，再也回不到它的画布，所以和删除运行记录一样先停再删
|
||
live = await session.scalar(
|
||
select(func.count()).select_from(Run).where(
|
||
Run.workflow_id == workflow_id, Run.status.in_(("queued", "running"))
|
||
)
|
||
)
|
||
if live:
|
||
raise HTTPException(409, f"该工作流还有 {live} 次运行正在进行，现在删除会使这些运行失去所属工作流。请先停止运行，再删除")
|
||
await session.delete(workflow)
|
||
await session.commit()
|
||
|
||
|
||
@router.post("/{workflow_id}/duplicate", response_model=WorkflowOut, status_code=201)
|
||
async def duplicate_workflow(
|
||
workflow_id: str, session: AsyncSession = Depends(get_session)
|
||
) -> Workflow:
|
||
source = await session.get(Workflow, workflow_id)
|
||
if not source:
|
||
raise HTTPException(404, _GONE)
|
||
copy = Workflow(
|
||
name=f"{source.name} 副本",
|
||
description=source.description,
|
||
graph=source.graph,
|
||
tags=list(source.tags or []),
|
||
)
|
||
session.add(copy)
|
||
await session.commit()
|
||
await session.refresh(copy)
|
||
return copy
|
||
|
||
|
||
class VersionOut(BaseModel):
|
||
id: str
|
||
version: int
|
||
note: str
|
||
created_at: Any = None
|
||
#: 这一版按哪一档发布的（published / governed）。没发布过、或者是记级别之前发布的为 null
|
||
level: str | None = None
|
||
|
||
model_config = {"from_attributes": True}
|
||
|
||
|
||
@router.get("/{workflow_id}/versions", response_model=list[VersionOut])
|
||
async def list_versions(
|
||
workflow_id: str, session: AsyncSession = Depends(get_session)
|
||
) -> list[WorkflowVersion]:
|
||
rows = await session.execute(
|
||
select(WorkflowVersion)
|
||
.where(WorkflowVersion.workflow_id == workflow_id)
|
||
.order_by(WorkflowVersion.version.desc())
|
||
)
|
||
return list(rows.scalars())
|
||
|
||
|
||
class VersionDetailOut(VersionOut):
|
||
workflow_id: str
|
||
graph: dict[str, Any]
|
||
graph_hash: str | None = None
|
||
#: 这一版入口节点声明的字段。正式运行跑的是已发布的那一版，表单得照它来填——
|
||
#: 画布上改过入口字段时，拿当前画布的字段去跑已发布版本会传错参数
|
||
input_fields: list[dict[str, Any]] = Field(default_factory=list)
|
||
published: bool = False
|
||
#: 发布时记下的数据目录版本 {"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}：direct 写死的
|
||
#: SQL 用到的表，possible Agent 可能查询的表（catalog_impact.recorded_versions，只记了直接引用的旧记录 possible
|
||
#: 为空）；没发布过、或者记录之前发布的为 null
|
||
catalog_versions: dict[str, Any] | None = None
|
||
#: 发布之后目录有变化的表：[{source, source_id, table, label, published, current, impact}]
|
||
catalog_changes: list[dict[str, Any]] = Field(default_factory=list)
|
||
|
||
|
||
@router.get("/{workflow_id}/versions/{version}", response_model=VersionDetailOut)
|
||
async def get_version(
|
||
workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
|
||
) -> VersionDetailOut:
|
||
"""取某一版的完整图。"""
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
snapshot = (
|
||
await session.execute(
|
||
select(WorkflowVersion).where(
|
||
WorkflowVersion.workflow_id == workflow_id,
|
||
WorkflowVersion.version == version,
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not snapshot:
|
||
raise HTTPException(404, f"「{workflow.name}」不存在 v{version} 版本")
|
||
|
||
graph = snapshot.graph or {}
|
||
fields: list[dict[str, Any]] = []
|
||
for node in graph.get("nodes") or []:
|
||
if node.get("type") == "input":
|
||
fields.extend(
|
||
f for f in ((node.get("data") or {}).get("config") or {}).get("fields") or []
|
||
if isinstance(f, dict)
|
||
)
|
||
return VersionDetailOut(
|
||
id=snapshot.id, version=snapshot.version, note=snapshot.note,
|
||
created_at=snapshot.created_at, workflow_id=workflow_id, graph=graph,
|
||
graph_hash=snapshot.graph_hash or graph_hash(graph), input_fields=fields,
|
||
published=workflow.published_version == snapshot.version, level=snapshot.level,
|
||
catalog_versions=recorded_versions(snapshot.catalog_versions),
|
||
catalog_changes=await catalog_drift(session, snapshot.catalog_versions),
|
||
)
|
||
|
||
|
||
@router.post("/{workflow_id}/versions/{version}/restore", response_model=WorkflowOut)
|
||
async def restore_version(
|
||
workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
|
||
) -> Workflow:
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
snapshot = (
|
||
await session.execute(
|
||
select(WorkflowVersion).where(
|
||
WorkflowVersion.workflow_id == workflow_id,
|
||
WorkflowVersion.version == version,
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not snapshot:
|
||
raise HTTPException(404, f"不存在 v{version} 版本")
|
||
|
||
workflow.graph = snapshot.graph
|
||
workflow.version += 1
|
||
session.add(
|
||
WorkflowVersion(
|
||
workflow_id=workflow.id,
|
||
version=workflow.version,
|
||
graph=snapshot.graph,
|
||
graph_hash=graph_hash(snapshot.graph),
|
||
note=f"回滚到 v{version}",
|
||
)
|
||
)
|
||
# 恢复旧版也是改图，和 PATCH 同一条规矩：旧图没过这次的闸，不能挂着受管 / 已发布
|
||
if workflow.status in ("published", "governed"):
|
||
workflow.status = "draft"
|
||
await session.commit()
|
||
await session.refresh(workflow)
|
||
return workflow
|
||
|
||
|
||
class PublishIn(BaseModel):
|
||
level: str = "published" # published | governed
|
||
version: int | None = None # 默认发布当前最新版本
|
||
|
||
|
||
@router.post("/{workflow_id}/publish")
|
||
async def publish_workflow(
|
||
workflow_id: str,
|
||
payload: PublishIn,
|
||
session: AsyncSession = Depends(get_session),
|
||
x_actor: str | None = Header(default=None),
|
||
) -> dict[str, Any]:
|
||
"""把某个版本立为正式版本。
|
||
|
||
formal 运行只能从这里发布过的版本发起。governed 级别额外过治理 lint：
|
||
错误会挡住发布——这就是"必经检查点"落地的地方。
|
||
返回 200 + ok/issues 而不是抛错，让前端能把问题列表渲染出来。
|
||
"""
|
||
if payload.level not in _LEVELS:
|
||
raise HTTPException(400, _BAD_LEVEL)
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
|
||
version = payload.version or workflow.version
|
||
snapshot = (
|
||
await session.execute(
|
||
select(WorkflowVersion).where(
|
||
WorkflowVersion.workflow_id == workflow_id,
|
||
WorkflowVersion.version == version,
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not snapshot:
|
||
raise HTTPException(404, f"不存在 v{version} 版本：画布上的修改还没有保存，请先保存再发布")
|
||
|
||
try:
|
||
spec = GraphSpec.model_validate(snapshot.graph)
|
||
except Exception as e: # noqa: BLE001
|
||
return {"ok": False, "issues": [{"level": "error", "message": f"无法解析工作流结构：{graph_error(e)}"}]}
|
||
|
||
checkers = await _sql_checkers(session, spec)
|
||
issues = [i.model_dump() for i in publish_issues(spec, level=payload.level, checkers=checkers,
|
||
providers=await _providers(session, spec))]
|
||
if any(i["level"] == "error" for i in issues):
|
||
return {"ok": False, "level": payload.level, "version": version, "issues": issues}
|
||
|
||
if not snapshot.graph_hash:
|
||
snapshot.graph_hash = graph_hash(snapshot.graph)
|
||
# 级别记在这一版上：之后改画布（status 退回 draft），从这一版发起的正式运行照样按发布时的级别出具
|
||
snapshot.level = payload.level
|
||
# 发布时的目录版本也记在这一版上：重新发布同一版等于按现在的目录重新认可它，跟着更新
|
||
snapshot.catalog_versions = await catalog_versions_for(session, spec)
|
||
workflow.status = payload.level
|
||
workflow.published_version = version
|
||
workflow.published_by = actor_of(x_actor)
|
||
await session.commit()
|
||
return {
|
||
"ok": True,
|
||
"level": payload.level,
|
||
"version": version,
|
||
"graph_hash": snapshot.graph_hash,
|
||
"published_by": workflow.published_by,
|
||
"catalog_versions": recorded_versions(snapshot.catalog_versions),
|
||
"issues": issues, # 剩下的都是警告
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 发布前检查与自动修复：两个接口都只读，不改库、不发布
|
||
# --------------------------------------------------------------------------
|
||
|
||
class PublishCheckIn(BaseModel):
|
||
level: str = "published"
|
||
#: 画布上还没保存的图。不带就用库里当前的草稿
|
||
graph: dict[str, Any] | None = None
|
||
|
||
|
||
class AutofixIn(PublishCheckIn):
|
||
#: 要应用的修复 id（publish-check 给的 fixes[].id）
|
||
apply: list[str] = Field(default_factory=list)
|
||
#: choice 类修复的选择：{fix_id: 值}，可以多选的给列表
|
||
choices: dict[str, Any] = Field(default_factory=dict)
|
||
#: 剩下的 error 交给 Copilot 试着修（结果同样只是预览）
|
||
assist: bool = False
|
||
provider: str | None = None
|
||
model: str | None = None
|
||
|
||
|
||
async def _draft(session: AsyncSession, workflow_id: str, payload: PublishCheckIn) -> dict[str, Any]:
|
||
if payload.level not in _LEVELS:
|
||
raise HTTPException(400, _BAD_LEVEL)
|
||
workflow = await session.get(Workflow, workflow_id)
|
||
if not workflow:
|
||
raise HTTPException(404, _GONE)
|
||
return payload.graph if payload.graph is not None else (workflow.graph or {})
|
||
|
||
|
||
async def _gate_context(session: AsyncSession, spec: GraphSpec) -> tuple[dict[str, Any], dict[str, Any]]:
|
||
"""修复要用、但门禁本身不查库的东西：子工作流当前的发布版本，钉在别处的口径卡有哪些指标。"""
|
||
versions: dict[str, Any] = {}
|
||
cards: dict[str, Any] = {}
|
||
for node in spec.nodes:
|
||
cfg = node.config
|
||
if node.type == NodeType.SUBGRAPH and cfg.get("workflow_id") and str(cfg["workflow_id"]) not in versions:
|
||
upstream = await session.get(Workflow, str(cfg["workflow_id"]))
|
||
versions[str(cfg["workflow_id"])] = upstream.published_version if upstream else None
|
||
elif node.type == NodeType.METRICS and isinstance(cfg.get("caliber_from"), dict):
|
||
ref = cfg["caliber_from"]
|
||
version = ref.get("workflow_version")
|
||
if not ref.get("workflow_id") or not str(version or "").isdigit():
|
||
continue
|
||
snapshot = (await session.execute(select(WorkflowVersion).where(
|
||
WorkflowVersion.workflow_id == str(ref["workflow_id"]),
|
||
WorkflowVersion.version == int(version)))).scalar_one_or_none()
|
||
card = next((n for n in ((snapshot.graph if snapshot else None) or {}).get("nodes") or []
|
||
if isinstance(n, dict) and n.get("id") == ref.get("node_id")), None)
|
||
if card is not None:
|
||
cards[node.id] = ((card.get("data") or {}).get("config") or {}).get("metrics") or []
|
||
return versions, cards
|
||
|
||
|
||
async def _sql_checkers(session: AsyncSession, spec: GraphSpec) -> dict[str, Any]:
|
||
"""图里调用工具节点写死的 SQL 用到的数据源，各建一个基于数据目录的 SQL 检查器。发布、发布前检查、
|
||
自动修复都传这一份给门禁（governance.lint_for_publish）：门禁本身不查库，三处口径才一致。"""
|
||
from app.data.sqlcheck import graph_sources, load_checkers_by_name
|
||
|
||
return await load_checkers_by_name(session, graph_sources(spec.nodes))
|
||
|
||
|
||
async def _providers(session: AsyncSession, spec: GraphSpec) -> list[Any]:
|
||
"""模型接入表：节点点名的接入不存在或已停用时，发布、发布前检查、自动修复的复核给 error、挡住发布，修复从启用的
|
||
接入里列候选（engine/provider_refs.py、autofix 的 provider.*）。和 _sql_checkers 一样由接口这一层读好交给门禁，
|
||
门禁本身不查库；三处传同一份。图里没点名任何接入时不读库。"""
|
||
return await load_providers(session, spec)
|
||
|
||
|
||
def _unreadable(level: str, e: Exception) -> dict[str, Any]:
|
||
return {"level": level, "ok": False, "fixes": [], "issues": [
|
||
{"level": "error", "node_id": None, "edge_id": None, "field": None, "code": None, "fix": None,
|
||
"message": f"无法解析工作流结构：{graph_error(e)}"}]}
|
||
|
||
|
||
@router.post("/{workflow_id}/publish-check")
|
||
async def publish_check(
|
||
workflow_id: str, payload: PublishCheckIn, session: AsyncSession = Depends(get_session)
|
||
) -> dict[str, Any]:
|
||
"""发布前检查：和真正发布同一套口径（validate + 门禁）列出问题，以及能怎么修。
|
||
|
||
只读：不改库、不发布。发布弹窗一打开就调它，不用等点了发布才知道被拦。
|
||
"""
|
||
from app.engine.autofix import check
|
||
|
||
graph = await _draft(session, workflow_id, payload)
|
||
try:
|
||
spec = GraphSpec.model_validate(graph)
|
||
except Exception as e: # noqa: BLE001
|
||
return _unreadable(payload.level, e)
|
||
versions, cards = await _gate_context(session, spec)
|
||
return check(spec, level=payload.level, versions=versions, cards=cards, checkers=await _sql_checkers(session, spec),
|
||
providers=await _providers(session, spec))
|
||
|
||
|
||
@router.post("/{workflow_id}/autofix")
|
||
async def autofix(
|
||
workflow_id: str, payload: AutofixIn, session: AsyncSession = Depends(get_session)
|
||
) -> dict[str, Any]:
|
||
"""应用选中的修复，返回修复后的整张图——只是预览。
|
||
|
||
不自动保存、不自动发布：人看过逐项变更再确认，前端走现有的保存接口存草稿，再检查、再发布。
|
||
每条修复都复核过（降低要求的、错误没变少或者冒出新错误的丢弃并写明原因）。assist 为真时，
|
||
剩下的 error 交给 Copilot，同一道复核通过才采纳。
|
||
"""
|
||
from app.engine.autofix import apply_fixes, diff_changes
|
||
|
||
graph = await _draft(session, workflow_id, payload)
|
||
try:
|
||
spec = GraphSpec.model_validate(graph)
|
||
except Exception as e: # noqa: BLE001
|
||
bad = _unreadable(payload.level, e)
|
||
return {"graph": graph, "changes": [], "ops": [], "applied": [], "rejected": [
|
||
{"fix_id": fid, "reason": "无法解析工作流结构，不能应用修复"} for fid in payload.apply],
|
||
"remaining": bad["issues"], "fixes": [], "handoff": [], "assist": None, "ok": False}
|
||
versions, cards = await _gate_context(session, spec)
|
||
checkers = await _sql_checkers(session, spec)
|
||
providers = await _providers(session, spec)
|
||
out = apply_fixes(graph, payload.apply, payload.choices, level=payload.level, versions=versions, cards=cards,
|
||
checkers=checkers, providers=providers)
|
||
out["assist"] = None
|
||
# 人在选项里选了「交给 Copilot」（handoff），和点「交给 Copilot」是一回事
|
||
if payload.assist or out["handoff"]:
|
||
if out["ok"]:
|
||
out["assist"] = {"ok": True, "summary": "修复后已没有阻止发布的错误，未再交给助手", "questions": []}
|
||
return out
|
||
if all(i.get("code") in PROVIDER_CODES for i in out["remaining"] if i["level"] == "error"):
|
||
# 剩下的只有点名接入的问题：换成哪个接入要人拿主意（助手的自查也不看接入表），不交给助手
|
||
out["assist"] = {"ok": False, "summary": PROVIDER_LEFT, "questions": []}
|
||
return out
|
||
from app.api.copilot import assist_publish_fix
|
||
|
||
helped = await assist_publish_fix(session, out["graph"], level=payload.level,
|
||
provider=payload.provider, model=payload.model)
|
||
out["assist"] = {"ok": helped["accepted"], "summary": helped["summary"], "questions": helped["questions"]}
|
||
if helped["accepted"]:
|
||
fixed = helped["graph"]
|
||
out["changes"] += diff_changes(out["graph"], fixed, fix_id="assist", label="助手的修改")
|
||
out["ops"] += helped["ops"]
|
||
out["applied"].append("assist")
|
||
fixed_spec = GraphSpec.model_validate(fixed)
|
||
after = apply_fixes(fixed, [], level=payload.level, versions=versions, cards=cards,
|
||
checkers=await _sql_checkers(session, fixed_spec),
|
||
providers=await _providers(session, fixed_spec))
|
||
out.update(graph=fixed, remaining=after["remaining"], fixes=after["fixes"], ok=after["ok"])
|
||
elif helped["reason"]:
|
||
out["rejected"].append({"fix_id": "assist", "reason": helped["reason"]})
|
||
return out
|
||
|
||
|
||
class ValidateIn(BaseModel):
|
||
graph: dict[str, Any]
|
||
|
||
|
||
@router.post("/validate")
|
||
async def validate(payload: ValidateIn, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||
"""画布上的实时校验。前端每次改动都会调它来标红。
|
||
|
||
点名的模型接入不存在或已停用只给警告：探索运行仍可发起，执行到那个节点时照实报错；发布时才挡住。"""
|
||
try:
|
||
spec = GraphSpec.model_validate(payload.graph)
|
||
except Exception as e: # noqa: BLE001
|
||
return {"ok": False, "issues": [{"level": "error", "message": f"无法解析工作流结构：{graph_error(e)}"}]}
|
||
result = validate_graph(spec).model_dump()
|
||
result["issues"] += [i.model_dump() for i in await load_issues(session, spec, level="warning")]
|
||
return result
|
||
|
||
|
||
@router.post("/variables")
|
||
async def variables(payload: ValidateIn) -> dict[str, Any]:
|
||
"""这张图里有哪些变量、谁产出、谁引用。
|
||
|
||
纯静态分析，不用跑图也不用有运行记录——用户在编排到一半时最需要它，
|
||
而那时候还没有任何一次运行可看。运行期的实际取值由前端从事件里补上。
|
||
"""
|
||
from app.engine.variables import analyze
|
||
|
||
try:
|
||
spec = GraphSpec.model_validate(payload.graph)
|
||
except Exception as e: # noqa: BLE001
|
||
return {"variables": [], "issues": [{"level": "error", "message": f"无法解析工作流结构：{graph_error(e)}"}]}
|
||
return analyze(spec).model_dump()
|