diff --git a/docs/best-practices/task-orchestration.mdx b/docs/best-practices/task-orchestration.mdx index 28d0e35d5..52b2abf43 100644 --- a/docs/best-practices/task-orchestration.mdx +++ b/docs/best-practices/task-orchestration.mdx @@ -238,6 +238,72 @@ analysis_tasks = { } ``` +### Async Loop Expansion + +A `task_type="loop"` start task expands into one subtask per CSV/text row on the async path as well as the sync path. + + +**Async parity ([PR #3326](https://github.com/MervinPraison/PraisonAI/pull/3326)):** A `task_type="loop"` start task now pre-expands one subtask per input-file row on the **async** path — `await process.aworkflow()` and `process.workflow()` produce the same subtask set. Before this fix, the async engine skipped pre-expansion and ran the loop task once as an ordinary task, so a 500-row input file processed only the first row. + + +```python +import asyncio +from praisonaiagents import Agent, Task, Process + +analyzer = Agent(name="Analyzer", instructions="Analyze one row at a time") + +loop_task = Task( + name="analyze_row", + description="Analyze row {row}", + agent=analyzer, + task_type="loop", + input_file="rows.csv", # one Task per row, on the async path too + is_start=True, +) + +process = Process(tasks={"analyze_row": loop_task}, agents=[analyzer]) + +async def main(): + async for task_id in process.aworkflow(): + print(f"Completed: {task_id}") + +asyncio.run(main()) +``` + +For an N-row CSV, the async iteration yields N subtasks (`analyze_row_1`, `analyze_row_2`, …). The parent loop task is marked `completed` after pre-expansion; the subtasks carry the actual work. + +```mermaid +graph LR + subgraph "Loop-task expansion" + Csv[📄 rows.csv] --> PreExp[🧩 pre-expansion] + PreExp --> S1[loop_start_1] + PreExp --> S2[loop_start_2] + PreExp --> S3[loop_start_3] + end + + subgraph "Engines (same output)" + Sync[⚙️ workflow] + Async[⚡ aworkflow] + end + + S1 --> Sync + S1 --> Async + S2 --> Sync + S2 --> Async + S3 --> Sync + S3 --> Async + + classDef input fill:#8B0000,stroke:#7C90A0,color:#fff + classDef process fill:#189AB4,stroke:#7C90A0,color:#fff + classDef sub fill:#F59E0B,stroke:#7C90A0,color:#fff + classDef engine fill:#10B981,stroke:#7C90A0,color:#fff + + class Csv input + class PreExp process + class S1,S2,S3 sub + class Sync,Async engine +``` + ### The Retry Pattern Implement robust retry logic: diff --git a/docs/features/thread-safety.mdx b/docs/features/thread-safety.mdx index 7e1a8ef3c..b96a60a4b 100644 --- a/docs/features/thread-safety.mdx +++ b/docs/features/thread-safety.mdx @@ -453,6 +453,7 @@ PraisonAI provides comprehensive thread-safety for HTTP server deployment: - If two launch calls use the same path on the same port, the second gets an auto-suffixed path (`/path_abc123`) and a warning is logged. - Server readiness is signalled deterministically (no fixed sleep); `.launch()` returns only after the port is accepting connections. The wait defaults to **5 seconds** and is configurable via the `PRAISONAI_SERVER_READY_TIMEOUT` environment variable. If the server doesn't become ready in time, `.launch()` still returns and a warning is logged — check server logs for startup errors. - `aworkflow()` state lock is created inside the running async context, so workflows remain stable when invoked under pytest-asyncio or when nested inside another loop. +- Per-turn tool tracking (`_turn_tools_used`) is now protected by the same lock that guards `chat_history`, so concurrent `chat()` / `achat()` turns on one `Agent` no longer corrupt hook or self-improve tool data. No new public API — the buffer stays private; the read/append helpers were made async-safe internally. ```python import threading