{"id":10041,"library":"pipecat-ai-flows","title":"Pipecat AI Flows","description":"Pipecat AI Flows provides a powerful conversation flow management system for Pipecat AI applications. It allows developers to define structured conversational experiences using dynamic nodes and functions, managing transitions and LLM interactions. The library is currently at version 1.0.0 and follows an active release cadence, with frequent updates preceding major version releases.","status":"active","version":"1.0.0","language":"en","source_language":"en","source_url":"https://github.com/pipecat-ai/pipecat-flows","tags":["ai","llm","conversational-ai","chatbot","flow-management","pipecat"],"install":[{"cmd":"pip install pipecat-ai-flows","lang":"bash","label":"Install latest version"}],"dependencies":[{"reason":"Core dependency for Pipecat AI functionalities, required version >=1.0.0","package":"pipecat-ai","optional":false},{"reason":"Requires Python 3.11 or higher","package":"python","optional":false}],"imports":[{"symbol":"FlowManager","correct":"from pipecat_ai_flows import FlowManager"},{"symbol":"NodeConfig","correct":"from pipecat_ai_flows import NodeConfig"},{"symbol":"flows_direct_function","correct":"from pipecat_ai_flows import flows_direct_function"},{"symbol":"ActionConfig","correct":"from pipecat_ai_flows import ActionConfig"}],"quickstart":{"code":"import asyncio\nfrom pipecat_ai_flows import FlowManager, NodeConfig, flows_direct_function\nfrom pipecat_ai_flows.llm import LLMService # Base class\nfrom pipecat.frames.frames import TextFrame, EndFrame # Required for type hints\nimport os\n\n# Minimal mock LLMService and Transport to make the example runnable\nclass MockLLM(LLMService):\n    def __init__(self):\n        super().__init__(\"mock_llm\")\n    async def process_input(self, input_frames):\n        for frame in input_frames:\n            if isinstance(frame, TextFrame):\n                yield TextFrame(f\"Mock LLM received: {frame.text}\")\n        yield EndFrame()\n\nclass MockTransport:\n    async def send_frame(self, frame):\n        if isinstance(frame, TextFrame):\n            print(f\"Transport received text: {frame.text}\")\n        elif isinstance(frame, EndFrame):\n            print(\"Transport received EndFrame\")\n    async def receive_audio_frame(self): return None\n    async def receive_text_frame(self): return None\n\n@flows_direct_function(cancel_on_interruption=True)\nasync def greet_user(flow_manager: FlowManager, user_name: str = \"there\"):\n    \"\"\"Greets the user by their name.\"\"\"\n    await flow_manager.transport.send_frame(TextFrame(f\"Hello, {user_name}!\"))\n    return \"Greeting complete.\", \"start\" # Transition back to start\n\nasync def main():\n    print(\"Setting up Pipecat AI Flow Manager...\")\n\n    # Define nodes for the conversation flow\n    start_node = NodeConfig(\n        name=\"start\",\n        task_messages=[\n            {\"role\": \"developer\", \"content\": \"Ask the user for their name or just say hello.\"}\n        ],\n        functions=[greet_user], # Make `greet_user` available from this node\n        next_node=\"ask_name_node\", # Define a transition\n    )\n\n    ask_name_node = NodeConfig(\n        name=\"ask_name_node\",\n        task_messages=[\n            {\"role\": \"developer\", \"content\": \"If the user hasn't provided a name, ask for it. Otherwise, acknowledge the name.\"}\n        ],\n        next_node=None # End of simple flow for this example\n    )\n\n    # Initialize the FlowManager with nodes and required services\n    flow_manager = FlowManager(\n        initial_node=start_node, # The starting point of the flow\n        llm_service=MockLLM(), # In a real app, use pipecat_ai.services.openai.OpenAILLMService etc.\n        transport=MockTransport(), # In a real app, use pipecat_ai.transports.daily.DailyService etc.\n    )\n\n    print(f\"Flow Manager initialized. Current node: {flow_manager.current_node.name}\")\n    print(\"\\nTo activate the flow and start a conversation, integrate this FlowManager with a Pipecat AI PipelineRunner.\")\n    print(\"For example: `pipeline = Pipeline(llm=flow_manager.llm_service, vad=..., stt=..., tts=..., transport=flow_manager.transport)`\")\n    print(\"Then: `await PipelineRunner().run(pipeline)`\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())","lang":"python","description":"This quickstart demonstrates how to define a conversation flow using `FlowManager` and `NodeConfig`, including how to register direct functions with `@flows_direct_function`. It uses mock services to define the structure of a flow without requiring a full Pipecat AI pipeline setup. To run a full conversational agent, the `FlowManager`'s `llm_service` and `transport` would be connected to actual Pipecat AI components and integrated into a `PipelineRunner`."},"warnings":[{"fix":"Upgrade your Python environment to 3.11 or higher and run `pip install --upgrade pipecat-ai` to ensure compatibility.","message":"Version 1.0.0 requires Python >= 3.11 and `pipecat-ai>=1.0.0`. Older Python or Pipecat AI versions are not supported.","severity":"breaking","affected_versions":">=1.0.0"},{"fix":"Review all `task_messages` and summary messages in your `NodeConfig` definitions and change `\"role\": \"user\"` to `\"role\": \"developer\"`.","message":"All task and summary messages in `NodeConfig` must now use `\"role\": \"developer\"` instead of `\"role\": \"user\"` to correctly distinguish application instructions from user speech.","severity":"breaking","affected_versions":">=1.0.0"},{"fix":"Migrate your flow definitions from `FlowConfig` to `NodeConfig` and remove the `flow_config` argument from `FlowManager` initialization. Refer to the migration guide for v1.0.0.","message":"Static Flows (configured via the `flow_config` argument and `FlowConfig` type) have been deprecated since v0.0.19 and are now removed in v1.0.0. Only Dynamic Flows using `NodeConfig` are supported.","severity":"breaking","affected_versions":">=1.0.0"},{"fix":"Use the `role_message` field in your configurations for bot personality. If you rely on LLM system messages, ensure your `pipecat-ai` setup correctly handles `LLMUpdateSettingsFrame`.","message":"The `role_message` field is now the preferred way to set the bot's role/personality. System instructions are sent via `LLMUpdateSettingsFrame` rather than as system messages in the conversation context.","severity":"gotcha","affected_versions":">=0.0.24"},{"fix":"Apply `@flows_direct_function` to your direct functions and set `cancel_on_interruption=True` (default) or `False` as needed, to manage interruption behavior.","message":"The `@flows_direct_function` decorator allows configuring specific behaviors like `cancel_on_interruption` for functions directly invoked by the flow manager. This is crucial for controlling function execution during user interruptions.","severity":"gotcha","affected_versions":">=0.0.23"},{"fix":"To make functions universally accessible, pass a list of `FlowsFunctionSchema` or `FlowsDirectFunction` objects to the `global_functions` parameter when initializing `FlowManager`.","message":"`FlowManager` now supports a `global_functions` parameter during initialization, making functions available at every node without explicit definition in each `NodeConfig`.","severity":"gotcha","affected_versions":">=0.0.22"}],"env_vars":null,"last_verified":"2026-04-17T00:00:00.000Z","next_check":"2026-07-16T00:00:00.000Z","problems":[{"fix":"Change the `\"role\"` for all bot-generated instructions to `\"developer\"` in `task_messages` arrays.","cause":"Using `\"role\": \"user\"` for bot instructions within `task_messages` in `NodeConfig` or summary messages.","error":"TypeError: 'role' field in task message must be 'developer'"},{"fix":"Migrate your flow definitions to use `NodeConfig` for dynamic flows. `FlowConfig` and the `flow_config` argument in `FlowManager` have been removed.","cause":"Attempting to import or use `FlowConfig` or the deprecated static flow configuration after version 1.0.0.","error":"ImportError: cannot import name 'FlowConfig' from 'pipecat_ai_flows'"},{"fix":"Run `pip install --upgrade pipecat-ai` to ensure you have `pipecat-ai` version 1.0.0 or newer.","cause":"Your installed `pipecat-ai` package is older than the minimum required version for `pipecat-ai-flows` v1.0.0.","error":"RuntimeError: A newer version of pipecat-ai is required. Please upgrade to pipecat-ai>=1.0.0."},{"fix":"Upgrade your Python environment to version 3.11 or higher. Check your active Python version with `python --version`.","cause":"Running `pipecat-ai-flows` v1.0.0 (which requires Python 3.11+) with an older Python interpreter.","error":"SyntaxError: invalid syntax (related to new Python features like f-strings, type hints)"}]}