Pipecat Flows Documentation

To run the Pipecat Flows examples, you need Python 3.11+ and the uv package manager.

1. Install dependencies: Run uv sync to install the package, then install pipecat-ai with the necessary extras.
2. Install provider-specific extras: Depending on which LLM you want to use, you must install the corresponding extra (e.g., google, anthropic, or aws).
3. Configure environment: Copy env.example to .env and populate it with your API keys.
4. Run: Use uv run to execute an example script.

# 1. Install package and base extras
uv sync
uv pip install "pipecat-ai[daily,openai,deepgram,cartesia,silero,examples]"

# 2. Install specific provider extras (choose one or more)
# For Google Gemini
uv pip install "pipecat-ai[daily,google,deepgram,cartesia,silero,examples]"
# For Anthropic
uv pip install "pipecat-ai[daily,anthropic,deepgram,cartesia,silero,examples]"
# For AWS Bedrock
uv pip install "pipecat-ai[daily,aws,deepgram,cartesia,silero,examples]"

# 3. Setup configuration
cp env.example .env

# 4. Run an example
uv run examples/food_ordering.py

Before running evals, ensure the following dependencies and configurations are met:

1. Development Dependencies:

   uv sync --group dev

2. Judge LLM (Ollama): The default harness uses a local Ollama instance with the gemma2:9b model to score eval: criteria:

   ollama pull gemma2:9b

   Note: You can use OpenAI as a judge by adding an eval: block to the scenario's judge: config.

3. API Keys: Ensure examples/.env contains the necessary keys for the bots being tested:

   • OPENAI_API_KEY (default provider)
   • GOOGLE_API_KEY (required for hello_world)
   • ANTHROPIC_API_KEY and GOOGLE_API_KEY (required for llm_switching)
   • CARTESIA_API_KEY and DEEPGRAM_API_KEY (required for Audio mode)

4. Audio Models: The first time you run an Audio mode eval, the system will download the local Kokoro and Moonshine models.

To add a new behavioral test scenario, follow these steps:

1. Create the Scenario File: Create a new .yaml file in evals/scenarios/. It is recommended to base it on an existing scenario.
2. Define Turns and Assertions:
   • Use user: turns to drive the flow.
   • Use expect: to assert behavior.
   • Function Calling: Assert the function_call to verify the correct Flows handler fired with the expected arguments.
   • Semantic Response: Use a response eval for a lenient, semantic check of the bot's reply. This also acts as a pace-maker, ensuring the harness waits for the bot to finish before the next turn.
   • Special Case (Terminal Turns): For turns that end the conversation (e.g., end_conversation), assert only the function_call. The end_conversation action tears down the pipeline immediately, so the farewell message may not be delivered to the harness.
3. Register the Scenario: Add the new scenario under its corresponding bot in evals/manifest.yaml.
4. Enable Eval Transport (If needed): If the bot is new, ensure it is eval-capable by adding an "eval" entry to its transport_params. PipelineWorker will handle the RTVI wiring automatically.

Pipecat Flows provides a suite of behavioral evaluations to verify that example bots correctly execute Flows/LLM functions, handle arguments, and respond appropriately. By default, these run in text-only mode, which is fast and deterministic, focusing on node transitions and function calling without the variance of audio I/O.

Running the full suite

To run all scenarios defined in the manifest (the recommended 'release gate' check), run the following from the repository root:

uv run pipecat eval suite evals/manifest.yaml

Iterating on a single scenario

For faster development, you can run a single bot in headless mode and drive a specific scenario against it in a separate terminal:

Terminal 1: Start the bot

uv run examples/food_ordering.py -t eval

Terminal 2: Run the scenario (verbose mode)

uv run pipecat eval run evals/scenarios/food_ordering_pizza.yaml -v

Configuration and Debugging

• Concurrency: You can tune the number of simultaneous scenarios by adjusting the concurrency: key in evals/manifest.yaml.
• Logs: If a test fails, inspect the logs located in evals/eval-runs/<timestamp>/logs/<scenario>.eval.log.
• Audio Recording: Add the -a flag to record conversation audio (primarily useful for audio-mode runs).

The ActionManager class is responsible for executing side effects during conversation state transitions. It handles built-in actions like text-to-speech (TTS) and ending conversations, as well as custom user-defined actions. Actions are triggered via execute_actions() and can be scheduled to run after an LLM response using schedule_deferred_post_actions().

Built-in Action Types:

• tts_say: Speaks text using the pipeline's TTS node.
• end_conversation: Terminates the conversation (optionally after a goodbye message).
• function: Executes an inline function within the pipeline at the appropriate time.

from pipecat_flows.manager import FlowManager
from pipecat.pipeline.worker import PipelineWorker
from pipecat_flows.actions import ActionManager

# Assuming worker and flow_manager are already initialized
action_manager = ActionManager(worker, flow_manager)

# Example: Executing a list of actions
await action_manager.execute_actions([
    {"type": "tts_say", "text": "Hello, how can I help you?"},
    {"type": "function", "handler": my_custom_handler_func, "param": "value"}
])

A FlowsDirectFunction is an async function that allows Pipecat Flows to automatically extract its schema from the function signature and docstring. This is a convenient alternative to manually defining a FlowsFunctionSchema.

Required Signature:

• The first parameter must be named flow_manager and be of type FlowManager.
• It should return a ConsolidatedFunctionResult (a tuple of (result, next_node_config)).
• Parameters should be described in the docstring for automatic schema generation.

Customizing Call Options: Use the @flows_tool_options decorator to override default behavior for interruption and timeouts.

Example:

@flows_tool_options(cancel_on_interruption=False, timeout_secs=30)
async def long_running_task(flow_manager: FlowManager, query: str) -> ConsolidatedFunctionResult:
    """Performs a long-running task."""
    # ... implementation
    return {"status": "complete"}, None

@flows_tool_options(cancel_on_interruption=False, timeout_secs=30)
async def long_running_task(flow_manager: FlowManager, query: str) -> ConsolidatedFunctionResult:
    """Performs a long-running task that should not be cancelled on interruption."""
    # ... implementation
    return {"status": "complete"}, None