### Install Dependencies

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Installs the necessary Python packages for the project using pip. This command reads the dependencies listed in the 'requirements.txt' file.

```bash
pip install -r requirements.txt
```

--------------------------------

### Add Demonstration Function in Python

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

This snippet shows how to add a new asynchronous demonstration function to the `main.py` file. It includes a placeholder for the demonstration logic and a print statement to indicate the start of the feature demonstration.

```python
async def demonstrate_my_feature():
    print("🎯 MY FEATURE DEMONSTRATION")
    # Your demonstration logic
```

--------------------------------

### Run Basic Synchronous Flow Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Command to navigate to the examples directory and run the basic synchronous flow tracing example.

```bash
cd examples
python basic_example.py
```

--------------------------------

### Run the Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Executes the main Python script that runs the comprehensive asynchronous example. This script orchestrates the flow execution and demonstrates PocketFlow's features.

```bash
python main.py
```

--------------------------------

### LangfuseTracer: Async Context Management

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Demonstrates enhanced methods on the LangfuseTracer for managing async trace context, including starting async traces and creating async contexts.

```Python
# Start async trace
async_context = await tracer.start_trace_async(flow_name, input_data)

# Create async context
context = tracer.create_async_context(flow_name, trace_id)
```

--------------------------------

### Update pocketflow-tracing Dependency

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Ensures you have the latest version of the pocketflow-tracing library installed for async support. This is a prerequisite for using the new features.

```bash
pip install --upgrade pocketflow-tracing
```

--------------------------------

### Migrate Database Operations to Async

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code demonstrates migrating a synchronous database operation to an asynchronous one. The 'Before' example uses a standard `Node` and synchronous database connection, while the 'After' example uses `AsyncNode` and an asynchronous database connection with `async with` for resource management.

```python
# Before
class DatabaseNode(Node):
    def exec(self, prep_res):
        conn = get_db_connection()
        result = conn.execute(query, prep_res)
        return result.fetchall()

# After
class DatabaseNode(AsyncNode):
    async def exec_async(self, prep_res):
        async with get_async_db_connection() as conn:
            result = await conn.execute(query, prep_res)
            return await result.fetchall()
```

--------------------------------

### Install pocketflow-tracing from Source or PyPI

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Details the methods for installing the pocketflow-tracing library, including installation from PyPI (when published), from a local source clone, and a development installation with extra dependencies.

```bash
# From PyPI (when published)
pip install pocketflow-tracing

# From source
git clone https://github.com/The-Pocket/PocketFlow.git
cd PocketFlow/cookbook/pocketflow-tracing
pip install .

# Development installation
pip install -e ".[dev]"
```

--------------------------------

### Run Asynchronous Flow Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Commands to run the basic asynchronous flow tracing example and the comprehensive asynchronous features example.

```bash
cd examples
python async_example.py

# For comprehensive async features demonstration
python comprehensive_async_example.py
```

--------------------------------

### LangfuseTracer: Async Flow Composition

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Shows enhanced LangfuseTracer methods for async flow composition, including starting flow composition and executing parallel async flows.

```Python
# Start flow composition
execution_id = await tracer.start_async_flow_composition(
    flow_name, "parallel", parent_execution_id
)

# Execute parallel flows
results = await tracer.execute_parallel_async_flows(
    flows, flow_names, parent_execution_id
)
```

--------------------------------

### Update Bash Commands to Run Comprehensive Async Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Shows the updated bash commands for running the comprehensive async example. The execution has been moved to a dedicated directory within the examples, requiring a `cd` command before running the main script.

```bash
# Old
python examples/comprehensive_async_example.py

# New
cd examples/comprehensive_async_example
python main.py
```

--------------------------------

### Manage Async Flow Composition with AsyncFlowCompositionManager

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Provides examples for AsyncFlowCompositionManager, which handles async flow composition and execution. It demonstrates executing flows in parallel, nested flows, and batch flows.

```Python
from pocketflow_tracing.async_composition import AsyncFlowCompositionManager

manager = AsyncFlowCompositionManager(tracer)

# Execute parallel flows
results = await manager.execute_parallel_flows(flows, flow_names)

# Execute nested flow
result = await manager.execute_nested_flow(flow, flow_name)

# Execute batch flows
results = await manager.execute_batch_flows(
    flow_factory, batch_data, flow_name, parallel=True
)
```

--------------------------------

### Execute Synchronous and Asynchronous Examples

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Provides bash commands to run the synchronous and asynchronous PocketFlow tracing examples. It also includes a command to execute the tracing tests.

```bash
# Basic synchronous example
cd cookbook/pocketflow-tracing/examples
python basic_example.py

# Asynchronous example  
python async_example.py

# Run tests
python ../test_tracing.py
```

--------------------------------

