### Initializing FlowManager in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Shows two methods for initializing the FlowManager: a static flow initialization which requires no arguments, and a dynamic flow initialization which accepts an initial node configuration. This method handles the setup of the flow manager's internal state. ```python async def initialize(self, initial_node: Optional[NodeConfig] = None) -> None: """Initialize the flow manager. Args: initial_node: Optional initial node configuration for dynamic flows. Raises: FlowInitializationError: If initialization fails. Examples: Static flow:: flow_manager = FlowManager( ... # Initialization parameters ) # Static flow: no initialization args required flow_manager.initialize() Dynamic flow:: flow_manager = FlowManager( ... # Initialization parameters ) # Dynamic flow: Initialize with the initial node configuration flow_manager.initialize(create_initial_node()) """ if self._initialized: logger.warning(f"{self.__class__.__name__} already initialized") return try: self._initialized = True logger.debug(f"Initialized {self.__class__.__name__}") # Set initial node node_name = None node = None if self._initial_node: # Static flow: self._initial_node is expected to be there node_name = self._initial_node node = self._nodes[self._initial_node] if not node: raise ValueError( f"Initial node '{self._initial_node}' not found in static flow configuration" ) else: # Dynamic flow: initial_node argument may have been provided (otherwise initial node # will be set later via set_node()) if initial_node: node_name = get_or_generate_node_name(initial_node) node = initial_node if node_name: logger.debug(f"Setting initial node: {node_name}") await self._set_node(node_name, node) except Exception as e: self._initialized = False raise FlowInitializationError(f"Failed to initialize flow: {str(e)}") from e ``` -------------------------------- ### Create LLM Adapter with Inheritance Detection Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Factory function that creates appropriate LLM adapters based on service type and inheritance hierarchy. Supports OpenAI, Anthropic, Google, and AWS Bedrock services with fallback inheritance checking. Returns provider-specific adapter instances or raises ValueError for unsupported types with detailed dependency installation instructions. ```python [docs] def create_adapter(llm, context_aggregator) -> LLMAdapter: """Create appropriate adapter based on context type and LLM service type or inheritance. Checks both direct class types and inheritance hierarchies to determine the appropriate adapter for any LLM service. Args: llm: LLM service instance or LLMSwitcher. context_aggregator: Context aggregator pair. Returns: Provider-specific adapter instance. Raises: ValueError: If LLM type is not supported or required dependency not installed. """ if isinstance(context_aggregator, LLMContextAggregatorPair): # Universal LLMContext is in use, so we need the universal adapter logger.debug("Creating universal adapter") return UniversalLLMAdapter() llm_type = type(llm).__name__ llm_class = type(llm) if llm_type == "OpenAILLMService": logger.debug("Creating OpenAI adapter") return OpenAIAdapter() if llm_type == "AnthropicLLMService": logger.debug("Creating Anthropic adapter") return AnthropicAdapter() if llm_type == "GoogleLLMService": logger.debug("Creating Google adapter") return GeminiAdapter() if llm_type == "AWSBedrockLLMService": logger.debug("Creating Bedrock adapter") return AWSBedrockAdapter() # Try to find OpenAILLMService for inheritance check try: module = sys.modules.get("pipecat.services.openai") if module: openai_service = getattr(module, "OpenAILLMService", None) if openai_service and issubclass(llm_class, openai_service): logger.debug(f"Creating OpenAI adapter for {llm_type}") return OpenAIAdapter() except (TypeError, AttributeError) as e: # Log but continue to error handling if issubclass check fails logger.warning(f"Error checking inheritance for {llm_type}: {str(e)}") # Error handling error_msg = ( f"Unsupported LLM type or missing dependency: {llm_type} (module: {llm_class.__module__})\n" ) error_msg += "Make sure you have installed the required dependency:\n" error_msg += "- For OpenAI: pip install 'pipecat-ai[openai]'\n" error_msg += "- For Anthropic: pip install 'pipecat-ai[anthropic]'\n" error_msg += "- For Google: pip install 'pipecat-ai[google]'\n" error_msg += "- For Bedrock: pip install 'pipecat-ai[aws]'" raise ValueError(error_msg) ``` -------------------------------- ### Setting Up Node in Flow Manager (Python) Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager This function sets up a new node in the flow manager by validating the configuration, clearing pending states, registering actions and functions, executing pre-actions, formatting tools for the LLM, and updating the context and state. It depends on class attributes like _initialized, _adapter, and logger, with inputs being node_id and node_config (a dictionary containing pre_actions, post_actions, functions, role_messages, task_messages, etc.). Outputs include updated internal state (_current_node, _current_functions) and executed actions; it raises FlowTransitionError if not initialized and may fail on node setup issues. ```python if not self._initialized: raise FlowTransitionError(f"{self.__class__.__name__} must be initialized first") try: # Clear any pending transition state when starting a new node # This ensures clean state regardless of how we arrived here: # - Normal transition flow (already cleared in _check_and_execute_transition) # - Direct calls to set_node/set_node_from_config self._pending_transition = None self._validate_node_config(node_id, node_config) logger.debug(f"Setting node: {node_id}") # Clear any deferred post-actions from previous node self._action_manager.clear_deferred_post_actions() # Register action handlers from config for action_list in [ node_config.get("pre_actions", []), node_config.get("post_actions", []), ]: for action in action_list: self._register_action_from_config(action) # Execute pre-actions if any if pre_actions := node_config.get("pre_actions"): await self._execute_actions(pre_actions=pre_actions) # Register functions and prepare tools tools: List[FlowsFunctionSchema | FlowsDirectFunctionWrapper] = [] new_functions: Set[str] = set() # Get functions list with default empty list if not provided functions_list = node_config.get("functions", []) async def register_function_schema(schema: FlowsFunctionSchema): """Helper to register a single FlowsFunctionSchema.""" tools.append(schema) await self._register_function( name=schema.name, new_functions=new_functions, handler=schema.handler, transition_to=schema.transition_to, transition_callback=schema.transition_callback, ) async def register_direct_function(func): """Helper to register a single direct function.""" direct_function = FlowsDirectFunctionWrapper(function=func) tools.append(direct_function) await self._register_function( name=direct_function.name, new_functions=new_functions, handler=direct_function, transition_to=None, transition_callback=None, ) for func_config in functions_list: # Handle direct functions if callable(func_config): await register_direct_function(func_config) # Handle Gemini's nested function declarations as a special case elif ( not isinstance(func_config, FlowsFunctionSchema) and "function_declarations" in func_config ): for declaration in func_config["function_declarations"]: # Convert each declaration to FlowsFunctionSchema and process it schema = self._adapter.convert_to_function_schema( {"function_declarations": [declaration]} ) await register_function_schema(schema) # Convert to FlowsFunctionSchema if needed and process it else: schema = ( func_config if isinstance(func_config, FlowsFunctionSchema) else self._adapter.convert_to_function_schema(func_config) ) await register_function_schema(schema) # Create ToolsSchema with standard function schemas standard_functions = [] for tool in tools: # Convert FlowsFunctionSchema to standard FunctionSchema for the LLM standard_functions.append(tool.to_function_schema()) # Use provider adapter to format tools, passing original configs for Gemini adapter formatted_tools = self._adapter.format_functions( standard_functions, original_configs=functions_list ) # Update LLM context await self._update_llm_context( role_messages=node_config.get("role_messages"), task_messages=node_config["task_messages"], functions=formatted_tools, strategy=node_config.get("context_strategy"), ) logger.debug("Updated LLM context") # Update state self._current_node = node_id self._current_functions = new_functions # Trigger completion with new context ``` -------------------------------- ### Legacy Action Handler in Python Source: https://reference-flows.pipecat.ai/en/latest/api/pipecat_flows This example shows a simple legacy action handler function that takes an action dictionary and asynchronously notifies based on text. It depends on an external async notify function. Input is the action dict; output is None. ```python async def simple_handler(action: dict): await notify(action["text"]) ``` -------------------------------- ### Module Imports and Dependencies Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/types Import statements and type annotations setup for the flow system. Includes standard library imports for UUID, dataclasses, and enums, along with comprehensive typing utilities from Python's typing module. ```python import uuid from dataclasses import dataclass from enum import Enum from typing import ( TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Mapping, Optional, Protocol, Tuple, TypedDict, TypeVar, Union, ) from pipecat.adapters.schemas.direct_function import BaseDirectFunctionWrapper from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat_flows.exceptions import InvalidFunctionError if TYPE_CHECKING: from pipecat_flows.manager import FlowManager ``` -------------------------------- ### Flow Action Handler in Python Source: https://reference-flows.pipecat.ai/en/latest/api/pipecat_flows This example demonstrates a modern flow action handler that takes an action dictionary and a FlowManager instance to perform notifications asynchronously. It requires a FlowManager with a transport.notify method. Inputs are action dict and flow_manager; output is None. ```python async def advanced_handler(action: dict, flow_manager: FlowManager): await flow_manager.transport.notify(action["text"]) ``` -------------------------------- ### Define entire conversation flow usingConfig JSON Source: https://reference-flows.pipecat.ai/en/latest/api/pipecat_flows This example shows a complete FlowConfig JSON structure for a static Pipecat conversation flow, including the initial node and a mapping of node configurations. Each node can contain role_messages, task_messages, functions, and actions. Note that static flows are deprecated and should be replaced by dynamic flows in future versions. ```JSON { "initial_node": "greeting", "nodes": { "greeting": { "role_messages": [...], "task_messages": [...], "functions": [...], "pre_actions": [] }, "process_order": { "task_messages": [...], "functions": [...], "post_actions": [] } } } ``` -------------------------------- ### Get Function Name from Dictionary Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Extracts function name from a dictionary definition. In universal LLMContext flows, raises RuntimeError as provider-specific definitions are disallowed. ```python def _get_function_name_from_dict(self, function_def: Dict[str, Any]) -> str: raise RuntimeError( "Provider-specific function definitions are not supported in flows using universal LLMContext. Use FlowsFunctionSchemas or direct functions instead." ) ``` -------------------------------- ### Set Node from Configuration in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Sets up a new conversation node and transitions to it using provided configuration. This method is used for manually transitioning between nodes in dynamic flows. Raises FlowTransitionError if the manager is not initialized or FlowError if node setup fails. Dependencies include 'NodeConfig' and 'get_or_generate_node_name'. ```python async def set_node_from_config(self, node_config: NodeConfig) -> None: """Set up a new conversation node and transition to it. Used to manually transition between nodes in a dynamic flow. Args: node_config: Configuration for the new node. Raises: FlowTransitionError: If manager not initialized. FlowError: If node setup fails. """ await self._set_node(get_or_generate_node_name(node_config), node_config) ``` -------------------------------- ### Registering Custom Action Handler in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Allows registration of a custom handler function for a specific action type. This enables extending the FlowManager's capabilities with custom logic. The handler can be either an async or sync function. An example is provided for registering a 'notify' action. ```python def register_action(self, action_type: str, handler: Callable) -> None: """Register a handler for a specific action type. Args: action_type: String identifier for the action (e.g., "tts_say"). handler: Async or sync function that handles the action. Example:: async def custom_notification(action: dict): text = action.get("text", "") await notify_user(text) flow_manager.register_action("notify", custom_notification) """ self._action_manager._register_action(action_type, handler) ``` -------------------------------- ### get_or_generate_node_name Source: https://reference-flows.pipecat.ai/en/latest/api/pipecat_flows Get the node name from configuration or generate a UUID if not set. ```APIDOC ## GET /pipecat_flows/types/get_or_generate_node_name ### Description Get the node name from configuration or generate a UUID if not set. ### Method GET ### Endpoint `/pipecat_flows/types/get_or_generate_node_name` ### Parameters #### Path Parameters None #### Query Parameters - **node_config** (NodeConfig) - Required - Node configuration dictionary. #### Request Body None ### Request Example None ### Response #### Success Response (200) - **name** (str) - Node name from config or generated UUID string. #### Response Example ```json { "name": "generated-uuid-or-config-name" } ``` ``` -------------------------------- ### Configure FlowsFunctionSchema with Parameters Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Returns a configured FlowsFunctionSchema object with parameters for name, description, properties, and callbacks. Defines the structure for flow function configuration in PipeCat AI workflows. Includes handler assignment and transition management settings. ```python return FlowsFunctionSchema( name=name, description=description, properties=properties, required=required, handler=handler, transition_to=transition_to, transition_callback=transition_callback, ) ``` -------------------------------- ### Format functions for Gemini API (Python) Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Converts function definitions from various formats (dicts, schemas) into Gemini's required format. Preserves original configurations when available and handles multiple schema types. ```python def format_functions( self, functions: List[Union[Dict[str, Any], FunctionSchema, FlowsFunctionSchema]], original_configs: Optional[List] = None, ) -> List[Dict[str, Any]]: """Format functions for Gemini's specific use. This special implementation processes both converted schemas and original configs to ensure Gemini's specific format is preserved when possible. Args: functions: List of function definitions (dicts or schema objects). original_configs: Optional original node configs, used to preserve native formats. Returns: List of functions formatted for Gemini. """ # TODO: here we should always be formatting into FunctionSchemas, not provider-specific format gemini_functions = [] # If original_configs is provided, extract functions from it if original_configs: for func_config in original_configs: if isinstance(func_config, FlowsFunctionSchema): # Convert FlowsFunctionSchema to Gemini format gemini_functions.append( { "name": func_config.name, "description": func_config.description, "parameters": { "type": "object", "properties": func_config.properties, "required": func_config.required, }, } ) elif isinstance(func_config, Callable): # Convert direct function to Gemini format direct_func = FlowsDirectFunctionWrapper(function=func_config) gemini_functions.append( { "name": direct_func.name, "description": direct_func.description, "parameters": { "type": "object", "properties": direct_func.properties, "required": direct_func.required, }, } ) elif "function_declarations" in func_config: # Already in Gemini format, use directly but remove handler/transition fields for decl in func_config["function_declarations"]: decl_copy = decl.copy() if "handler" in decl_copy: del decl_copy["handler"] if "transition_to" in decl_copy: del decl_copy["transition_to"] if "transition_callback" in decl_copy: del decl_copy["transition_callback"] gemini_functions.append(decl_copy) else: # If no original configs, use the converted schemas for func in functions: if isinstance(func, (FunctionSchema, FlowsFunctionSchema)): # Convert to Gemini format gemini_functions.append( { "name": func.name, "description": func.description, "parameters": { "type": "object", "properties": func.properties, "required": func.required, }, } ) # Return empty list if no functions if not gemini_functions: return [] # Format as Gemini expects - an array with a single object containing function_declarations return [{"function_declarations": gemini_functions}] ``` -------------------------------- ### Format Functions for Universal LLMContext Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Formats function definitions for use with universal LLMContext. Accepts various input formats and returns a ToolsSchema or NOT_GIVEN indicator. Ensures compatibility with Pipecat's universal context requirements. ```python def format_functions( self, functions: List[Union[Dict[str, Any], FunctionSchema, FlowsFunctionSchema]], original_configs: Optional[List] = None, ) -> ToolsSchema | NotGiven: formatted_functions = super().format_functions(functions, original_configs=original_configs) if not formatted_functions: return NOT_GIVEN return formatted_functions ``` -------------------------------- ### Base LLMAdapter Class Definition Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Defines the base adapter for LLM-specific format handling. It outlines methods for extracting function names and formatting functions for provider-specific use. Subclasses must implement the abstract methods to support specific LLM providers. ```python import sys from typing import Any, Callable, Dict, List, Optional, Union from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.processors.aggregators.llm_context import NOT_GIVEN, LLMContext, NotGiven from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat_flows.types import FlowsDirectFunctionWrapper, FlowsFunctionSchema class LLMAdapter: """Base adapter for LLM-specific format handling. Adapters normalize differences between LLM providers to allow the flow system to work consistently across different providers while handling format differences internally. Each provider has specific requirements for function calling, message formatting, and tool definitions. Supported providers: - OpenAI: Uses function calling format - Anthropic: Uses native function format - Google Gemini: Uses function declarations format - AWS Bedrock: Uses Anthropic-compatible format """ def get_function_name(self, function_def: Union[Dict[str, Any], FlowsFunctionSchema]) -> str: """Extract function name from provider-specific function definition or schema. Args: function_def: Provider-specific function definition or schema. Returns: Function name extracted from the definition. """ if isinstance(function_def, (FlowsFunctionSchema)): return function_def.name return self._get_function_name_from_dict(function_def) def _get_function_name_from_dict(self, function_def: Dict[str, Any]) -> str: """Extract function name from provider-specific function definition. Args: function_def: Provider-specific function definition dictionary. Returns: Function name extracted from the definition. Raises: NotImplementedError: Must be implemented by subclasses. """ raise NotImplementedError("Subclasses must implement this method") def format_functions( self, functions: List[Union[Dict[str, Any], FunctionSchema, FlowsFunctionSchema]], original_configs: Optional[List] = None, ) -> List[Dict[str, Any]] | ToolsSchema | NotGiven: """Format functions for provider-specific use. The formatted functions/tools will be used in an LLMSetToolsFrame. The NotGiven return type is only used by the subclass UniversalLLMAdapter, as the universal LLMContext expects NOT_GIVEN when there are no tools. All other LLM-specific contexts expect either a list of dicts or a ToolsSchema, with an empty list meaning "no tools". Args: functions: List of function definitions (dicts or schema objects). original_configs: Optional original node configs, used by some adapters. Returns: List of functions formatted for the provider. """ # Return empty list to mean "no tools" if not functions: return [] # Convert to standard FunctionSchema objects for the ToolsSchema standard_functions = [] for func in functions: if isinstance(func, FlowsFunctionSchema): # Extract just the FunctionSchema part for the LLM standard_functions.append( FunctionSchema( name=func.name, description=func.description, properties=func.properties, required=func.required, ) ) elif isinstance(func, FunctionSchema): # Already a standard FunctionSchema standard_functions.append(func) else: # Convert legacy dictionary format to FunctionSchema flows_schema = self.convert_to_function_schema(func) # Extract just the FunctionSchema part for the LLM standard_functions.append( FunctionSchema( name=flows_schema.name, description=flows_schema.description, properties=flows_schema.properties, required=flows_schema.required, ) ``` -------------------------------- ### Generate Summary from LLM Context Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Generates a conversation summary using an LLM with a system prompt and message history. Supports both OpenAILLMContext and generic LLMContext inputs. Returns None if generation fails. ```python async def generate_summary( self, llm: Any, summary_prompt: str, context: OpenAILLMContext | LLMContext ) -> Optional[str]: try: if isinstance(context, LLMContext): messages = context.get_messages() else: messages = context.messages prompt_messages = [ { "role": "system", "content": summary_prompt, }, { "role": "user", "content": f"Conversation history: {messages}", }, ] summary_context = LLMContext(messages=prompt_messages) return await llm.run_inference(summary_context) except Exception as e: logger.error(f"Summary generation failed: {e}", exc_info=True) return None ``` -------------------------------- ### Getting Current Context in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Retrieves the current conversation context, which includes system messages, user messages, and assistant responses. This function requires a context aggregator to be available, otherwise it raises a FlowError. It can return messages from either an LLMContext or a standard context object. ```python def get_current_context(self) -> List[dict]: """Get the current conversation context. Returns: List of messages in the current context, including system messages, user messages, and assistant responses. Raises: FlowError: If context aggregator is not available. """ if not self._context_aggregator: raise FlowError("No context aggregator available") context = self._context_aggregator.user()._context if isinstance(context, LLMContext): return context.get_messages() else: return context.messages ``` -------------------------------- ### Implement FlowManager class and initialization in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Provides the full Python implementation of the FlowManager, including required imports, type aliases, and the __init__ method that sets up task handling, LLM services, action management, and context strategies. Handles deprecation warnings for the tts and static flow parameters. Dependencies include pipecat framework components and loguru for logging. ```Python import asyncio\nimport inspect\nimport sys\nimport warnings\nfrom typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Union, cast\n\nfrom loguru import logger\nfrom pipecat.frames.frames import (\n FunctionCallResultProperties,\n LLMMessagesAppendFrame,\n LLMMessagesUpdateFrame,\n LLMRunFrame,\n LLMSetToolsFrame,\n)\nfrom pipecat.pipeline.llm_switcher import LLMSwitcher\nfrom pipecat.pipeline.task import PipelineTask\nfrom pipecat.processors.aggregators.llm_context import LLMContext\nfrom pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext\nfrom pipecat.services.llm_service import FunctionCallParams\nfrom pipecat.transports.base_transport import BaseTransport\n\nfrom pipecat_flows.actions import ActionError, ActionManager\nfrom pipecat_flows.adapters import create_adapter\nfrom pipecat_flows.exceptions import (\n FlowError,\n FlowInitializationError\n FlowTransitionError,\n InvalidFunctionError,\n)\nfrom pipecat_flows.types import (\n ActionConfig,\n ConsolidatedFunctionResult,\n ContextStrategy,\n ContextStrategyConfig,\n FlowArgs,\n FlowConfig,\n FlowResult,\n FlowsDirectFunctionWrapper,\n FlowsFunctionSchema,\n FunctionHandler,\n NodeConfig,\n get_or_generate_node_name,\n)\n\nif TYPE_CHECKING:\n from pipecat.services.anthropic.llm import AnthropicLLMService\n from pipecat.services.google.llm import GoogleLLMService\n from pipecat.services.openai.llm import OpenAILLMService\n\n LLMService = Union[OpenAILLMService, AnthropicLLMService, GoogleLLMService]\nelse:\n LLMService = Any\n\nclass FlowManager:\n """Manages conversation flows supporting both static and dynamic configurations.\n\n The FlowManager orchestrates conversation flows by managing state transitions,\n function registration, and message handling across different LLM providers.\n It supports both predefined static flows and runtime-determined dynamic flows\n with comprehensive action handling and error management.\n\n The manager coordinates all aspects of a conversation including LLM context\n management, function registration, state transitions, action execution, and\n provider-specific format handling.\n """\n\n def __init__(\n self,\n *,\n task: PipelineTask,\n llm: LLMService | LLMSwitcher,\n context_aggregator: Any,\n tts: Optional[Any = None,\n flow_config: Optional[FlowConfig] = None,\n context_strategy: Optional[ContextStrategyConfig] = None,\n transport: Optional[BaseTransport] = None,\n ):\n """Initialize the flow manager.\n\n Args:\n task: PipelineTask instance for queueing frames.\n llm: LLM service or LLMSwitcher.\n context_aggregator: Context aggregator for updating user context.\n tts: Text-to-speech service for voice actions.\n\n .. deprecated:: 0.0.18\n The tts parameter is deprecated and will be removed in 1.0.0.\n\n flow_config: Static flow configuration. If provided, operates in static\n mode with predefined nodes.\n\n .. deprecated:: 0.0.19\n Static flows are deprecated and will be removed in 1.0.0.\n Use dynamic flows instead.\n\n context_strategy: Context strategy configuration for managing conversation\n context during transitions.\n transport: Transport instance for communication.\n\n Raises:\n ValueError: If any transition handler is not a valid async callable.\n """\n if tts is not None:\n warnings.warn(\n "The 'tts' parameter is deprecated and will be removed in 1.0.0.",\n DeprecationWarning,\n stacklevel=2,\n )\n\n self._task = task\n self._llm = llm\n self._action_manager = ActionManager(task, flow_manager=self)\n self._adapter = create_adapter(llm, context_aggregator)\n self._initialized = False\n self._context_aggregator = context_aggregator\n self._pending_transition: Optional[Dict[str, Any]] = None\n self._context_strategy = context_strategy or ContextStrategyConfig(\n strategy=ContextStrategy.APPEND\n )\n self._transport = transport\n\n # Set up static or dynamic mode\n if flow_config:\n warnings.warn( ``` -------------------------------- ### Format summary message for Gemini (Python) Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Converts a summary string into a Gemini-formatted user message. Summary messages are always expected in OpenAI format for compatibility with LLMMessagesUpdateFrame. ```python def format_summary_message(self, summary: str) -> dict: """Format summary as a user message for Gemini. Note: summary messages are always expected in OpenAI format, because summarization triggers an LLMMessagesUpdateFrame, which expects OpenAI-style messages. Args: summary: The generated summary text. Returns: Gemini-formatted user message containing the summary. """ ``` -------------------------------- ### Initialize ActionManager with Built-in Actions Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/actions Initializes the ActionManager, registering built-in action handlers for 'tts_say', 'end_conversation', and 'function'. It also sets up event handlers for frame processing and manages ongoing action counts. ```python class ActionManager: """Manages the registration and execution of flow actions. Actions are executed during state transitions and can include: - Text-to-speech output - Database updates - External API calls - Custom user-defined actions Built-in actions: - tts_say: Speak text using TTS - end_conversation: End the current conversation - function: Execute inline functions in the pipeline Custom actions can be registered using register_action(). """ def __init__(self, task: PipelineTask, flow_manager: "FlowManager"): """Initialize the action manager. Args: task: PipelineTask instance used to queue frames. flow_manager: FlowManager instance that this ActionManager is part of. """ self._action_handlers: Dict[str, Callable] = {} self._task = task self._flow_manager = flow_manager self._ongoing_actions_count = 0 self._ongoing_actions_finished_event = asyncio.Event() self._deferred_post_actions: List[ActionConfig] = [] # Register built-in actions self._register_action("tts_say", self._handle_tts_action) self._register_action("end_conversation", self._handle_end_action) self._register_action("function", self._handle_function_action) # Add pipeline observation task.set_reached_downstream_filter( (ActionFinishedFrame, FunctionActionFrame, BotStoppedSpeakingFrame) ) @task.event_handler("on_frame_reached_downstream") async def on_frame_reached_downstream(task, frame): if isinstance(frame, FunctionActionFrame): # Run function action await frame.function(frame.action, flow_manager) self._decrement_ongoing_actions_count() elif isinstance(frame, BotStoppedSpeakingFrame): # Execute deferred post-actions if the bot's turn is over. # A BotStoppedSpeakingFrame only indicates that the bot's turn is over if there are # no ongoing actions (otherwise one of those actions may have been responsible for it). if self._ongoing_actions_count == 0: await self._execute_deferred_post_actions() elif isinstance(frame, ActionFinishedFrame): # Handle action finished self._decrement_ongoing_actions_count() def _register_action(self, action_type: str, handler: Callable) -> None: """Register a handler for a specific action type. Args: action_type: String identifier for the action (e.g., "tts_say"). handler: Async or sync function that handles the action. Raises: ValueError: If handler is not callable. """ if not callable(handler): raise ValueError("Action handler must be callable") self._action_handlers[action_type] = handler logger.debug(f"Registered handler for action type: {action_type}") async def execute_actions(self, actions: Optional[List[ActionConfig]]) -> None: """Execute a list of actions. Args: actions: List of action configurations to execute. Raises: ActionError: If action execution fails. Note: Each action must have a 'type' field matching a registered handler. """ if not actions: return ``` -------------------------------- ### Schedule deferred post-actions for LLM completion Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/actions Stores a list of actions to be executed after the next LLM response completion. The actions are stored in the instance variable _deferred_post_actions. ```python def schedule_deferred_post_actions(self, post_actions: List[ActionConfig]) -> None: """Schedule "deferred" post-actions to be executed after next LLM completion. Args: post_actions: List of actions to execute after LLM response. """ self._deferred_post_actions = post_actions ``` -------------------------------- ### Configure role and task messages with functions in JSON Source: https://reference-flows.pipecat.ai/en/latest/api/pipecat_flows This snippet demonstrates how to define role_messages, task_messages, functions, pre_actions, post_actions, context_strategy, and respond_immediately in a Pipecat flow node configuration. It is useful for setting up the bot's personality, tasks, and execution behavior. The configuration relies on the Pipecat library and expects valid function and action definitions. ```JSON { "role_messages": [ { "role": "system", "content": "You are a helpful assistant..." } ], "task_messages": [ { "role": "system", "content": "Ask the user for their name..." } ], "functions": [...], "pre_actions": [...],n "post_actions": [...], "_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND), "respond_immediately": true } ``` -------------------------------- ### Internal Method: Set Node in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Internal helper method to set up a new conversation node and transition to it. It orchestrates the node transition process, including executing pre-actions, setting messages, registering functions, updating LLM context and state, triggering LLM completion, and executing post-actions. Dependencies include 'NodeConfig'. ```python async def _set_node(self, node_id: str, node_config: NodeConfig) -> None: """Set up a new conversation node and transition to it. Handles the complete node transition process in the following order: 1. Execute pre-actions (if any) 2. Set up messages (role and task) 3. Register node functions 4. Update LLM context with messages and tools 5. Update state (current node and functions) 6. Trigger LLM completion with new context 7. Execute post-actions (if any) Args: node_id: Identifier for the new node. node_config: Complete configuration for the node. Raises: ``` -------------------------------- ### OpenAI Adapter: Convert Function Definition to FlowsFunctionSchema Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Converts an OpenAI-formatted function definition dictionary into a FlowsFunctionSchema object. It extracts core function details such as name, description, properties, and required fields, along with flow-specific fields like handler, transition_to, and transition_callback. ```python class OpenAIAdapter(LLMAdapter): """Format adapter for OpenAI. Handles OpenAI's function calling format, which is used as the default format in the flow system. """ def _get_function_name_from_dict(self, function_def: Dict[str, Any]) -> str: """Extract function name from OpenAI function definition. Args: function_def: OpenAI-formatted function definition dictionary. Returns: Function name from the definition. """ return function_def["function"]["name"] def format_summary_message(self, summary: str) -> dict: """Format summary as a system message for OpenAI. Args: summary: The generated summary text. Returns: OpenAI-formatted system message containing the summary. """ return {"role": "system", "content": f"Here's a summary of the conversation:\n{summary}"} def convert_to_function_schema(self, function_def: Dict[str, Any]) -> FlowsFunctionSchema: """Convert OpenAI function definition to FlowsFunctionSchema. Args: function_def: OpenAI function definition. Returns: FlowsFunctionSchema equivalent with flow-specific fields. """ func_data = function_def["function"] name = func_data["name"] description = func_data.get("description", "") parameters = func_data.get("parameters", {}) or {} properties = parameters.get("properties", {}) required = parameters.get("required", []) # Extract Flows-specific fields handler = func_data.get("handler") transition_to = func_data.get("transition_to") transition_callback = func_data.get("transition_callback") return FlowsFunctionSchema( name=name, description=description, properties=properties, required=required, handler=handler, transition_to=transition_to, transition_callback=transition_callback, ) ``` -------------------------------- ### Format Summary as System Message Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Formats a generated summary string into a system message dictionary compatible with LLMContext. Used for injecting summaries into conversation flows. ```python def format_summary_message(self, summary: str) -> dict: return {"role": "system", "content": f"Here's a summary of the conversation:\n{summary}"} ``` -------------------------------- ### Context Strategy Configuration Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/types Configuration class for managing context strategies, including the selected strategy and an optional summary prompt for the RESET_WITH_SUMMARY strategy. Includes validation to ensure the summary prompt is provided when required. ```python @dataclass class ContextStrategyConfig: """Configuration for context management. Parameters: strategy: Strategy to use for context management. summary_prompt: Required prompt text when using RESET_WITH_SUMMARY. """ strategy: ContextStrategy summary_prompt: Optional[str] = None def __post_init__(self): """Validate configuration. Raises: ValueError: If summary_prompt is missing when using RESET_WITH_SUMMARY. """ if self.strategy == ContextStrategy.RESET_WITH_SUMMARY and not self.summary_prompt: raise ValueError("summary_prompt is required when using RESET_WITH_SUMMARY strategy") ``` -------------------------------- ### Execute Pre and Post Actions in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager This method executes optional pre-actions before and post-actions after updating the conversation context. It relies on an internal ActionManager for execution and supports asynchronous operations. Inputs are lists of ActionConfig objects; no outputs are returned, but it raises exceptions if actions fail. ```python post_actions: Optional[List[ActionConfig]] = None, ) -> None: """Execute pre and post actions. Args: pre_actions: Actions to execute before context update. post_actions: Actions to execute after context update. """ if pre_actions: await self._action_manager.execute_actions(pre_actions) if post_actions: await self._action_manager.execute_actions(post_actions) ``` -------------------------------- ### Generate Conversation Summary for LLM Context Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Asynchronously generates a conversation summary using the provided LLM adapter and context. This is crucial for strategies that rely on summarizing past interactions. ```python async def _create_conversation_summary( self, summary_prompt: str, context: OpenAILLMContext | LLMContext, ) -> Optional[str]: """Generate a conversation summary from a given context.""" return await self._adapter.generate_summary(self._llm, summary_prompt, context) ``` -------------------------------- ### Convert Gemini function definition to FlowsFunctionSchema Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/adapters Converts Gemini function definitions to FlowsFunctionSchema format. Handles function declarations with support for multiple declarations by using the first one. Extracts standard function properties including name, description, parameters, and required fields along with flow-specific metadata like handler and transition information. ```python def convert_to_function_schema(self, function_def: Dict[str, Any]) -> FlowsFunctionSchema: """Convert Gemini function definition to FlowsFunctionSchema. Args: function_def: Gemini function definition. Returns: FlowsFunctionSchema equivalent with flow-specific fields. """ if "function_declarations" in function_def: # Use first declaration if there are multiple decl = function_def["function_declarations"][0] # If we have function declarations, the handler might be in the declaration handler = decl.get("handler") transition_to = decl.get("transition_to") transition_callback = decl.get("transition_callback") else: decl = function_def # Otherwise, the handler might be at the top level handler = function_def.get("handler") transition_to = function_def.get("transition_to") transition_callback = function_def.get("transition_callback") name = decl["name"] description = decl.get("description", "") parameters = decl.get("parameters", {}) or {} properties = parameters.get("properties", {}) required = parameters.get("required", []) return FlowsFunctionSchema( name=name, description=description, properties=properties, required=required, handler=handler, transition_to=transition_to, transition_callback=transition_callback, ) ``` -------------------------------- ### Create Transition Function in Python Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/manager Creates a transition function for flow management. Validates transition configuration and handles both static and dynamic transitions. Supports handler execution and result processing. ```python async def _create_transition_func( self, name: str, handler: Optional[Callable | FlowsDirectFunctionWrapper], transition_to: Optional[str], transition_callback: Optional[Callable] = None, ) -> Callable: """Create a transition function for the given name and handler.""" if transition_to and transition_callback: raise ValueError( f"Function {name} cannot have both transition_to and transition_callback" ) # Validate transition callback if provided if transition_callback: self._validate_transition_callback(name, transition_callback) async def transition_func(params: FunctionCallParams) -> None: """Inner function that handles the actual tool invocation.""" try: logger.debug(f"Function called: {name}") # Execute handler if present is_transition_only_function = False acknowledged_result = {"status": "acknowledged"} if handler: # Invoke the handler with the provided arguments if isinstance(handler, FlowsDirectFunctionWrapper): handler_response = await handler.invoke(params.arguments, self) else: handler_response = await self._call_handler(handler, params.arguments) # Support both "consolidated" handlers that return (result, next_node) and handlers # that return just the result. if isinstance(handler_response, tuple): result, next_node = handler_response if result is None: result = acknowledged_result is_transition_only_function = True else: result = handler_response next_node = None # FlowsDirectFunctions should always be "consolidated" functions that return a tuple if isinstance(handler, FlowsDirectFunctionWrapper): raise InvalidFunctionError( f"Direct function {name} expected to return a tuple (result, next_node) but got {type(result)}" ) else: result = acknowledged_result next_node = None is_transition_only_function = True ``` -------------------------------- ### Handle Function Action Source: https://reference-flows.pipecat.ai/en/latest/_modules/pipecat_flows/actions This asynchronous function handles queuing functions to run inline in the pipeline. It expects a 'handler' key in the action containing the function to execute. It increments the ongoing actions count and queues a FunctionActionFrame for execution when the pipeline is done with previous work. ```python async def _handle_function_action(self, action: dict) -> None: """Built-in handler for queuing functions to run inline in the pipeline. This handler queues a FunctionActionFrame to be executed when the pipeline is done with all the work queued before it. It expects a 'handler' key in the action containing the function to execute. Args: action: Action configuration dictionary. Required 'handler' key containing the function to execute. """ handler = action.get("handler") if not handler: logger.error("Function action missing 'handler' field") return # Mark that we're starting the action self._increment_ongoing_actions_count() # Queue the action frame (we're queueing rather than running it here to ensure it happens ```