Durable execution with Temporal

pip install "gwenflow[temporal]"
temporal server start-dev   # a local Temporal server for development

import asyncio
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker

with workflow.unsafe.imports_passed_through():
    from gwenflow import Agent, ChatOpenAI, Tool
    from gwenflow.durability import TemporalDurability, worker_options


def get_exchange_rate(currency_iso: str) -> str:
    """Get the current USD exchange rate for a currency, given its ISO code."""
    ...


agent = Agent(
    name="fx",                          # stable: activity names derive from it
    instructions="Answer in one sentence and mention the date of the rate.",
    llm=ChatOpenAI(model="gpt-4o-mini"),
    tools=[Tool(get_exchange_rate)],
    durability=TemporalDurability(),    # one instance per agent
)


@workflow.defn
class ExchangeRates:
    @workflow.run
    async def run(self, currency: str) -> str:
        await agent.arun(f"What's the exchange rate of {currency}?")   # model and tools are activities
        response = await agent.arun("And the yen?")                     # same run, same history
        return response.content


async def main():
    client = await Client.connect("localhost:7233")
    async with Worker(client, task_queue="agents", **worker_options(ExchangeRates, agents=[agent])):
        print(await client.execute_workflow(ExchangeRates.run, "EUR", id="exchange-rates", task_queue="agents"))

asyncio.run(main())

from datetime import timedelta
from gwenflow.durability import AgentWorkflow


@workflow.defn
class FxDesk(AgentWorkflow):
    __gwenflow_agents__ = [agent]   # registered by worker_options(FxDesk)

    @workflow.run
    async def run(self, prompt: str) -> str:
        response = await self.run_agent(agent, prompt, timeout=timedelta(hours=1))
        return response.content

handle = await client.start_workflow(FxDesk.run, "Convert 1000 USD into euros.", id="fx-1", task_queue="fx")

pending = await handle.query(FxDesk.get_pending_approvals)    # [{"tool_call_id", "tool_name", "tool_args", ...}]
await handle.signal(FxDesk.approve, args=[pending[0]["tool_call_id"], True])   # or False to deny

print(await handle.result())

from datetime import timedelta
from temporalio.common import RetryPolicy

TemporalDurability(
    activity_config={"start_to_close_timeout": timedelta(seconds=60)},          # every activity
    model_activity_config={"start_to_close_timeout": timedelta(minutes=10)},    # model requests and summarizer
    tool_activity_config={                                                      # per tool name
        "search_web": {"start_to_close_timeout": timedelta(minutes=2), "retry_policy": RetryPolicy(maximum_attempts=3)},
    },
)

def restart(agent):
    workflow.continue_as_new(args=[[m.to_dict() for m in agent.history.messages]])

agent = Agent(name="assistant", llm=..., durability=TemporalDurability(continue_as_new=restart))


@workflow.defn
class Assistant:
    @workflow.run
    async def run(self, messages: list) -> str:
        copy = agent.durability.workflow_agent(agent)   # the per-run copy, seeded with the history carried over
        for message in messages:
            copy.history.add_message(message)
        ...

from gwenflow.flows import Flow

flow = Flow.from_file("flows/news_summary.yaml", durability=TemporalDurability())


@workflow.defn
class NewsSummary:
    @workflow.run
    async def run(self) -> str:
        results = await flow.arun()   # each node runs as an activity
        return results["summarize"]


worker = Worker(client, task_queue="flows", **worker_options(NewsSummary))

from gwenflow.durability import DurableFlow

worker = Worker(client, task_queue="flows", **worker_options(DurableFlow))

handle = await client.start_workflow(DurableFlow.run, flow.to_dict(), id="news-1", task_queue="flows")
await handle.query(DurableFlow.progress)   # {"fetch_news": "done", "summarize": "running"}
results = await handle.result()

from datetime import timedelta
from temporalio.client import Schedule, ScheduleActionStartWorkflow, ScheduleIntervalSpec, ScheduleSpec

await client.create_schedule(
    "news-summary-daily",
    Schedule(
        action=ScheduleActionStartWorkflow(NewsSummary.run, id="news-summary", task_queue="flows"),
        spec=ScheduleSpec(intervals=[ScheduleIntervalSpec(every=timedelta(days=1))]),
    ),
)

Worker(client, task_queue="...", **worker_options(*workflow_classes, agents=[...]))

import dataclasses
from temporalio.client import Client
from temporalio.converter import DataConverter
from gwenflow.durability import ClaimCheckCodec, S3BlobStore

store = S3BlobStore("my-bucket", prefix="temporal-payloads/", endpoint_url="https://s3.fr-par.scw.cloud")
converter = dataclasses.replace(DataConverter.default, payload_codec=ClaimCheckCodec(store, threshold=128 * 1024))
client = await Client.connect("localhost:7233", data_converter=converter)