### LangfuseTracer: Async Node Lifecycle Management

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Illustrates enhanced LangfuseTracer methods for managing async node lifecycles. This includes starting node lifecycles, transitioning states, and suspending/resuming nodes.

```Python
# Start async node lifecycle
span_id = await tracer.start_async_node_lifecycle(node_id, node_name, phase)

# Transition node state
await tracer.transition_async_node_state(node_id, "running", phase)

# Suspend/resume node
await tracer.suspend_async_node(node_id, "await_operation")
await tracer.resume_async_node(node_id, "operation_complete")
```

--------------------------------

### Set Up Langfuse Environment

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Configures environment variables for Langfuse integration by creating a '.env' file. This file stores the Langfuse host and authentication keys required for tracing and monitoring.

```bash
echo "LANGFUSE_HOST=your-langfuse-host" > .env
echo "LANGFUSE_PUBLIC_KEY=your-public-key" >> .env
echo "LANGFUSE_SECRET_KEY=your-secret-key" >> .env
```

--------------------------------

### Install pocketflow-tracing

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Installs the pocketflow-tracing package using pip. Supports installation from source and development installations.

```bash
pip install pocketflow-tracing
```

```bash
git clone https://github.com/The-Pocket/PocketFlow.git
cd PocketFlow/cookbook/pocketflow-tracing
pip install .
```

```bash
pip install -e ".[dev]"
```

--------------------------------

### Migrate to Mixed Sync/Async Flows

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python example illustrates migrating to a mixed synchronous and asynchronous flow structure. It defines an `AsyncFlow` that incorporates both a traditional synchronous node (`SyncProcessingNode`) and a new asynchronous node (`AsyncProcessingNode`).

```python
@trace_flow(flow_name="MixedFlow")
class MixedFlow(AsyncFlow):
    def __init__(self):
        sync_node = SyncProcessingNode()      # Still synchronous
        async_node = AsyncProcessingNode()    # Now asynchronous
        
        sync_node.next(async_node)
        super().__init__(start=sync_node)
```

--------------------------------

### Enable Debug Mode with Bash

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

This command demonstrates how to enable debug mode for the PocketFlow tracing example by setting the `POCKETFLOW_DEBUG` environment variable to 1. It is followed by the command to run the main Python script.

```bash
export POCKETFLOW_DEBUG=1
python main.py
```

--------------------------------

### Test PocketFlow Components with Bash

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Provides bash commands to test individual components of the refactored PocketFlow example. This includes testing nodes, flow creation, data fetching utilities, processing utilities, concurrent utilities, and the main demonstration script.

```bash
python nodes.py
```

```bash
python flow.py
```

```bash
python utils/async_data_fetch.py
```

```bash
python utils/async_data_process.py
```

```bash
python utils/concurrent_utils.py
```

```bash
python main.py
```

--------------------------------

### Async Node Execution Tracing Wrapper

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Provides an example of how asynchronous node methods are wrapped to add tracing. It includes starting a span for the node's phase, executing the original async method, and ending the span, capturing any exceptions that occur.

```Python
async def traced_async_method(*args, **kwargs):
    span_id = self._tracer.start_node_span(node_name, node_id, phase)
    try:
        result = await original_method(*args, **kwargs)
        self._tracer.end_node_span(span_id, input_data=args, output_data=result)
        return result
    except Exception as e:
        self._tracer.end_node_span(span_id, input_data=args, error=e)
        raise
```

--------------------------------

### Custom Async Node Implementation

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Provides a Python code example for defining a custom asynchronous node in PocketFlow. It includes methods for asynchronous preparation ('prep_async'), execution ('exec_async'), and post-processing ('post_async'), along with sharing data between steps.

```python
class MyCustomNode(AsyncNode):
    async def prep_async(self, shared):
        # Preparation logic
        return prep_data
    
    async def exec_async(self, prep_res):
        # Main execution logic
        return result
    
    async def post_async(self, shared, prep_res, exec_res):
        # Post-processing logic
        shared["my_result"] = exec_res
        return "default"
```

--------------------------------

### Manage Async Node Lifecycles with AsyncLifecycleManager

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Illustrates the usage of AsyncLifecycleManager for tracking async node lifecycles and events. It covers starting lifecycle tracking and transitioning node states like running, suspended, and resumed.

```Python
from pocketflow_tracing.async_lifecycle import AsyncLifecycleManager
from pocketflow_tracing.core import AsyncNodeState

manager = AsyncLifecycleManager(tracer)

# Start lifecycle tracking
lifecycle = manager.start_lifecycle(node_id, node_name, phase)

# Transition states
manager.transition_state(node_id, AsyncNodeState.RUNNING)
manager.suspend_node(node_id, "await_operation")
manager.resume_node(node_id, "operation_complete")
```

--------------------------------

