feat: workflows v2

python/beeai_framework/workflows/v2/decorators/_and.py

@@ -0,0 +1,16 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.workflows.v2.types import AsyncMethod, AsyncMethodSet
+
+
+def _and(*methods: AsyncMethod | str) -> AsyncMethodSet:
+    asm = AsyncMethodSet()
+
+    for method in methods:
+        if isinstance(method, str):
+            asm.methods.append(method)
+        elif callable(method):
+            asm.methods.append(method.__name__)
+
+    return asm

python/beeai_framework/workflows/v2/decorators/_or.py

@@ -0,0 +1,17 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.workflows.v2.types import AsyncMethod, AsyncMethodSet
+
+
+def _or(*methods: AsyncMethod | str) -> AsyncMethodSet:
+    asm = AsyncMethodSet()
+
+    for method in methods:
+        if isinstance(method, str):
+            asm.methods.append(method)
+        elif callable(method):
+            asm.methods.append(method.__name__)
+
+    asm.condition = "or"
+    return asm

python/beeai_framework/workflows/v2/decorators/after.py

@@ -0,0 +1,15 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Callable
+
+from beeai_framework.workflows.v2.types import AsyncMethod, AsyncMethodSet
+
+
+def after(dependency: str | AsyncMethod | AsyncMethodSet) -> Callable[[AsyncMethod], AsyncMethod]:
+    def decorator(func: AsyncMethod) -> AsyncMethod:
+        func._is_step = True  # type: ignore[attr-defined]
+        func._dependency = dependency  # type: ignore[attr-defined]
+        return func
+
+    return decorator

python/beeai_framework/workflows/v2/decorators/end.py

@@ -0,0 +1,20 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+
+from collections.abc import Callable, Coroutine
+from typing import Any, Concatenate, ParamSpec, TypeVar
+
+from beeai_framework.backend.message import AnyMessage
+from beeai_framework.workflows.v2.workflow import Workflow
+
+S = TypeVar("S", bound=Workflow)
+P = ParamSpec("P")
+
+EndMethod = Callable[Concatenate[S, P], Coroutine[Any, Any, list[AnyMessage]]]
+
+
+def end(func: EndMethod[S, P]) -> EndMethod[S, P]:
+    func._is_step = True  # type: ignore[attr-defined]
+    func._is_end = True  # type: ignore[attr-defined]
+    return func

python/beeai_framework/workflows/v2/decorators/fork.py

@@ -0,0 +1,9 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.workflows.v2.types import AsyncMethod
+
+
+def fork(func: AsyncMethod) -> AsyncMethod:
+    func._is_fork = True  # type: ignore[attr-defined]
+    return func

python/beeai_framework/workflows/v2/decorators/join.py

@@ -0,0 +1,9 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.workflows.v2.types import AsyncMethod
+
+
+def join(func: AsyncMethod) -> AsyncMethod:
+    func._is_join = True  # type: ignore[attr-defined]
+    return func

python/beeai_framework/workflows/v2/decorators/retry.py

@@ -0,0 +1,15 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+
+from collections.abc import Callable
+
+from beeai_framework.workflows.v2.types import AsyncMethod
+
+
+def retry(n: int = 1) -> Callable[[AsyncMethod], AsyncMethod]:
+    def decorator(func: AsyncMethod) -> AsyncMethod:
+        func._retries = n  # type: ignore[attr-defined]
+        return func
+
+    return decorator

python/beeai_framework/workflows/v2/decorators/start.py

@@ -0,0 +1,19 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+
+from collections.abc import Callable, Coroutine
+from typing import Any, TypeVar
+
+from beeai_framework.backend.message import AnyMessage
+from beeai_framework.workflows.v2.workflow import Workflow
+
+S = TypeVar("S", bound=Workflow)
+AnyCoroutine = Coroutine[Any, Any, Any]
+StartMethod = Callable[[S, list[AnyMessage]], AnyCoroutine]
+
+
+def start(func: StartMethod[S]) -> StartMethod[S]:
+    func._is_step = True  # type: ignore[attr-defined]
+    func._is_start = True  # type: ignore[attr-defined]
+    return func

python/beeai_framework/workflows/v2/decorators/when.py

@@ -0,0 +1,20 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Callable
+
+from beeai_framework.workflows.v2.types import AsyncMethod
+
+Predicate = Callable[..., bool]
+
+
+def when(predicate: Predicate) -> Callable[[AsyncMethod], AsyncMethod]:
+    """
+    Async decorator: runs the async function only if `predicate` returns True.
+    """
+
+    def decorator(func: AsyncMethod) -> AsyncMethod:
+        func._when_predicate = predicate  # type: ignore[attr-defined]
+        return func
+
+    return decorator

