AI orchestration or general durable execution

from typing import Annotated

from zenml import pipeline, step, wait
from zenml.config.retry_config import StepRetryConfig

@step(retry=StepRetryConfig(max_retries=3, delay=10, backoff=2))
def plan(goal: str) -> Annotated[list[str], "tasks"]:
  return call_llm(f"Break down: {goal}")

@step
def research(task: str) -> Annotated[str, "finding"]:
  return call_llm(f"Research: {task}")

@step
def deploy(findings: list[str]) -> Annotated[dict, "release"]:
  return ship(findings)

@pipeline(dynamic=True)
def agent_pipeline(goal: str):
  tasks = plan(goal)
  findings = research.map(task=tasks)   # fan-out decided at run time

  # Pause for approval. The run goes PAUSED, the process can be
  # torn down, and a human resolves it from the dashboard or CLI.
  approved = wait(schema=bool, question="Approve and deploy?")
  if approved:
      deploy(findings)

agent_pipeline("prepare the release notes")

# Resolve and resume later; completed steps are reused:
#   zenml pipeline runs wait-conditions resolve --run <id> --interactive
#   zenml pipeline runs resume <id>
# Same code on another cloud, no edits:
#   zenml stack set vertex_stack && python agent_pipeline.py

from datetime import timedelta
from temporalio import activity, workflow

@activity.defn
async def plan(goal: str) -> list[str]:
  return await call_llm(f"Break down: {goal}")

@activity.defn
async def research(task: str) -> str:
  return await call_llm(f"Research: {task}")

@workflow.defn
class AgentFlow:
  def __init__(self) -> None:
      self._approved: bool | None = None

  @workflow.signal
  def approve(self, ok: bool) -> None:
      self._approved = ok

  @workflow.run
  async def run(self, goal: str) -> str:
      tasks = await workflow.execute_activity(
          plan, goal, start_to_close_timeout=timedelta(minutes=5)
      )
      findings = [
          await workflow.execute_activity(
              research, t, start_to_close_timeout=timedelta(minutes=5)
          )
          for t in tasks
      ]
      # Durable wait: the Worker holds the workflow, history
      # replays it after a restart.
      await workflow.wait_condition(lambda: self._approved is not None)
      return "\n".join(findings) if self._approved else "Rejected"

# Run via: await client.execute_workflow(AgentFlow.run, goal,
#   id=..., task_queue=...); approval arrives via
# client.get_workflow_handle(...).signal(AgentFlow.approve, True).