### Test Async Data Process Utilities

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Executes the 'async_data_process.py' script to test the asynchronous data processing utilities. This verifies different data processing strategies.

```bash
python utils/async_data_process.py
```

--------------------------------

### Mermaid Flowchart Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/guide_for_pocketflow.md

An example of a flowchart using the Mermaid syntax to visualize the flow of an AI system, including decision points and subgraphs.

```Mermaid
flowchart LR
    start[Start] --> batch[Batch]
    batch --> check[Check]
    check -->|OK| process
    check -->|Error| fix[Fix]
    fix --> check
    
    subgraph process[Process]
        step1[Step 1] --> step2[Step 2]
    end
    
    process --> endNode[End]
```

--------------------------------

### Test Nodes

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Runs the 'nodes.py' script to test the individual asynchronous node implementations independently. This helps in verifying the functionality of each node.

```bash
python nodes.py
```

--------------------------------

### Test Async Data Fetch Utilities

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Runs the 'async_data_fetch.py' script within the 'utils' directory to test the asynchronous data fetching utilities. This includes testing API calls and fallback mechanisms.

```bash
python utils/async_data_fetch.py
```

--------------------------------

### Python Main Entry Point Demonstrations

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

The main entry point orchestrates various demonstrations of asynchronous patterns, including basic flows, concurrent execution, error handling, and performance monitoring.

```python
# demonstrate_basic_async_flow(): Basic flow demonstration
# demonstrate_concurrent_flows(): Concurrent execution patterns
# demonstrate_error_handling(): Error recovery mechanisms
# demonstrate_performance_monitoring(): Performance tracking
# demonstrate_nested_flows(): Nested flow execution
```

--------------------------------

### Test Flows

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Executes the 'flow.py' script to test the defined PocketFlow asynchronous flows. This verifies the orchestration and execution of sequences of nodes.

```bash
python flow.py
```

--------------------------------

### Test Concurrent Utilities

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Runs the 'concurrent_utils.py' script to test utilities related to concurrent processing. This includes testing concurrency limits and performance measurement.

```bash
python utils/concurrent_utils.py
```

--------------------------------

### Install Langfuse Package

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Installs the necessary Langfuse package using pip. This is a prerequisite for using Langfuse tracing features.

```bash
pip install langfuse
```

--------------------------------

### Custom Async Flow Creation

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/README.md

Demonstrates how to create a new asynchronous flow in PocketFlow using Python. This involves instantiating nodes, defining their connections, and returning a configured 'AsyncFlow' object.

```python
def create_my_custom_flow():
    # Create and connect nodes
    node1 = MyCustomNode()
    node2 = AnotherNode()
    node1 >> node2
    
    # Return configured flow
    return AsyncFlow(start=node1)
```

--------------------------------

### Configure Langfuse Credentials

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Sets up environment variables for Langfuse integration by copying an example file and editing it with user-specific credentials and host information.

```bash
cp .env.example .env
```

```env
LANGFUSE_SECRET_KEY=your-langfuse-secret-key
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key
LANGFUSE_HOST=your-langfuse-host-url
POCKETFLOW_TRACING_DEBUG=true
```

--------------------------------

### Custom Tracer Configuration from Environment

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Example of creating a custom TracingConfig by loading from environment variables and then modifying specific settings like debug mode.

```python
from tracing import TracingConfig, LangfuseTracer

# Create custom configuration
config = TracingConfig.from_env()
config.debug = True

# Use tracer directly (for advanced use cases)
tracer = LangfuseTracer(config)
```

--------------------------------

### Python Shared Store Design Example

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/guide_for_pocketflow.md

Demonstrates a basic in-memory shared store design using nested dictionaries for storing user context and results in PocketFlow. This serves as a foundation for data management within the system.

```Python
shared = {
    "user": {
        "id": "user123",
        "context": {                # Another nested dict
            "weather": {"temp": 72, "condition": "sunny"},
            "location": "San Francisco"
        }
    },
    "results": {}                   # Empty dict to store outputs
}
```

--------------------------------

### Create Question-Answering Flow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/guide_for_pocketflow.md

Implements a function to create a simple question-answering flow by defining and connecting the GetQuestionNode and AnswerNode. The flow starts with the GetQuestionNode.

```Python
from pocketflow import Flow
from nodes import GetQuestionNode, AnswerNode

def create_qa_flow():
    """Create and return a question-answering flow."""
    # Create nodes
    get_question_node = GetQuestionNode()
    answer_node = AnswerNode()
    
    # Connect nodes in sequence
    get_question_node >> answer_node
    
    # Create flow starting with input node
    return Flow(start=get_question_node)
```

--------------------------------

### Python Flow Creation Functions

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Flow creation functions were organized to support various asynchronous patterns, including concurrent processing, performance monitoring, and error handling demonstrations.

