Temporal for Enterprise AI: Architecture & Integration

from datetime import timedelta
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker

@activity.defn
async def execute_llm_tool_call(prompt: str) -> str:
  """Non-deterministic activity executing external LLM call."""
  # API call to LLM provider
  return f"Processed LLM tool output for prompt: {prompt[:30]}"

@activity.defn
async def rollback_transaction(transaction_id: str):
  """Compensating saga activity to reverse state on failure."""
  print(f"Rolling back financial transaction: {transaction_id}")

@workflow.defn
class AgentExecutionWorkflow:
  def __init__(self):
      self._approved = False

  @workflow.signal
  def approve_execution(self):
      self._approved = True

  @workflow.run
  async def run(self, prompt: str) -> str:
      # Step 1: Execute primary LLM Activity
      result = await workflow.execute_activity(
          execute_llm_tool_call,
          prompt,
          start_to_close_timeout=timedelta(seconds=60),
      )

      # Step 2: Durable wait for human approval signal (can wait days)
      await workflow.wait_condition(lambda: self._approved)

      return f"Final Approved Agent Output: {result}"

async def main():
  client = await Client.connect("localhost:7233")
  worker = Worker(
      client,
      task_queue="agent-task-queue",
      workflows=[AgentExecutionWorkflow],
      activities=[execute_llm_tool_call, rollback_transaction],
  )
  print("Temporal Worker initialized and polling task queue...")
  await worker.run()

if __name__ == "__main__":
  import asyncio
  asyncio.run(main())