Parallel Workflows

from polos import workflow, WorkflowContext
from polos.types.types import BatchWorkflowInput

@workflow(id="parallel_review")
async def parallel_review(ctx: WorkflowContext, payload):
    reviewers = ["alice", "bob", "charlie"]

    # Create batch of review requests
    requests = [
        BatchWorkflowInput(
            id="single_review",
            payload={"reviewer": r, "document_id": payload["doc_id"]},
        )
        for r in reviewers
    ]

    # Run all in parallel
    results = await ctx.step.batch_invoke_and_wait("reviews", requests)

    # Aggregate
    approved = sum(1 for r in results if r.result.get("approved"))
    return {"approved": approved, "total": len(results)}

import { defineWorkflow } from '@polos/sdk';
import type { BatchWorkflowInput } from '@polos/sdk';

const parallelReview = defineWorkflow<Record<string, unknown>, unknown, AggregatedReview>(
  { id: 'parallel_review' },
  async (ctx, payload) => {
    const reviewers = (payload['reviewers'] as string[]) ?? ['alice', 'bob', 'charlie'];

    // Create batch of review requests
    const reviewRequests: BatchWorkflowInput[] = reviewers.map((reviewer) => ({
      workflow: 'single_review',
      payload: { reviewerId: reviewer, documentId: payload['documentId'] },
    }));

    // Run all in parallel
    const results = await ctx.step.batchInvokeAndWait<ReviewResult>(
      'parallel_reviews',
      reviewRequests,
    );

    // Aggregate
    const approvedCount = results.filter((r) => r.approved).length;
    return { approved: approvedCount, total: results.length };
  },
);

@workflow(id="launch_background_tasks")
async def launch_tasks(ctx: WorkflowContext, payload):
    requests = [
        BatchWorkflowInput(id="background_task", payload={"task_id": i})
        for i in range(10)
    ]

    # Launch without waiting
    handles = await ctx.step.batch_invoke("launch", requests)

    return {"launched": len(handles), "ids": [h.id for h in handles]}

const fireAndForgetBatch = defineWorkflow<FireAndForgetPayload, unknown, Record<string, unknown>>(
  { id: 'fire_and_forget_batch' },
  async (ctx, payload) => {
    const taskRequests: BatchWorkflowInput[] = payload.tasks.map((task) => ({
      workflow: 'background_task',
      payload: { taskId: task.id, data: task.data },
    }));

    // Launch without waiting
    const handles = await ctx.step.batchInvoke(
      'launch_background_tasks',
      taskRequests,
    );

    return { launched: handles.length, executionIds: handles.map((h) => h.executionId) };
  },
);

@workflow(id="process_chunks")
async def process_chunks(ctx: WorkflowContext, payload):
    data = payload["data"]
    chunk_size = 10

    # Split into chunks
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]

    # Process in parallel
    requests = [
        BatchWorkflowInput(id="process_chunk", payload={"chunk": c, "index": i})
        for i, c in enumerate(chunks)
    ]
    results = await ctx.step.batch_invoke_and_wait("chunks", requests)

    # Combine results
    all_processed = []
    for r in results:
        all_processed.extend(r.result.get("processed", []))
    return {"total": len(all_processed)}

const dataChunkProcessor = defineWorkflow<ChunkProcessorPayload, unknown, ChunkProcessorResult>(
  { id: 'data_chunk_processor' },
  async (ctx, payload) => {
    const data = payload.data;
    const chunkSize = payload.chunkSize;

    // Split into chunks
    const chunks: unknown[][] = [];
    for (let i = 0; i < data.length; i += chunkSize) {
      chunks.push(data.slice(i, i + chunkSize));
    }

    // Process in parallel
    const chunkRequests: BatchWorkflowInput[] = chunks.map((chunk, i) => ({
      workflow: 'process_chunk',
      payload: { chunk, chunkIndex: i },
    }));
    const results = await ctx.step.batchInvokeAndWait<ChunkResult>(
      'parallel_chunks',
      chunkRequests,
    );

    // Combine results
    const allProcessed: unknown[] = [];
    for (const result of results) {
      allProcessed.push(...result.processed);
    }
    return { totalItems: data.length, chunksProcessed: results.length, processedItems: allProcessed.length, results: allProcessed };
  },
);

git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/13-parallel-review
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
uv sync
python main.py

git clone https://github.com/polos-dev/polos.git
cd polos/typescript-examples/13-parallel-review
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
npm install
npx tsx main.ts