```python
# create_comprehensive_async_flow(): Main flow creation function
# create_concurrent_data_flow(): Concurrent processing flow
# create_performance_monitoring_flow(): Performance tracking flow
# create_simple_async_flow(): Basic testing flow
# create_fallback_demonstration_flow(): Error handling demonstration flow
```

--------------------------------

### Migrate Synchronous Node to AsyncNode

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Converts a standard synchronous PocketFlow Node to an asynchronous AsyncNode. This involves renaming methods like `prep` to `prep_async`, `exec` to `exec_async`, and `post` to `post_async`, and marking them as `async`.

```python
from pocketflow import Node

class DataProcessingNode(Node):
    def prep(self, shared):
        return shared.get("input_data")
    
    def exec(self, prep_res):
        # Synchronous processing
        return process_data(prep_res)
    
    def post(self, shared, prep_res, exec_res):
        shared["output"] = exec_res
        return "default"
```

```python
from pocketflow import AsyncNode

class DataProcessingNode(AsyncNode):
    async def prep_async(self, shared):
        return shared.get("input_data")
    
    async def exec_async(self, prep_res):
        # Asynchronous processing with proper tracing
        return await async_process_data(prep_res)
    
    async def post_async(self, shared, prep_res, exec_res):
        shared["output"] = exec_res
        return "default"
```

--------------------------------

### Execute PocketFlows Concurrently

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Demonstrates how to run multiple PocketFlow instances concurrently using `asyncio.gather`. This is useful for parallel processing of independent tasks, with each flow benefiting from async tracing.

```python
import asyncio
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="ConcurrentFlow")
class ConcurrentFlow(AsyncFlow):
    # ... flow definition ...

async def run_concurrent_flows():
    flows = [ConcurrentFlow() for _ in range(5)]
    shared_data = [{"input": f"data_{i}"} for i in range(5)]
    
    # Execute flows concurrently with proper tracing
    results = await asyncio.gather(*[
        flow.run_async(shared) 
        for flow, shared in zip(flows, shared_data)
    ])
    
    return results
```

--------------------------------

### Concurrent Flow Execution with Tracing

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Example of tracing multiple asynchronous flows concurrently using asyncio.gather, with each flow having a name.

```python
import asyncio
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="ConcurrentFlow")
class ConcurrentFlow(AsyncFlow):
    # ... your async flow definition ...
    pass

async def run_concurrent_flows():
    flows = [ConcurrentFlow() for _ in range(5)]
    shared_data = [{"input": f"data_{i}"} for i in range(5)]

    # Execute flows concurrently with proper tracing
    results = await asyncio.gather(*[
        flow.run_async(shared)
        for flow, shared in zip(flows, shared_data)
    ])

    return results
```

--------------------------------

### Migrate Synchronous Flow to AsyncFlow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Transforms a synchronous PocketFlow Flow into an asynchronous AsyncFlow. This includes inheriting from `AsyncFlow` and optionally implementing `prep_async` and `post_async` methods for flow-level async operations.

```python
from pocketflow import Flow
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="DataProcessingFlow")
class DataProcessingFlow(Flow):
    def __init__(self):
        node1 = DataProcessingNode()
        node2 = OutputNode()
        node1.next(node2)
        super().__init__(start=node1)
```

```python
from pocketflow import AsyncFlow
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="DataProcessingFlow")
class DataProcessingFlow(AsyncFlow):
    def __init__(self):
        node1 = DataProcessingNode()  # Now async
        node2 = OutputNode()          # Now async
        node1.next(node2)
        super().__init__(start=node1)
    
    async def prep_async(self, shared):
        """Optional: Flow-level async preparation"""
        return await super().prep_async(shared)
    
    async def post_async(self, shared, prep_res, exec_res):
        """Optional: Flow-level async post-processing"""
        return await super().post_async(shared, prep_res, exec_res)
```

--------------------------------

### Manage Async Tracing Context with AsyncTraceContext

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Demonstrates how to use AsyncTraceContext to manage tracing context across asynchronous boundaries. It shows creation, usage as an async context manager, and retrieval of the current context.

```Python
from pocketflow_tracing.core import AsyncTraceContext

# Create async context
context = AsyncTraceContext(tracer, trace_id, flow_name)

# Use as async context manager
async with context:
    # Your async operations here
    pass

# Get current context
current = AsyncTraceContext.get_current_context()
```

--------------------------------

### Write Async Flow Unit Tests

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code provides a template for unit testing asynchronous flows using `pytest` and `asyncio`. It includes a test function `test_async_flow` that runs an `YourAsyncFlow` instance and asserts the expected outcome and shared data.

