{"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.","language":"python","status":"active","last_verified":"Fri Apr 17","install":{"commands":["pip install pipecat-ai-flows"],"cli":null},"imports":["from pipecat_ai_flows import FlowManager","from pipecat_ai_flows import NodeConfig","from pipecat_ai_flows import flows_direct_function","from pipecat_ai_flows import ActionConfig"],"auth":{"required":false,"env_vars":[]},"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`.","tag":null,"tag_description":null,"last_tested":null,"results":[]},"compatibility":null}