AI orchestration or durable execution on Postgres

from typing import Annotated

from zenml import log_metadata, pipeline, step, wait

@step
def plan(goal: str) -> Annotated[list[str], "tasks"]:
  tasks = call_llm(f"Break down: {goal}")
  # No llm() primitive: attach whatever you want recorded.
  log_metadata(metadata={"model": "fast", "tasks": len(tasks)})
  return tasks

@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:
#   zenml stack set sagemaker && python agent_pipeline.py

from dbos import DBOS

@DBOS.step()
def plan(goal: str) -> list[str]:
  return call_llm(f"Break down: {goal}")

@DBOS.step()
def research(task: str) -> str:
  return call_llm(f"Research: {task}")

@DBOS.step()
def deploy(findings: list[str]) -> dict:
  return ship(findings)

@DBOS.workflow()
def agent_workflow(goal: str) -> dict:
  tasks = plan(goal)
  findings = [research(t) for t in tasks]
  # Durable wait: checkpointed in Postgres. Another process
  # delivers the answer with DBOS.send(workflow_id, ok, "approval").
  approved = DBOS.recv("approval", timeout_seconds=86400)
  if not approved:
      return {"status": "rejected"}
  return deploy(findings)

# Durability comes from the library plus your Postgres.
# No orchestrator process to operate, and no artifact
# store: what a step returned is yours to persist.
DBOS(config={"name": "agent"})
DBOS.launch()