```python
import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_flow():
    flow = YourAsyncFlow()
    shared = {"test_input": "data"}
    
    result = await flow.run_async(shared)
    
    assert result == "expected_result"
    assert "expected_output" in shared
```

--------------------------------

### Python Utility Functions for Async Operations

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Utility functions were created to handle asynchronous data fetching, processing, and concurrent execution. These utilities are designed for modularity and reusability.

```python
# async_data_fetch.py:
# fetch_data_from_api(): Async data fetching with timeout
# get_fallback_data(): Fallback data generation
# validate_query_params(): Parameter validation

# async_data_process.py:
# process_data_advanced(): Advanced processing strategy
# process_data_simple(): Simple processing strategy
# determine_processing_strategy(): Strategy selection logic

# concurrent_utils.py:
# process_items_concurrently(): Concurrent item processing
# execute_flows_concurrently(): Concurrent flow execution
# measure_performance(): Performance measurement utilities
```

--------------------------------

### Compose Nested Async Flows

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Illustrates how to create nested asynchronous flows where one flow can trigger and await the execution of another `AsyncFlow`. This is useful for structuring complex workflows with modular sub-processes.

```python
@trace_flow(flow_name="MainFlow")
class MainFlow(AsyncFlow):
    async def run_async(self, shared):
        # Run main processing
        result = await super().run_async(shared)
        
        # Run nested analysis flow
        if result == "success":
            analysis_flow = AnalysisFlow()
            analysis_result = await analysis_flow.run_async(shared)
            shared["analysis"] = analysis_result
        
        return result
```

--------------------------------

### Update Python Imports for Comprehensive Async Flow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Demonstrates the changes required for importing the ComprehensiveAsyncFlow in Python. The old import path is replaced with a new structure that uses a dedicated flow module and a factory function.

```python
# Old
from examples.comprehensive_async_example import ComprehensiveAsyncFlow

# New
from examples.comprehensive_async_example.flow import create_comprehensive_async_flow
flow = create_comprehensive_async_flow()
```

--------------------------------

### Handle Async Errors with AsyncErrorHandler

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Shows how to use AsyncErrorHandler for comprehensive async error handling. It includes handling errors, registering custom error handlers for specific error types like timeouts.

```Python
from pocketflow_tracing.async_error_handling import AsyncErrorHandler, AsyncErrorType

handler = AsyncErrorHandler(tracer)

# Handle async error
error_event = await handler.handle_async_error(
    error, node_id="node_1", flow_name="TestFlow"
)

# Register custom error handler
handler.register_error_handler(AsyncErrorType.TIMEOUT, custom_handler)
```

--------------------------------

### Migrate Synchronous Execution to Async Execution

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Updates the execution of a PocketFlow to use the asynchronous `run_async` method. This requires running the flow within an asyncio event loop, typically using `asyncio.run()`.

```python
flow = DataProcessingFlow()
shared = {"input_data": "test"}
result = flow.run(shared)
```

```python
import asyncio

async def main():
    flow = DataProcessingFlow()
    shared = {"input_data": "test"}
    result = await flow.run_async(shared)
    return result

# Run the async flow
result = asyncio.run(main())
```

--------------------------------

### Python Utility Function to Call LLM

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/guide_for_pocketflow.md

A Python function that utilizes the OpenAI library to interact with a language model. It takes a prompt as input and returns the model's response. Includes an example of how to use the function.

```Python
# utils/call_llm.py
from openai import OpenAI

def call_llm(prompt):
    client = OpenAI(api_key="YOUR_API_KEY_HERE")
    r = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return r.choices[0].message.content
    
if __name__ == "__main__":
    prompt = "What is the meaning of life?"
    print(call_llm(prompt))
```

--------------------------------

### LangfuseTracer: Async Error Handling

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Details enhanced LangfuseTracer methods for async error handling, including handling general async errors, safe async execution with retries and timeouts, and executing with specific timeouts.

```Python
# Handle async error
error_event = await tracer.handle_async_error(
    error, node_id, flow_name, context_data
)

# Safe async execution
result = await tracer.safe_async_execute(
    coro, node_id, flow_name, timeout_seconds=10, retry_count=3
)

# Execute with timeout
result = await tracer.with_timeout(coro, timeout_seconds, node_id, flow_name)
```

--------------------------------

### Python Project Structure Reorganization

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

The project structure was refactored from a single monolithic file to a modular structure with dedicated directories for main logic, nodes, flows, and utilities. This improves maintainability and organization.

```text
Before:
examples/
└── comprehensive_async_example.py  (466 lines, everything in one file)

After:
examples/comprehensive_async_example/
├── main.py                    # Entry point and demonstrations (300 lines)
├── nodes.py                   # Node class definitions (300 lines)
├── flow.py                    # Flow creation functions (300 lines)
├── utils/                     # Utility functions directory
│   ├── __init__.py
│   ├── async_data_fetch.py    # Data fetching utilities (130 lines)
│   ├── async_data_process.py  # Data processing utilities (150 lines)
│   └── concurrent_utils.py    # Concurrent processing utilities (180 lines)
└── requirements.txt           # Project dependencies
```