python/beeai_framework/workflows/v2/events.py

@@ -0,0 +1,40 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from pydantic import BaseModel, ConfigDict
+
+from beeai_framework.workflows.v2.step import WorkflowStep
+
+
+class WorkflowStartEvent(BaseModel): ...
+
+
+class WorkflowStepEvent(BaseModel):
+    step: WorkflowStep
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+
+class WorkflowStartStepEvent(WorkflowStepEvent):
+    pass
+
+
+class WorkflowErrorEvent(WorkflowStepEvent):
+    error: Exception
+    attempt: int
+
+
+class WorkflowRetryStepEvent(WorkflowStepEvent):
+    error: Exception
+    attempt: int
+
+
+class WorkflowSuccessEvent(BaseModel): ...
+
+
+workflow_v2_event_types: dict[str, type] = {
+    "start": WorkflowStartEvent,
+    "start_step": WorkflowStartStepEvent,
+    "error": WorkflowErrorEvent,
+    "retry_step": WorkflowRetryStepEvent,
+    "success": WorkflowSuccessEvent,
+}

python/beeai_framework/workflows/v2/step.py

@@ -0,0 +1,115 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+from beeai_framework.workflows.v2.decorators.when import Predicate
+from beeai_framework.workflows.v2.types import AsyncMethod, ExecutionCondition
+
+
+class WorkflowStepExecution(BaseModel):
+    inputs: list[Any] = []
+    output: Any | None
+    error: Exception | None
+    started_at: datetime | None
+    ended_at: datetime | None
+    duration: float
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+
+class WorkflowStep:
+    def __init__(
+        self,
+        func: AsyncMethod,
+        start: bool = False,
+        end: bool = False,
+        fork: bool = False,
+        join: bool = False,
+        retries: int = 0,
+    ) -> None:
+        self._func = func
+        self._name = func.__name__
+
+        self._is_start = start
+        self._is_end = end
+        self._is_fork = fork
+        self._is_join = join
+        self._retries = retries
+
+        self._condition: ExecutionCondition = "and"
+        self.forked: list[WorkflowStep] = []
+        self._dependencies: list[WorkflowStep] = []
+        self._dependents: list[WorkflowStep] = []
+        self.completed_dependencies: list[WorkflowStep] = []
+        self.inputs: dict[int, Any] = {}
+
+        self.completed_event = asyncio.Event()
+        self._predicates: list[Predicate] = []
+        self._executions: list[WorkflowStepExecution] = []
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def func(self) -> AsyncMethod:
+        return self._func
+
+    @property
+    def is_start(self) -> bool:
+        return self._is_start
+
+    @property
+    def is_end(self) -> bool:
+        return self._is_end
+
+    @property
+    def is_fork(self) -> bool:
+        return self._is_fork
+
+    @property
+    def is_join(self) -> bool:
+        return self._is_join
+
+    @property
+    def retries(self) -> int:
+        return self._retries
+
+    @property
+    def condition(self) -> ExecutionCondition:
+        return self._condition
+
+    @condition.setter
+    def condition(self, value: ExecutionCondition) -> None:
+        self._condition = value
+
+    def add_predicate(self, predicate: Predicate) -> None:
+        self._predicates.append(predicate)
+
+    @property
+    def predicates(self) -> list[Predicate]:
+        return self._predicates
+
+    @property
+    def executions(self) -> list[WorkflowStepExecution]:
+        return self._executions
+
+    def add_dependency(self, dep: "WorkflowStep") -> None:
+        self._dependencies.append(dep)
+        dep._dependents.append(self)
+
+    @property
+    def dependencies(self) -> list["WorkflowStep"]:
+        return self._dependencies
+
+    @property
+    def dependents(self) -> list["WorkflowStep"]:
+        return self._dependents
+
+    def last_execution(self) -> WorkflowStepExecution | None:
+        return self.executions[-1] if self.executions else None

python/beeai_framework/workflows/v2/types.py

@@ -0,0 +1,16 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Callable, Coroutine
+from typing import Any, Literal
+
+from pydantic import BaseModel
+
+AsyncMethod = Callable[..., Coroutine[Any, Any, Any]]
+
+ExecutionCondition = Literal["and", "or"]
+
+
+class AsyncMethodSet(BaseModel):
+    methods: list[str] = []
+    condition: ExecutionCondition = "and"