AI orchestration or durable execution for services

from typing import Annotated

from zenml import pipeline, step, wait

@step
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>
# No server in the execution path. Artifacts land in your bucket.
# Same code on another cloud:
#   zenml stack set vertex_stack && python agent_pipeline.py

import restate
from restate import Workflow, WorkflowContext, WorkflowSharedContext

# Durable promises live on Workflow handlers, keyed by workflow id.
agent = Workflow("agent")

@agent.main()
async def run(ctx: WorkflowContext, goal: str) -> dict:
  # ctx.run_typed journals each result, so a crash resumes here
  # rather than re-executing the call.
  tasks = await ctx.run_typed("plan", call_llm, prompt=f"Break down: {goal}")
  findings = [
      await ctx.run_typed(f"research-{i}", call_llm, prompt=f"Research: {t}")
      for i, t in enumerate(tasks)
  ]

  # Durable promise: journalled by the server, resolved by the
  # approve handler below, survives restarts.
  approved = await ctx.promise("approval", type_hint=bool).value()
  if not approved:
      return {"status": "rejected"}
  return await ctx.run_typed("deploy", ship, findings=findings)

@agent.handler()
async def approve(ctx: WorkflowSharedContext, ok: bool) -> None:
  await ctx.promise("approval", type_hint=bool).resolve(ok)

# Handlers register with the Restate server, which coordinates
# invocations and owns the journal.
app = restate.app([agent])