Pipecat Flows provides a framework for building structured conversations in your AI applications. It enables you to create both predefined conversation paths and dynamically generated flows while handling the complexities of state management and LLM interactions.
The framework consists of:
If you're already using Pipecat:
pip install pipecat-ai-flows
If you're starting fresh:
# Basic installation
pip install pipecat-ai-flows
# Install Pipecat with specific LLM provider options:
pip install "pipecat-ai[daily,openai,deepgram,cartesia]" # For OpenAI
pip install "pipecat-ai[daily,anthropic,deepgram,cartesia]" # For Anthropic
pip install "pipecat-ai[daily,google,deepgram,cartesia]" # For Google
Here's a basic example of setting up a conversation flow:
from pipecat_flows import FlowManager
# Initialize flow manager with static configuration
flow_manager = FlowManager(task, llm, tts, flow_config=flow_config)
# Or with dynamic flow handling
flow_manager = FlowManager(
task,
llm,
tts,
transition_callback=handle_transitions
)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
await flow_manager.initialize(messages)
await task.queue_frames([context_aggregator.user().get_context_frame()])
For more detailed examples and guides, visit our documentation.
Each conversation flow consists of nodes that define the conversation structure. A node includes:
Messages set the context for the LLM at each state:
"messages": [
{
"role": "system",
"content": "You are handling pizza orders. Ask for size selection."
}
]
Functions come in two types:
{
"type": "function",
"function": {
"name": "select_size",
"handler": select_size_handler,
"description": "Select pizza size",
"parameters": {
"type": "object",
"properties": {
"size": {"type": "string", "enum": ["small", "medium", "large"]}
}
},
"transition_to": "next_node" # Optional: Specify next node
}
}
{
"type": "function",
"function": {
"name": "next_step",
"description": "Move to next state",
"parameters": {"type": "object", "properties": {}},
"transition_to": "target_node" # Required: Specify target node
}
}
Functions can:
Actions execute during state transitions:
"pre_actions": [
{
"type": "tts_say",
"text": "Processing your order..."
}
]
Pipecat Flows automatically handles format differences between LLM providers:
OpenAI Format
"functions": [{
"type": "function",
"function": {
"name": "function_name",
"description": "description",
"parameters": {...}
}
}]
Anthropic Format
"functions": [{
"name": "function_name",
"description": "description",
"input_schema": {...}
}]
Google (Gemini) Format
"functions": [{
"function_declarations": [{
"name": "function_name",
"description": "description",
"parameters": {...}
}]
}]
The FlowManager handles both static and dynamic flows through a unified interface:
# Define flow configuration upfront
flow_config = {
"initial_node": "greeting",
"nodes": {
"greeting": {
"messages": [...],
"functions": [{
"type": "function",
"function": {
"name": "collect_name",
"description": "Record user's name",
"parameters": {...},
"handler": collect_name_handler, # Specify handler
"transition_to": "next_step" # Specify transition
}
}]
}
}
}
# Initialize with static configuration
flow_manager = FlowManager(task, llm, tts, flow_config=flow_config)
# Define transition handling
async def handle_transitions(function_name: str, args: Dict, flow_manager):
if function_name == "collect_age":
await flow_manager.set_node("next_step", create_next_node())
# Initialize with transition callback
flow_manager = FlowManager(task, llm, tts, transition_callback=handle_transitions)
The repository includes several complete example implementations in the examples/ directory.
In the examples/static directory, you'll find these examples:
food_ordering.py - A restaurant order flow demonstrating node and edge functionsmovie_explorer_openai.py - Movie information bot demonstrating real API integration with TMDBmovie_explorer_anthropic.py - The same movie information demo adapted for Anthropic's formatmovie_explorer_gemini.py - The same movie explorer demo adapted for Google Gemini's formatpatient_intake.py - A medical intake system showing complex state managementrestaurant_reservation.py - A reservation system with availability checkingtravel_planner.py - A vacation planning assistant with parallel pathsIn the examples/dynamic directory, you'll find these examples:
insurance_openai.py - An insurance quote system using OpenAI's formatinsurance_anthropic.py - The same insurance system adapted for Anthropic's formatinsurance_gemini.py - The insurance system implemented with Google's formatEach LLM provider (OpenAI, Anthropic, Google) has slightly different function calling formats, but Pipecat Flows handles these differences internally while maintaining a consistent API for developers.
To run these examples:
Setup Virtual Environment (recommended):
python3 -m venv venv
source venv/bin/activate
Installation:
Install the package in development mode:
pip install -e .
Install Pipecat with required options for examples:
pip install "pipecat-ai[daily,openai,deepgram,cartesia,silero,examples]"
If you're running Google or Anthropic examples, you will need to update the installed options. For example:
# Install Google Gemini
pip install "pipecat-ai[daily,google,deepgram,cartesia,silero,examples]"
# Install Anthropic
pip install "pipecat-ai[daily,anthropic,deepgram,cartesia,silero,examples]"
Configuration:
Copy env.example to .env in the examples directory:
cp env.example .env
Add your API keys and configuration:
Looking for a Daily API key and room URL? Sign up on the Daily Dashboard.
Running:
python examples/static/food_ordering.py -u YOUR_DAILY_ROOM_URL
The package includes a comprehensive test suite covering the core functionality.
Create Virtual Environment:
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
Install Test Dependencies:
pip install -r dev-requirements.txt -r test-requirements.txt
pip install "pipecat-ai[google,openai,anthropic]"
pip install -e .
Run all tests:
pytest tests/
Run specific test file:
pytest tests/test_state.py
Run specific test:
pytest tests/test_state.py -k test_initialization
Run with coverage report:
pytest tests/ --cov=pipecat_flows
A visual editor for creating and managing Pipecat conversation flows.
While the underlying system is flexible with node naming, the editor follows these conventions for clarity:
These conventions help maintain readable and maintainable flows while preserving technical flexibility.
The editor is available online at flows.pipecat.ai.
Clone the repository
git clone git@github.com:pipecat-ai/pipecat-flows.git
Navigate to project directory
cd pipecat-flows/editor
Install dependencies
npm install
Start development server
npm run dev
Open the page in your browser: http://localhost:5173.
The editor/examples/ directory contains sample flow configurations:
food_ordering.jsonmovie_explorer.pypatient_intake.jsonrestaurant_reservation.jsontravel_planner.jsonTo use an example:
See the examples directory for the complete files and documentation.
npm start - Start production servernpm run dev - Start development servernpm run build - Build for productionnpm run preview - Preview production build locallynpm run preview:prod - Preview production build with base pathnpm run lint - Check for linting issuesnpm run lint:fix - Fix linting issuesnpm run format - Format code with Prettiernpm run format:check - Check code formattingnpm run docs - Generate documentationnpm run docs:serve - Serve documentation locallyThe Pipecat Flows Editor project uses JSDoc for documentation. To generate and view the documentation:
Generate documentation:
npm run docs
Serve documentation locally:
npm run docs:serve
View in browser by opening: http://localhost:8080
We welcome contributions from the community! Whether you're fixing bugs, improving documentation, or adding new features, here's how you can help:
Before submitting a pull request, please check existing issues and PRs to avoid duplicates.
We aim to review all contributions promptly and provide constructive feedback to help get your changes merged.
FAQs
Conversation Flow management for Pipecat AI applications
We found that pipecat-ai-flows demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.