--------------------------------

### Trace an Asynchronous Flow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Illustrates tracing an asynchronous PocketFlow using the `@trace_flow` decorator. This example involves fetching and processing data asynchronously, showing the decorator's application to `AsyncFlow` classes.

```python
from pocketflow import AsyncNode, AsyncFlow
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="AsyncDataProcessingFlow")
class AsyncDataProcessingFlow(AsyncFlow):
    def __init__(self):
        fetch_node = AsyncDataFetchNode()
        process_node = AsyncDataProcessNode()
        fetch_node - "process" >> process_node
        super().__init__(start=fetch_node)

# Usage
flow = AsyncDataProcessingFlow()
shared = {"query": "machine learning tutorials"}
result = await flow.run_async(shared)
```

--------------------------------

### Python Node Class Refactoring

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/changelog/20250712-comprehensive-async-example-refactor.md

Node classes were refactored to utilize new utility functions for improved clarity and functionality. New nodes for concurrent processing and performance monitoring were introduced.

```python
# AsyncDataFetchNode: Refactored to use utility functions from utils.async_data_fetch
# AsyncDataProcessNode: Simplified using utils.async_data_process utilities
# AsyncFallbackProcessNode: Streamlined fallback handling
# AsyncConcurrentProcessNode: New node for concurrent processing demonstrations
# AsyncPerformanceMonitorNode: New node for performance monitoring
```

--------------------------------

### Write Async Concurrent Performance Tests

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code demonstrates how to write performance tests for concurrent asynchronous operations using `pytest`, `asyncio`, and `time`. It measures the execution time of multiple `YourAsyncFlow` instances running concurrently via `asyncio.gather`.

```python
import time

@pytest.mark.asyncio
async def test_concurrent_performance():
    flows = [YourAsyncFlow() for _ in range(10)]
    shared_data = [{"input": f"data_{i}"} for i in range(10)]
    
    start_time = time.time()
    results = await asyncio.gather(*[
        flow.run_async(shared) 
        for flow, shared in zip(flows, shared_data)
    ])
    execution_time = time.time() - start_time
    
    # Should be faster than sequential execution
    assert execution_time < 5.0  # Adjust based on your requirements
    assert len(results) == 10
```

--------------------------------

### Implement Error Handling and Retries in AsyncNode

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

Shows how to implement robust error handling within an `AsyncNode` using `try-except` blocks and defining a `exec_fallback_async` method. The node is configured with automatic retries and a wait time between attempts.

```python
class RobustAsyncNode(AsyncNode):
    def __init__(self, max_retries=3):
        super().__init__(max_retries=max_retries, wait=1.0)
    
    async def exec_async(self, prep_res):
        try:
            return await risky_async_operation(prep_res)
        except ConnectionError as e:
            # This will trigger automatic retry
            raise
    
    async def exec_fallback_async(self, prep_res, exc):
        """Fallback when all retries are exhausted"""
        return await safe_fallback_operation(prep_res)
```

--------------------------------

### Migrate Batch Processing to Async

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code shows the migration of a synchronous batch processing function to an asynchronous one using `asyncio.gather`. The 'Before' version processes items sequentially, while the 'After' version processes items concurrently for improved performance.

```python
# Before
def process_batch(items):
    results = []
    for item in items:
        result = process_item(item)
        results.append(result)
    return results

# After
async def process_batch_async(items):
    # Process items concurrently
    results = await asyncio.gather(*[
        process_item_async(item) for item in items
    ])
    return results
```

--------------------------------

### Monitor Async Flow Performance with PocketFlow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code demonstrates how to use the `@trace_flow` decorator to monitor the performance of an asynchronous flow. It logs the execution duration and warns if it exceeds a threshold, utilizing `time.time()` for measurements and custom logging.

```python
import time

@trace_flow(flow_name="MonitoredFlow")
class MonitoredFlow(AsyncFlow):
    async def prep_async(self, shared):
        shared["start_time"] = time.time()
        return await super().prep_async(shared)
    
    async def post_async(self, shared, prep_res, exec_res):
        duration = time.time() - shared["start_time"]
        shared["execution_time"] = duration
        
        # Log performance metrics
        if duration > 5.0:
            print(f"⚠️ Slow execution detected: {duration:.2f}s")
        
        return await super().post_async(shared, prep_res, exec_res)
```

--------------------------------

### Configure Enhanced Tracing for Async Operations

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python snippet shows how to configure PocketFlow's tracing for asynchronous operations using `TracingConfig`. It enables various async-specific tracing options like lifecycle tracking, context propagation, concurrent flow monitoring, error recovery, and performance monitoring.

