董宏凯
1706a24182
提交范围聚焦于工作流编排新增：新增 workflow_planning 路由与前后端运行入口（workflow_planning.py、workflow_proposal_planner.py、workflow_deep_research_adapter.py），支持基于会话的提案-规划-执行流程，满足任务驱动式多智能体协作。 完成网关层运行时适配：扩展 workflow_resource_checks、workflow_executor、workflow_runs、workflow_agent_runner 与依赖注入，新增/更新配置（config.yaml、workflow_config），并补充 lead_agent 路径与节点实现，确保 Planner 与普通工作流运行态衔接一致。 补齐持久化与建模：新增 workflow_planning 持久化模块与 MySQL migration（20260831_01_workflow_planning_sessions），注册 SQL 与模型、内存 store，并扩展错误/表达式/validator/节点语义以承载新会话态输入输出。 增强文档与测试：新增会话式多智能体编排中文说明、更新开发文档与 AGENTS 指南、补充 workflow_engine/workflow_runs 回归测试，提高新路径的行为可验证性；同时留存 seed 资产（workflow-planner）便于一体化部署。
381 lines
12 KiB
Python
381 lines
12 KiB
Python
"""Frozen workflow graph schema v1.0 (internal DTO).
|
|
|
|
Coze canvas DTO is converted into this structure at the Gateway boundary.
|
|
The runtime must never depend on Coze-private fields.
|
|
|
|
Wire JSON uses camelCase aliases; Python code uses snake_case.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from hashlib import sha256
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
SCHEMA_VERSION = "1.0"
|
|
|
|
NodeType = Literal[
|
|
"start",
|
|
"output",
|
|
"agent",
|
|
"skill",
|
|
"http",
|
|
"sql_read",
|
|
"code",
|
|
"transform",
|
|
"evidence_normalizer",
|
|
"deep_research_write",
|
|
"condition",
|
|
"merge",
|
|
"human_input",
|
|
"subworkflow",
|
|
"loop",
|
|
]
|
|
|
|
ALL_NODE_TYPES: tuple[NodeType, ...] = (
|
|
"start",
|
|
"output",
|
|
"agent",
|
|
"skill",
|
|
"http",
|
|
"sql_read",
|
|
"code",
|
|
"transform",
|
|
"evidence_normalizer",
|
|
"deep_research_write",
|
|
"condition",
|
|
"merge",
|
|
"human_input",
|
|
"subworkflow",
|
|
"loop",
|
|
)
|
|
|
|
RunStatus = Literal[
|
|
"queued",
|
|
"running",
|
|
"awaiting_input",
|
|
"completed",
|
|
"failed",
|
|
"cancel_requested",
|
|
"cancelled",
|
|
]
|
|
|
|
ALL_RUN_STATUSES: tuple[RunStatus, ...] = (
|
|
"queued",
|
|
"running",
|
|
"awaiting_input",
|
|
"completed",
|
|
"failed",
|
|
"cancel_requested",
|
|
"cancelled",
|
|
)
|
|
|
|
TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset({"completed", "failed", "cancelled"})
|
|
|
|
NodeRunStatus = Literal[
|
|
"queued",
|
|
"running",
|
|
"completed",
|
|
"failed",
|
|
"skipped",
|
|
"cancelled",
|
|
]
|
|
|
|
SkillMode = Literal["agent_skill", "callable_skill"]
|
|
MergeStrategy = Literal["all", "first_success", "any_success"]
|
|
HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE"]
|
|
AgentResponseMode = Literal["text", "json"]
|
|
AgentThreadMode = Literal["isolated_per_node", "shared_run_thread"]
|
|
ExecutionMode = Literal["normal"]
|
|
|
|
|
|
class WorkflowGraphSettings(BaseModel):
|
|
"""Per-graph resource limits (still capped by system limits at publish)."""
|
|
|
|
run_timeout_seconds: int = Field(default=1800, ge=1, alias="runTimeoutSeconds")
|
|
node_timeout_seconds: int = Field(default=300, ge=1, alias="nodeTimeoutSeconds")
|
|
max_steps: int = Field(default=200, ge=1, alias="maxSteps")
|
|
max_loop_iterations: int = Field(default=5, ge=1, alias="maxLoopIterations")
|
|
max_parallelism: int = Field(default=4, ge=1, alias="maxParallelism")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class WorkflowNode(BaseModel):
|
|
"""A single node on the internal standard graph."""
|
|
|
|
id: str = Field(min_length=1)
|
|
type: NodeType
|
|
name: str = ""
|
|
config: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
@field_validator("id")
|
|
@classmethod
|
|
def _id_not_blank(cls, value: str) -> str:
|
|
stripped = value.strip()
|
|
if not stripped:
|
|
raise ValueError("node id must be non-empty")
|
|
return stripped
|
|
|
|
|
|
class WorkflowEdge(BaseModel):
|
|
"""Directed edge. ``source_port`` names a condition / loop outlet when set."""
|
|
|
|
id: str = Field(min_length=1)
|
|
source: str
|
|
target: str
|
|
source_port: str | None = Field(default=None, alias="sourcePort")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class WorkflowGraph(BaseModel):
|
|
"""Internal standard workflow graph (publish / run snapshot)."""
|
|
|
|
schema_version: Literal["1.0"] = Field(default=SCHEMA_VERSION, alias="schemaVersion")
|
|
id: str = Field(min_length=1)
|
|
name: str = ""
|
|
description: str = ""
|
|
input_schema: dict[str, Any] = Field(default_factory=dict, alias="inputSchema")
|
|
output_schema: dict[str, Any] = Field(default_factory=dict, alias="outputSchema")
|
|
nodes: list[WorkflowNode] = Field(default_factory=list)
|
|
edges: list[WorkflowEdge] = Field(default_factory=list)
|
|
settings: WorkflowGraphSettings = Field(default_factory=WorkflowGraphSettings)
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
@model_validator(mode="after")
|
|
def _unique_ids(self) -> WorkflowGraph:
|
|
node_ids = [n.id for n in self.nodes]
|
|
if len(node_ids) != len(set(node_ids)):
|
|
raise ValueError("duplicate node id")
|
|
edge_ids = [e.id for e in self.edges]
|
|
if len(edge_ids) != len(set(edge_ids)):
|
|
raise ValueError("duplicate edge id")
|
|
known = set(node_ids)
|
|
for edge in self.edges:
|
|
if edge.source not in known:
|
|
raise ValueError(f"edge {edge.id} source not found: {edge.source}")
|
|
if edge.target not in known:
|
|
raise ValueError(f"edge {edge.id} target not found: {edge.target}")
|
|
return self
|
|
|
|
def node_map(self) -> dict[str, WorkflowNode]:
|
|
return {node.id: node for node in self.nodes}
|
|
|
|
def canonical_json(self) -> str:
|
|
"""Deterministic JSON used for ``graph_hash`` (sorted keys, no whitespace)."""
|
|
return self.model_dump_json(by_alias=True, exclude_none=True)
|
|
|
|
def graph_hash(self) -> str:
|
|
return sha256(self.canonical_json().encode("utf-8")).hexdigest()
|
|
|
|
|
|
class ArtifactRef(BaseModel):
|
|
"""Reference to a large/binary result; full bytes live outside events."""
|
|
|
|
artifact_id: str = Field(alias="artifactId")
|
|
name: str
|
|
mime_type: str = Field(default="application/octet-stream", alias="mimeType")
|
|
path: str = ""
|
|
size_bytes: int | None = Field(default=None, alias="sizeBytes")
|
|
preview: str | None = None
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class NodeResult(BaseModel):
|
|
"""Uniform result returned by every node executor."""
|
|
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
messages: list[dict[str, Any]] = Field(default_factory=list)
|
|
artifacts: list[ArtifactRef] = Field(default_factory=list)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
warnings: list[str] = Field(default_factory=list)
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class StartRunRequest(BaseModel):
|
|
"""Body for ``POST /api/workflows/{workflow_id}/runs``."""
|
|
|
|
version_id: str = Field(alias="versionId")
|
|
inputs: dict[str, Any] = Field(default_factory=dict)
|
|
idempotency_key: str | None = Field(default=None, alias="idempotencyKey")
|
|
execution_mode: ExecutionMode = Field(default="normal", alias="executionMode")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class ResumeRunRequest(BaseModel):
|
|
"""Body for ``POST /api/workflows/runs/{run_id}/resume``."""
|
|
|
|
resume_token: str = Field(alias="resumeToken")
|
|
action: str
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
# ── Typed node config helpers (optional; validators may use later) ───────────
|
|
|
|
|
|
class AgentNodeConfig(BaseModel):
|
|
agent_id: str = Field(alias="agentId")
|
|
prompt_template: str = Field(default="", alias="promptTemplate")
|
|
input_bindings: dict[str, Any] = Field(default_factory=dict, alias="inputBindings")
|
|
response_mode: AgentResponseMode = Field(default="text", alias="responseMode")
|
|
response_schema: dict[str, Any] = Field(default_factory=dict, alias="responseSchema")
|
|
thread_mode: AgentThreadMode = Field(default="isolated_per_node", alias="threadMode")
|
|
timeout_seconds: int = Field(default=600, ge=1, alias="timeoutSeconds")
|
|
skill_names: list[str] = Field(default_factory=list, alias="skillNames")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class SkillNodeConfig(BaseModel):
|
|
mode: SkillMode = "agent_skill"
|
|
agent_id: str | None = Field(default=None, alias="agentId")
|
|
skill_names: list[str] = Field(default_factory=list, alias="skillNames")
|
|
prompt_template: str = Field(default="", alias="promptTemplate")
|
|
input_bindings: dict[str, Any] = Field(default_factory=dict, alias="inputBindings")
|
|
# callable_skill fields (phase 4+; accepted in schema, not executed in phase 0)
|
|
entry: str | None = None
|
|
input_schema: dict[str, Any] = Field(default_factory=dict, alias="inputSchema")
|
|
output_schema: dict[str, Any] = Field(default_factory=dict, alias="outputSchema")
|
|
timeout_seconds: int = Field(default=300, ge=1, alias="timeoutSeconds")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class HttpNodeConfig(BaseModel):
|
|
method: HttpMethod = "GET"
|
|
url: str
|
|
headers: dict[str, str] = Field(default_factory=dict)
|
|
query: dict[str, Any] = Field(default_factory=dict)
|
|
body: Any = None
|
|
credential_ref: str | None = Field(default=None, alias="credentialRef")
|
|
timeout_seconds: int = Field(default=60, ge=1, alias="timeoutSeconds")
|
|
max_response_bytes: int = Field(default=1_048_576, ge=1, alias="maxResponseBytes")
|
|
allow_redirects: bool = Field(default=False, alias="allowRedirects")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class SqlReadNodeConfig(BaseModel):
|
|
data_source_id: str = Field(alias="dataSourceId")
|
|
statement: str
|
|
parameters: dict[str, Any] = Field(default_factory=dict)
|
|
max_rows: int = Field(default=1000, ge=1, alias="maxRows")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class CodeNodeConfig(BaseModel):
|
|
language: Literal["python"] = "python"
|
|
source: str
|
|
timeout_seconds: int = Field(default=60, ge=1, alias="timeoutSeconds")
|
|
allow_network: bool = Field(default=False, alias="allowNetwork")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class TransformNodeConfig(BaseModel):
|
|
operations: list[dict[str, Any]] = Field(default_factory=list)
|
|
template: str | None = None
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class ConditionBranch(BaseModel):
|
|
name: str
|
|
expression: str
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class ConditionNodeConfig(BaseModel):
|
|
branches: list[ConditionBranch] = Field(default_factory=list)
|
|
default_branch: str | None = Field(default=None, alias="defaultBranch")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class MergeNodeConfig(BaseModel):
|
|
strategy: MergeStrategy = "all"
|
|
sources: list[str] = Field(default_factory=list)
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class HumanInputNodeConfig(BaseModel):
|
|
form_schema: dict[str, Any] = Field(default_factory=dict, alias="formSchema")
|
|
prompt: str = ""
|
|
actions: list[str] = Field(default_factory=lambda: ["submit"])
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class SubworkflowNodeConfig(BaseModel):
|
|
workflow_id: str = Field(alias="workflowId")
|
|
version_id: str = Field(alias="versionId")
|
|
input_mapping: dict[str, str] = Field(default_factory=dict, alias="inputMapping")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class LoopNodeConfig(BaseModel):
|
|
max_iterations: int = Field(default=5, ge=1, alias="maxIterations")
|
|
continue_when: str = Field(default="", alias="continueWhen")
|
|
body_entry: str = Field(alias="bodyEntry")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
class OutputNodeConfig(BaseModel):
|
|
mapping: dict[str, str] = Field(default_factory=dict)
|
|
llm_repair: bool = Field(default=False, alias="llmRepair")
|
|
|
|
model_config = {"extra": "forbid", "populate_by_name": True}
|
|
|
|
|
|
__all__ = [
|
|
"SCHEMA_VERSION",
|
|
"ALL_NODE_TYPES",
|
|
"ALL_RUN_STATUSES",
|
|
"TERMINAL_RUN_STATUSES",
|
|
"AgentNodeConfig",
|
|
"AgentResponseMode",
|
|
"AgentThreadMode",
|
|
"ArtifactRef",
|
|
"CodeNodeConfig",
|
|
"ConditionBranch",
|
|
"ConditionNodeConfig",
|
|
"ExecutionMode",
|
|
"HttpMethod",
|
|
"HttpNodeConfig",
|
|
"HumanInputNodeConfig",
|
|
"LoopNodeConfig",
|
|
"MergeNodeConfig",
|
|
"MergeStrategy",
|
|
"NodeResult",
|
|
"NodeRunStatus",
|
|
"NodeType",
|
|
"OutputNodeConfig",
|
|
"ResumeRunRequest",
|
|
"RunStatus",
|
|
"SkillMode",
|
|
"SkillNodeConfig",
|
|
"SqlReadNodeConfig",
|
|
"StartRunRequest",
|
|
"SubworkflowNodeConfig",
|
|
"TransformNodeConfig",
|
|
"WorkflowEdge",
|
|
"WorkflowGraph",
|
|
"WorkflowGraphSettings",
|
|
"WorkflowNode",
|
|
]
|