From 1d7338194767baa769768d693a3fa669bdf021ed Mon Sep 17 00:00:00 2001 From: huangzhuang_3rd Date: Wed, 15 Jul 2026 16:37:04 +0800 Subject: [PATCH] feat: initial release - Hunzi agent framework v0.1.0 --- app/workflow/models.py | 168 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 app/workflow/models.py diff --git a/app/workflow/models.py b/app/workflow/models.py new file mode 100644 index 0000000..a0c7661 --- /dev/null +++ b/app/workflow/models.py @@ -0,0 +1,168 @@ +"""Workflow definition models — dataclasses for describing DAG-based workflows.""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from typing import Any, Dict, List, Optional + + +# --------------------------------------------------------------------------- +# Enums +# --------------------------------------------------------------------------- + +class StepStatus(str, Enum): + """Execution status of a single workflow step.""" + + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + SKIPPED = "skipped" + + +class WorkflowStatus(str, Enum): + """Overall workflow execution status.""" + + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + + +# --------------------------------------------------------------------------- +# Step Definition +# --------------------------------------------------------------------------- + +@dataclass +class StepDef: + """工作流步骤定义""" + + name: str + agent_id: str # 使用哪个 Agent + prompt: str # 给 Agent 的提示 + inputs: List[str] = field(default_factory=list) # 依赖的上游步骤名 + condition: Optional[str] = None # 条件表达式(可选),如 "step_a.output == 'yes'" + parallel: bool = False # 是否可并行(与同层级无依赖的其他步骤一起执行) + + +# --------------------------------------------------------------------------- +# Workflow Definition +# --------------------------------------------------------------------------- + +@dataclass +class Workflow: + """工作流定义""" + + id: str = field(default_factory=lambda: str(uuid.uuid4())) + name: str = "" + description: str = "" + steps: List[StepDef] = field(default_factory=list) + created_at: datetime = field(default_factory=datetime.utcnow) + + # -- serialization -------------------------------------------------------- + + def to_dict(self) -> Dict[str, Any]: + """Serialize workflow to a plain dict.""" + return { + "id": self.id, + "name": self.name, + "description": self.description, + "steps": [ + { + "name": step.name, + "agent_id": step.agent_id, + "prompt": step.prompt, + "inputs": step.inputs, + "condition": step.condition, + "parallel": step.parallel, + } + for step in self.steps + ], + "created_at": self.created_at.isoformat(), + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Workflow": + """Deserialize workflow from a plain dict.""" + return cls( + id=data.get("id", str(uuid.uuid4())), + name=data.get("name", ""), + description=data.get("description", ""), + steps=[ + StepDef( + name=s["name"], + agent_id=s["agent_id"], + prompt=s["prompt"], + inputs=s.get("inputs", []), + condition=s.get("condition"), + parallel=s.get("parallel", False), + ) + for s in data.get("steps", []) + ], + created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else datetime.utcnow(), + ) + + +# --------------------------------------------------------------------------- +# Execution Results +# --------------------------------------------------------------------------- + +@dataclass +class StepResult: + """Result of executing a single step.""" + + name: str + status: StepStatus = StepStatus.PENDING + output: Any = None + error: Optional[str] = None + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + + +@dataclass +class WorkflowResult: + """Complete result of a workflow execution.""" + + workflow_id: str + workflow_name: str + status: WorkflowStatus = WorkflowStatus.PENDING + step_results: Dict[str, StepResult] = field(default_factory=dict) + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + error: Optional[str] = None + + # -- helpers -------------------------------------------------------------- + + @property + def all_completed(self) -> bool: + return all( + r.status in (StepStatus.COMPLETED, StepStatus.SKIPPED) + for r in self.step_results.values() + ) + + @property + def any_failed(self) -> bool: + return any(r.status == StepStatus.FAILED for r in self.step_results.values()) + + def to_dict(self) -> Dict[str, Any]: + return { + "workflow_id": self.workflow_id, + "workflow_name": self.workflow_name, + "status": self.status.value, + "step_results": { + name: { + "status": r.status.value, + "output": r.output, + "error": r.error, + "started_at": r.started_at.isoformat() if r.started_at else None, + "completed_at": r.completed_at.isoformat() if r.completed_at else None, + } + for name, r in self.step_results.items() + }, + "started_at": self.started_at.isoformat() if self.started_at else None, + "completed_at": self.completed_at.isoformat() if self.completed_at else None, + "error": self.error, + }