```python
from pocketflow_tracing import TracingConfig

config = TracingConfig(
    # Existing options
    trace_inputs=True,
    trace_outputs=True,
    trace_errors=True,
    
    # New async-specific options
    trace_async_lifecycle=True,      # Track async node lifecycle events
    trace_async_context=True,        # Track async context propagation
    trace_concurrent_flows=True,     # Track concurrent flow execution
    async_error_recovery=True,       # Enable async error recovery
    performance_monitoring=True,     # Enable performance monitoring
)
```

--------------------------------

### Troubleshoot Mixed Sync/Async Deadlocks

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code addresses the problem of deadlocks caused by calling synchronous blocking functions from an asynchronous context. The 'Problem' shows `problematic_async` calling `sync_blocking_function`, while the 'Solution' demonstrates using `fixed_async` with `async_non_blocking_function`.

```python
# Problem: Calling sync code from async context
async def problematic_async():
    return sync_blocking_function()  # Can cause deadlocks

# Solution: Use proper async alternatives
async def fixed_async():
    return await async_non_blocking_function()
```

--------------------------------

### Troubleshoot Async Context Loss

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code snippet illustrates a common issue in asynchronous programming: context loss across async boundaries. The 'Problem' shows a `problematic_function` where context might be lost, while the 'Solution' highlights using proper async context management for correct propagation.

```python
# Problem: Context not propagated
async def problematic_function():
    # Context might be lost here
    pass

# Solution: Use proper async context management
async def fixed_function():
    # Context is automatically managed by the tracing system
    pass
```

--------------------------------

### Python Async Node Tracing Implementation

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/examples/comprehensive_async_example/design-async.md

This snippet outlines the implementation of asynchronous node tracing within the pocketflow system. It focuses on extending existing node tracing capabilities to manage async function calls, capture lifecycle events (start, suspend, resume, complete), maintain parent-child relationships for async operations, and handle concurrent async node execution.

```Python
class AsyncNode:
    def __init__(self, node_id, flow_id):
        self.node_id = node_id
        self.flow_id = flow_id
        self.start_time = None
        self.end_time = None
        self.status = 'pending'
        self.children = []
        self.parent = None

    async def execute(self):
        self.start_time = time.time()
        self.status = 'running'
        # Simulate async operation
        await asyncio.sleep(1)
        self.end_time = time.time()
        self.status = 'completed'

    def add_child(self, child_node):
        self.children.append(child_node)
        child_node.parent = self

    def trace_event(self, event_type):
        # Logic to record tracing events (start, suspend, resume, complete)
        print(f"Node {self.node_id} event: {event_type}")

# Example usage within a flow
async def run_async_flow():
    flow_id = 'flow_1'
    node1 = AsyncNode('node_1', flow_id)
    node2 = AsyncNode('node_2', flow_id)

    node1.add_child(node2)

    await node1.execute()
    await node2.execute()

    # Tracing flow execution
    print(f"Flow {flow_id} execution completed.")

```

--------------------------------

### Main Entry Point for PocketFlow Application

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/guide_for_pocketflow.md

Serves as the main entry point for the project. It initializes a shared context, creates a question-answering flow using `create_qa_flow()`, runs the flow, and then prints the captured question and the LLM-generated answer.

```Python
from flow import create_qa_flow

# Example main function
# Please replace this with your own main function
def main():
    shared = {
        "question": None,  # Will be populated by GetQuestionNode from user input
        "answer": None     # Will be populated by AnswerNode
    }

    # Create the flow and run it
    qa_flow = create_qa_flow()
    qa_flow.run(shared)
    print(f"Question: {shared['question']}")
    print(f"Answer: {shared['answer']}")

if __name__ == "__main__":
    main()
```

--------------------------------

### Trace a Basic Synchronous Flow

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Demonstrates how to use the `@trace_flow` decorator to add tracing to a synchronous PocketFlow. It defines a simple flow with greeting and uppercase nodes and shows how to run it.

```python
from pocketflow import Node, Flow
from pocketflow_tracing import trace_flow

@trace_flow(flow_name="BasicGreetingFlow")
class BasicGreetingFlow(Flow):
    def __init__(self):
        greeting_node = GreetingNode()
        uppercase_node = UppercaseNode()
        greeting_node >> uppercase_node
        super().__init__(start=greeting_node)

# Usage
flow = BasicGreetingFlow()
shared = {"name": "PocketFlow User"}
result = flow.run(shared)
```

--------------------------------

### Trace Flow with Custom Tracing Configuration

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Demonstrates creating a custom TracingConfig object with specific parameters like Langfuse keys, host, and debugging options, then applying it to a flow.

