feat: initial release - Hunzi agent framework v0.1.0

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,
+        }