Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions docs/best-practices/task-orchestration.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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.

<Note>
**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.
</Note>

```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:
Expand Down
1 change: 1 addition & 0 deletions docs/features/thread-safety.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading