temporal-workflows

Files (1)

• SKILL.md

SKILL.md (first lines)

# Temporal Workflow Patterns

This skill provides comprehensive guidance for building robust Temporal workflows with OpenAI agents integration.

## Core Concepts

Temporal is a durable execution platform that orchestrates distributed systems through workflows, activities, and workers.

**Key Terminology:**
- **Workflow**: Definition of how tasks should be executed (idempotent, durable)
- **Activity**: Unit of work that can be retried, timed out, and scheduled
- **Worker**: Long-running process that polls for tasks and executes activities
- **Activity Task**: An instruction to an activity instance with inputs and a result
- **Workflow Execution**: Single run of a workflow with a unique ID

## Workflow Design Principles

### 1. Idempotency
Workflows must be idempotent - running the same workflow multiple times with the same inputs should produce the same result.

```python
@workflow.defn
class ProcessOrderWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> OrderResult:
        # Check if already processed
        existing = await workflow.execute_activity(
            check_order_status,
            order_id,
            task_queue="order-processing"
        )

        if existing.processed:
            return OrderResult(
                status="already_processed",
                order_id=order_id
            )

        # Process the order
        result = await workflow.execute_activity(
            process_order,
            order_id,
            task_queue="order-processing"
        )

        return result
```

### 2. Deterministic Logic
Workflows should not depend on external state or nondeterministic operations.

**Good:**
```python
@workflow.defn
class GenerateReportWorkflow:
    @workflow.run
    async def run(self, date: datetime) -> Report:
        # Use inputs only, not external state
        data = await workflow.execute_activity(
            fetch_data,
            date,
            task_queue="report-generation"
        )
        return data
```

**Bad:**
```python
@workflow.defn
class GenerateReportWorkflow:
    @workflow.run
    async def run(self) -> Report:
        # Reading from global state - nondeterministic
        return global_data_store.get_latest()
```

### 3. Error Handling
Use Temporal's built-in retry mechanisms for transient failures.

```python
@activity.defn
async def call_openai_api(prompt: str) -> str:
    max_attempts = 3
    initial_interval = timedelta(seconds=1)
    backoff_coefficient = 2.0

    for attempt in range(max_attempts):
        try:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content
        except RateLimitError as e:
            if attempt < max_attempts - 1:
                await asyncio.sleep(initial_interval * (backoff_coefficient ** attempt))
                continue
            raise
```

### 4. Signal Handling
Use signals for asynchronous communication and cancellation.

```python
from temporalio import workflow

class ProcessingWorkflow:
    @workflow.run
    async def run(self, data: str) -> None:
        # Set up signal handler
        query_handler = workflow.wait_condition(
            self.processing_query.set,
            timeout=timedelta(hours=24)
        )

        # Start activity
        await workflow.execute_activity(
            long_running_task,
            data,

From yasinnerten/temporal-openai-agents-sdk, Apache-2.0.

More from yasinnerten/temporal-openai-agents-sdk