Parallel Workflows

import concurrent.futures
from typing import List, Tuple

def run_parallel_nodes(
    node_fns: list,
    state: dict,
    timeout_seconds: int = 30
) -> List[Tuple[str, str]]:
    """
    Run multiple workflow nodes in parallel.
    Returns list of (node_name, condition) pairs.
    """
    results = []

    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Submit all nodes for parallel execution
        futures = {
            executor.submit(fn, state): name
            for name, fn in node_fns
        }

        # Collect results as they complete
        for future in concurrent.futures.as_completed(futures, timeout=timeout_seconds):
            node_name = futures[future]
            try:
                condition = future.result()
                results.append((node_name, condition))
                print(f"  [parallel] {node_name} → {condition}")
            except Exception as e:
                results.append((node_name, "error"))
                state["data"][f"{node_name}_error"] = str(e)
                print(f"  [parallel] {node_name} FAILED: {e}")

    return results

# Research 3 competitors in parallel
def research_asana(state):
    state["data"]["asana"] = {"price": "$10/mo", "key_feature": "Timeline view"}
    return "success"

def research_trello(state):
    state["data"]["trello"] = {"price": "$5/mo", "key_feature": "Kanban boards"}
    return "success"

def research_monday(state):
    state["data"]["monday"] = {"price": "$8/mo", "key_feature": "Custom workflows"}
    return "success"

state = {"data": {}}
results = run_parallel_nodes(
    [("research_asana", research_asana),
     ("research_trello", research_trello),
     ("research_monday", research_monday)],
    state
)
print(f"Completed {len(results)} parallel branches")
print("Competitors found:", list(state["data"].keys()))