```python
from tracing import TracingConfig

config = TracingConfig(
    langfuse_secret_key="your-secret-key",
    langfuse_public_key="your-public-key", 
    langfuse_host="https://your-langfuse-instance.com",
    debug=True,
    trace_inputs=True,
    trace_outputs=True,
    trace_errors=True
)

@trace_flow(config=config)
class MyFlow(Flow):
    pass
```

--------------------------------

### Define Project Dependencies

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Specifies the project's dependencies using TOML format, including the Langfuse SDK, python-dotenv for environment variable management, and Pydantic for data validation. Version constraints are also provided.

```toml
dependencies = [
    "langfuse>=2.0.0,<3.0.0",  # v2 SDK as preferred
    "python-dotenv>=1.0.0",   # Environment variable loading
    "pydantic>=2.0.0"         # Data validation
]
```

--------------------------------

### Trace Flow with Default Environment Variables

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Demonstrates how to use the trace_flow decorator with default environment variable configurations for tracing a flow.

```python
from tracing import trace_flow

@trace_flow()
class MyFlow(Flow):
    pass
```

--------------------------------

### Configure Langfuse for Tracing

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Sets up essential environment variables for Langfuse integration, including secret key, public key, and host URL. These are required for the tracing functionality to send data to Langfuse.

```env
LANGFUSE_SECRET_KEY=your-langfuse-secret-key
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key
LANGFUSE_HOST=your-langfuse-host-url
```

--------------------------------

### PocketFlow Tracing Environment Variables

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Lists and explains the environment variables used to configure PocketFlow tracing, including required Langfuse settings and optional tracing parameters.

```env
# Required Langfuse configuration
LANGFUSE_SECRET_KEY=your-secret-key
LANGFUSE_PUBLIC_KEY=your-public-key
LANGFUSE_HOST=your-langfuse-host

# Optional tracing configuration
POCKETFLOW_TRACING_DEBUG=true
POCKETFLOW_TRACE_INPUTS=true
POCKETFLOW_TRACE_OUTPUTS=true
POCKETFLOW_TRACE_PREP=true
POCKETFLOW_TRACE_EXEC=true
POCKETFLOW_TRACE_POST=true
POCKETFLOW_TRACE_ERRORS=true

# Optional session/user tracking
POCKETFLOW_SESSION_ID=your-session-id
POCKETFLOW_USER_ID=your-user-id
```

--------------------------------

### Langfuse Configuration Environment Variables

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Specifies the required environment variables for configuring the Langfuse tracing backend. These include Langfuse API keys, the Langfuse host address, and a debug flag for tracing.

```Shell
# Required environment variables
LANGFUSE_SECRET_KEY=your-secret-key
LANGFUSE_PUBLIC_KEY=your-public-key
LANGFUSE_HOST=http://192.168.1.216:3000  # User's preferred host
POCKETFLOW_TRACING_DEBUG=true
```

--------------------------------

### Troubleshoot Async Flow Memory Leaks

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/docs/async_migration_guide.md

This Python code addresses memory leaks in long-running asynchronous flows due to unmanaged resources. The 'Problem' shows `problematic_flow` without resource cleanup, while the 'Solution' uses `fixed_flow` with `async with resource_manager()` for automatic resource management.

```python
# Problem: Not cleaning up resources
async def problematic_flow():
    # Resources not cleaned up
    pass

# Solution: Use proper resource management
async def fixed_flow():
    async with resource_manager():
        # Resources automatically cleaned up
        pass
```

--------------------------------

### Nested Flow Composition with Tracing

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/README.md

Shows how to compose flows, where a main flow executes a nested analysis flow, ensuring proper context propagation for tracing.

```python
@trace_flow(flow_name="MainFlow")
class MainFlow(AsyncFlow):
    async def run_async(self, shared):
        # Run main processing
        result = await super().run_async(shared)

        # Run nested analysis flow with proper context
        if result == "success":
            analysis_flow = AnalysisFlow()
            analysis_result = await analysis_flow.run_async(shared)
            shared["analysis"] = analysis_result

        return result
```

--------------------------------

### Configure Tracing Behavior

Source: https://github.com/redreamality/pocketflow-tracing/blob/main/tracing-in-depth.md

Defines optional environment variables to control the behavior of PocketFlow tracing, such as enabling debug mode and specifying which parts of the flow (inputs, outputs, preparation, execution, post-processing, errors) should be traced.

```env
POCKETFLOW_TRACING_DEBUG=true
POCKETFLOW_TRACE_INPUTS=true
POCKETFLOW_TRACE_OUTPUTS=true
POCKETFLOW_TRACE_PREP=true
POCKETFLOW_TRACE_EXEC=true
POCKETFLOW_TRACE_POST=true
POCKETFLOW_TRACE_ERRORS=true
```