phowell/pocketflow
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update parallel for translations

Browse Source
This commit is contained in:
zachary62
2025-05-05 16:52:13 -04:00
parent f4029f4999
commit 4526a358b7
15 changed files with 538 additions and 534 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cookbook/pocketflow-parallel-batch/README.md
+58 -41
View File
@@ -1,52 +1,69 @@
# Sequential vs Parallel Processing
# PocketFlow Parallel Batch Translation Example
Demonstrates how AsyncParallelBatchNode accelerates processing by 3x over AsyncBatchNode.
This example demonstrates translating a document (the main PocketFlow README.md) into multiple target languages concurrently using PocketFlow's parallel batch processing capabilities (`AsyncParallelBatchNode` and `AsyncFlow`).
## Features
It showcases how to leverage asynchronous operations and parallelism to potentially speed up I/O-bound tasks, such as making multiple LLM API calls simultaneously.
- Processes identical tasks with two approaches
- Compares sequential vs parallel execution time
- Shows 3x speed improvement with parallel processing
## Goal
## Run It
Translate the content of `PocketFlow/README.md` into a predefined list of languages:
`["Chinese", "Spanish", "Japanese", "German", "Russian", "Portuguese", "French", "Korean"]`
The primary focus is to execute these translation tasks *in parallel* and measure the total time taken, allowing for comparison with a sequential approach (like the one demonstrated in the standard `pocketflow-batch` example).
## PocketFlow Concepts Used
- **`AsyncParallelBatchNode`**: Processes an iterable (the list of target languages) by running an asynchronous task (translation using an LLM) for each item concurrently.
- **`AsyncFlow`**: Manages the execution of flows containing asynchronous nodes.
- **Asynchronous Utility**: A helper function (`call_llm_async`) that interacts with the Anthropic API asynchronously.
## File Structure
```
pocketflow-parallel-batch/
├── main.py # Defines the PocketFlow node and flow, orchestrates the parallel translation
├── utils.py # Contains the asynchronous `call_llm_async` utility using Anthropic API
├── requirements.txt # Dependencies: pocketflow, anthropic, python-dotenv, httpx
└── README.md # This explanation file
```
## Setup
1. **Navigate to the example directory**:
```bash
cd cookbook/pocketflow-parallel-batch
```
2. **Create and activate a virtual environment** (recommended):
```bash
python -m venv venv
source venv/bin/activate # Or `venv\Scripts\activate` on Windows
```
3. **Install dependencies**:
```bash
pip install -r requirements.txt
```
4. **Set up Anthropic API Key**: Create a `.env` file in this directory:
```
ANTHROPIC_API_KEY=your_anthropic_api_key_here
```
Or, set the `ANTHROPIC_API_KEY` environment variable.
## Running the Example
Execute the main script from within the `pocketflow-parallel-batch` directory:
```bash
pip install pocketflow
python main.py
```
## Output
The script will:
1. Read the content of `../../README.md`.
2. Initiate the `AsyncFlow`.
3. The `ParallelTranslateReadme` node will concurrently request translations for the README content into each target language via the Anthropic API.
4. Print status messages for each requested and received translation.
5. Report the list of languages for which translations were successfully generated.
6. Display the total time taken for the entire parallel process.
```
=== Running Sequential (AsyncBatchNode) ===
[Sequential] Summarizing file1.txt...
[Sequential] Summarizing file2.txt...
[Sequential] Summarizing file3.txt...
## Parallel vs. Sequential Comparison
=== Running Parallel (AsyncParallelBatchNode) ===
[Parallel] Summarizing file1.txt...
[Parallel] Summarizing file2.txt...
[Parallel] Summarizing file3.txt...
Sequential took: 3.00 seconds
Parallel took: 1.00 seconds
```
## Key Points
- **Sequential**: Total time = sum of all item times
- Good for: Rate-limited APIs, maintaining order
- **Parallel**: Total time ≈ longest single item time
- Good for: I/O-bound tasks, independent operations
## Tech Dive Deep
- **Python's GIL** prevents true CPU-bound parallelism, but LLM calls are I/O-bound
- **Async/await** overlaps waiting time between requests
- Example: `await client.chat.completions.create(...)`
- See: [OpenAI's async usage](https://github.com/openai/openai-python?tab=readme-ov-file#async-usage)
For maximum performance and cost efficiency, consider using batch APIs:
- [OpenAI's Batch API](https://platform.openai.com/docs/guides/batch) lets you process multiple prompts in a single request
- Reduces overhead and can be more cost-effective for large workloads
Note the total execution time reported by this script. Compare it to the time it would take if each language translation were performed one after the other (sequentially). For tasks involving multiple independent API calls like this, the parallel approach using `AsyncParallelBatchNode` is expected to be significantly faster, limited primarily by the LLM API's response time and potential rate limits, rather than the sum of individual call durations.
cookbook/pocketflow-parallel-batch/main.py
+83 -83
View File
@@ -1,103 +1,103 @@
import asyncio
import time
import os
from pocketflow import AsyncFlow, AsyncParallelBatchNode
from utils import call_llm
from pocketflow import AsyncBatchNode, AsyncParallelBatchNode, AsyncFlow
####################################
# Dummy async function (1s delay)
####################################
async def dummy_llm_summarize(text):
"""Simulates an async LLM call that takes 1 second."""
await asyncio.sleep(1)
return f"Summarized({len(text)} chars)"
###############################################
# 1) AsyncBatchNode (sequential) version
###############################################
class SummariesAsyncNode(AsyncBatchNode):
"""
Processes items sequentially in an async manner.
The next item won't start until the previous item has finished.
"""
# --- Node Definitions ---
class TranslateTextNodeParallel(AsyncParallelBatchNode):
"""Translates README into multiple languages in parallel and saves files."""
async def prep_async(self, shared):
# Return a list of items to process.
# Each item is (filename, content).
return list(shared["data"].items())
"""Reads text and target languages from shared store."""
text = shared.get("text", "(No text provided)")
languages = shared.get("languages", [])
return [(text, lang) for lang in languages]
async def exec_async(self, item):
filename, content = item
print(f"[Sequential] Summarizing {filename}...")
summary = await dummy_llm_summarize(content)
return (filename, summary)
async def exec_async(self, data_tuple):
"""Calls the async LLM utility for each target language."""
text, language = data_tuple
prompt = f"""
Please translate the following markdown file into {language}.
But keep the original markdown format, links and code blocks.
Directly return the translated text, without any other text or comments.
Original:
{text}
Translated:"""
result = await call_llm(prompt)
print(f"Translated {language} text")
return {"language": language, "translation": result}
async def post_async(self, shared, prep_res, exec_res_list):
# exec_res_list is a list of (filename, summary)
shared["sequential_summaries"] = dict(exec_res_list)
return "done_sequential"
"""Stores the dictionary of {language: translation} pairs and writes to files."""
output_dir = shared.get("output_dir", "translations")
os.makedirs(output_dir, exist_ok=True)
for result in exec_res_list:
if isinstance(result, dict):
language = result.get("language", "unknown")
translation = result.get("translation", "")
filename = os.path.join(output_dir, f"README_{language.upper()}.md")
try:
import aiofiles
async with aiofiles.open(filename, "w", encoding="utf-8") as f:
await f.write(translation)
print(f"Saved translation to {filename}")
except ImportError:
with open(filename, "w", encoding="utf-8") as f:
f.write(translation)
print(f"Saved translation to {filename} (sync fallback)")
except Exception as e:
print(f"Error writing file {filename}: {e}")
else:
print(f"Warning: Skipping invalid result item: {result}")
return "default"
###############################################
# 2) AsyncParallelBatchNode (concurrent) version
###############################################
# --- Flow Creation ---
class SummariesAsyncParallelNode(AsyncParallelBatchNode):
"""
Processes items in parallel. Many LLM calls start at once.
"""
def create_parallel_translation_flow():
"""Creates and returns the parallel translation flow."""
translate_node = TranslateTextNodeParallel(max_retries=3)
return AsyncFlow(start=translate_node)
async def prep_async(self, shared):
return list(shared["data"].items())
async def exec_async(self, item):
filename, content = item
print(f"[Parallel] Summarizing {filename}...")
summary = await dummy_llm_summarize(content)
return (filename, summary)
async def post_async(self, shared, prep_res, exec_res_list):
shared["parallel_summaries"] = dict(exec_res_list)
return "done_parallel"
###############################################
# Demo comparing the two approaches
###############################################
# --- Main Execution ---
async def main():
# We'll use the same data for both flows
shared_data = {
"data": {
"file1.txt": "Hello world 1",
"file2.txt": "Hello world 2",
"file3.txt": "Hello world 3",
}
source_readme_path = "../../README.md"
try:
with open(source_readme_path, "r", encoding='utf-8') as f:
text = f.read()
except FileNotFoundError:
print(f"Error: Could not find the source README file at {source_readme_path}")
exit(1)
except Exception as e:
print(f"Error reading file {source_readme_path}: {e}")
exit(1)
shared = {
"text": text,
"languages": ["Chinese", "Spanish", "Japanese", "German", "Russian", "Portuguese", "French", "Korean"],
"output_dir": "translations"
}
# 1) Run the sequential version
seq_node = SummariesAsyncNode()
seq_flow = AsyncFlow(start=seq_node)
translation_flow = create_parallel_translation_flow()
print("\n=== Running Sequential (AsyncBatchNode) ===")
t0 = time.time()
await seq_flow.run_async(shared_data)
t1 = time.time()
print(f"Starting parallel translation into {len(shared['languages'])} languages...")
start_time = time.perf_counter()
# 2) Run the parallel version
par_node = SummariesAsyncParallelNode()
par_flow = AsyncFlow(start=par_node)
await translation_flow.run_async(shared)
print("\n=== Running Parallel (AsyncParallelBatchNode) ===")
t2 = time.time()
await par_flow.run_async(shared_data)
t3 = time.time()
# Show times
print("\n--- Results ---")
print(f"Sequential Summaries: {shared_data.get('sequential_summaries')}")
print(f"Parallel Summaries: {shared_data.get('parallel_summaries')}")
print(f"Sequential took: {t1 - t0:.2f} seconds")
print(f"Parallel took: {t3 - t2:.2f} seconds")
end_time = time.perf_counter()
duration = end_time - start_time
print(f"\nTotal parallel translation time: {duration:.4f} seconds")
print("\n=== Translation Complete ===")
print(f"Translations saved to: {shared['output_dir']}")
print("============================")
if __name__ == "__main__":
asyncio.run(main())
asyncio.run(main())
cookbook/pocketflow-parallel-batch/requirements.txt
+5
View File
@@ -0,0 +1,5 @@
pocketflow>=0.0.2
anthropic>=0.15.0
python-dotenv
httpx
aiofiles
cookbook/pocketflow-parallel-batch/utils.py
+30
View File
@@ -0,0 +1,30 @@
import os
import asyncio
from anthropic import AsyncAnthropic
# Async version of the simple wrapper, using Anthropic
async def call_llm(prompt):
"""Async wrapper for Anthropic API call."""
client = AsyncAnthropic(api_key=os.environ.get("ANTHROPIC_API_KEY", "your-api-key"))
response = await client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=20000,
thinking={
"type": "enabled",
"budget_tokens": 16000
},
messages=[
{"role": "user", "content": prompt}
],
)
return response.content[1].text
if __name__ == "__main__":
async def run_test():
print("## Testing async call_llm with Anthropic")
prompt = "In a few words, what is the meaning of life?"
print(f"## Prompt: {prompt}")
response = await call_llm(prompt)
print(f"## Response: {response}")
asyncio.run(run_test())