AI orchestration vs event-driven durable functions

from typing import Annotated

from zenml import pipeline, step, wait
from zenml.client import Client

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

@step
def draft(brief: str) -> Annotated[str, "draft"]:
  return call_llm(f"Write a draft from:\n{brief}")

@step(enable_cache=False)
def evaluate(text: str) -> Annotated[dict, "scores"]:
  return score(text)

@pipeline(dynamic=True)
def review_pipeline(topic: str):
  text = draft(research(topic))
  approved = wait(
      schema=bool,
      question="Approve this draft for evaluation?",
      name="human_approval",
  )
  if approved:
      evaluate(text)

review_pipeline("AI orchestration")

# The run pauses at wait(). Resolve it in the dashboard, or:
#   zenml pipeline runs wait-conditions resolve --run <id> --interactive
#   zenml pipeline runs resume <id>
# Artifacts land in your own bucket, versioned, loadable by name:
Client().get_artifact_version("draft").load()

# Serve it, and every request becomes a tracked run:
#   zenml pipeline deploy run.review_pipeline --name review-svc

import datetime

import inngest

client = inngest.Inngest(app_id="review")

@client.create_function(
  fn_id="review",
  trigger=inngest.TriggerEvent(event="review/requested"),
)
def review(ctx: inngest.ContextSync) -> dict:
  topic = ctx.event.data["topic"]

  brief = ctx.step.run(
      "research", lambda: call_llm(f"Research: {topic}")
  )
  text = ctx.step.run(
      "draft", lambda: call_llm(f"Write a draft from:\n{brief}")
  )

  # Durable wait: releases compute, resumes on the event.
  approval = ctx.step.wait_for_event(
      "await-approval",
      event="review/approved",
      timeout=datetime.timedelta(hours=24),
  )
  if approval is None:
      return {"status": "timed out"}

  return ctx.step.run("evaluate", lambda: score(text))

# Each step.run is memoized, so a retry doesn't repeat
# finished work. The platform drives the function and holds
# the run history.