wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use the new pipecat-ai-flows module

Browse Source
This commit is contained in:
Mark Backman
2024-11-18 21:29:35 -05:00
parent b3dfeb61c4
commit 97659ca3f0
5 changed files with 11 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
examples/foundational/25-conversation-flow.py
View File

@@ -11,10 +11,10 @@ import sys
import aiohttp
from dotenv import load_dotenv
from loguru import logger
from pipecat_flows import FlowManager
from runner import configure
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.flows.manager import FlowManager
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -64,7 +64,7 @@ flow_config = {
"messages": [
{
"role": "system",
"content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose.",
"content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose. Start off by greeting them. Be friendly and casual; you're taking an order for food over the phone.",
}
],
"functions": [
@@ -95,7 +95,10 @@ flow_config = {
- Use the end function ONLY when the user confirms they are done with their order
After each size selection, confirm the selection and ask if they want to change it or complete their order.
Only use the end function after the user confirms they are satisfied with their order.""",
Only use the end function after the user confirms they are satisfied with their order.
Start off by acknowledging the user's choice. Once they've chosen a size, ask if they'd like anything else.
Remember to be friendly and casual.""",
}
],
"functions": [
@@ -139,7 +142,10 @@ flow_config = {
- Use the end function ONLY when the user confirms they are done with their order
After each roll count selection, confirm the count and ask if they want to change it or complete their order.
Only use the end function after the user confirms they are satisfied with their order.""",
Only use the end function after the user confirms they are satisfied with their order.
Start off by acknowledging the user's choice. Once they've chosen a size, ask if they'd like anything else.
Remember to be friendly and casual.""",
}
],
"functions": [

1
pyproject.toml
View File

@@ -51,6 +51,7 @@ gladia = [ "websockets~=13.1" ]
google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.17.2" ]
gstreamer = [ "pygobject~=3.48.2" ]
fireworks = [ "openai~=1.37.2" ]
flows = [ "pipecat-ai-flows~=0.0.1" ]
krisp = [ "pipecat-ai-krisp~=0.2.0" ]
langchain = [ "langchain~=0.2.14", "langchain-community~=0.2.12", "langchain-openai~=0.1.20" ]
livekit = [ "livekit~=0.17.5", "livekit-api~=0.7.1", "tenacity~=8.5.0" ]

10
src/pipecat/flows/__init__.py
View File

@@ -1,10 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from .manager import FlowManager
from .state import FlowState, NodeConfig
__all__ = ["FlowState", "FlowManager", "NodeConfig"]

223
src/pipecat/flows/manager.py
View File

@@ -1,223 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from asyncio import iscoroutinefunction
from typing import Callable, Dict, List, Optional
from loguru import logger
from pipecat.frames.frames import (
EndFrame,
LLMMessagesAppendFrame,
LLMMessagesUpdateFrame,
LLMSetToolsFrame,
TTSSpeakFrame,
)
from .state import FlowState
class FlowManager:
"""Manages conversation flows in a Pipecat pipeline.
This manager handles the progression through a flow defined by nodes, where each node
represents a state in the conversation. Each node has:
- Messages for the LLM (system/user/assistant messages)
- Available functions that can be called
- Optional pre-actions to execute before LLM inference
- Optional post-actions to execute after LLM inference
The flow is defined by a configuration that specifies:
- Initial node
- Available nodes and their configurations
- Transitions between nodes via function calls
"""
def __init__(self, flow_config: dict, task, tts=None):
"""Initialize the flow manager.
Args:
flow_config: Dictionary containing the complete flow configuration,
including initial_node and node configurations
task: PipelineTask instance used to queue frames into the pipeline
"""
self.flow = FlowState(flow_config)
self.initialized = False
self.task = task
self.tts = tts
self.action_handlers: Dict[str, Callable] = {}
# Register built-in actions
self.register_action("tts_say", self._handle_tts_action)
self.register_action("end_conversation", self._handle_end_action)
async def initialize(self, initial_messages: List[dict]):
"""Initialize the flow with starting messages and functions.
This method sets up the initial context, combining any system-level
messages with the initial node's message and functions.
Args:
initial_messages: List of initial messages (typically system messages)
to include in the context
"""
if not self.initialized:
messages = initial_messages + self.flow.get_current_messages()
await self.task.queue_frame(LLMMessagesUpdateFrame(messages=messages))
await self.task.queue_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions()))
self.initialized = True
logger.debug(f"Initialized flow at node: {self.flow.current_node}")
else:
logger.warning("Attempted to initialize FlowManager multiple times")
async def register_functions(self, llm_service):
"""Register all functions from the flow configuration with the LLM service.
This method sets up function handlers for all functions defined across all nodes.
When a function is called, it will automatically trigger the appropriate node
transition.
Note: This registers handlers for all possible functions, but the LLM's access
to functions is controlled separately through LLMSetToolsFrame. The LLM will
only see the functions available in the current node.
Args:
llm_service: The LLM service to register functions with
"""
async def handle_function_call(
function_name, tool_call_id, arguments, llm, context, result_callback
):
await self.handle_transition(function_name)
await result_callback("Acknowledged")
# Register all functions from all nodes
for node in self.flow.nodes.values():
for function in node.functions:
function_name = function["function"]["name"]
llm_service.register_function(function_name, handle_function_call)
def register_action(self, action_type: str, handler: Callable):
"""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.
Should accept action configuration as parameter.
"""
if not callable(handler):
raise ValueError("Action handler must be callable")
self.action_handlers[action_type] = handler
async def _execute_actions(self, actions: Optional[List[dict]]) -> None:
"""Execute actions specified for the current node.
Args:
actions: List of action configurations to execute
Note:
Each action must have a 'type' field matching a registered handler
"""
if not actions:
return
for action in actions:
action_type = action["type"]
if action_type in self.action_handlers:
handler = self.action_handlers[action_type]
try:
if iscoroutinefunction(handler):
await handler(action)
else:
handler(action)
except Exception as e:
logger.warning(f"Error executing action {action_type}: {e}")
else:
logger.warning(f"No handler registered for action type: {action_type}")
async def _handle_tts_action(self, action: dict):
"""Built-in handler for TTS actions that speak immediately.
This handler attempts to use the TTS service directly to speak the text
immediately, bypassing the pipeline queue. If no TTS service is available,
it falls back to queueing the text through the pipeline.
Args:
action: Dictionary containing the action configuration.
Must include a 'text' key with the text to speak.
"""
if self.tts:
# Direct call to TTS service to speak text immediately
await self.tts.say(action["text"])
else:
# Fall back to queued TTS if no direct service available
await self.task.queue_frame(TTSSpeakFrame(text=action["text"]))
async def _handle_end_action(self, action: dict):
"""Built-in handler for ending the conversation.
This handler queues an EndFrame to terminate the conversation. If the action
includes a 'text' key, it will queue that text to be spoken before ending.
Args:
action: Dictionary containing the action configuration.
Optional 'text' key for a goodbye message.
"""
if action.get("text"): # Optional goodbye message
await self.task.queue_frame(TTSSpeakFrame(text=action["text"]))
await self.task.queue_frame(EndFrame())
async def handle_transition(self, function_name: str):
"""Handle node transition triggered by a function call.
This method:
1. Validates the function call against available functions
2. Transitions to the new node if appropriate
3. Executes any pre-actions before updating the LLM context
4. Updates the LLM context with new messages and available functions
5. Executes any post-actions after updating the LLM context
Args:
function_name: Name of the function that was called
Raises:
RuntimeError: If handle_transition is called before initialization
"""
if not self.initialized:
raise RuntimeError("FlowManager must be initialized before handling transitions")
available_functions = self.flow.get_available_function_names()
current_node = self.flow.get_current_node()
if function_name in available_functions:
new_node = self.flow.transition(function_name)
if new_node:
# Only execute actions if we actually changed nodes
is_new_node = new_node != current_node
# Execute pre-actions before updating LLM context
if is_new_node and self.flow.get_current_pre_actions():
logger.debug(f"Executing pre-actions for node {new_node}")
await self._execute_actions(self.flow.get_current_pre_actions())
# Update LLM context and tools
current_messages = self.flow.get_current_messages()
await self.task.queue_frame(LLMMessagesAppendFrame(messages=current_messages))
await self.task.queue_frame(
LLMSetToolsFrame(tools=self.flow.get_current_functions())
)
# Execute post-actions after updating LLM context
if is_new_node and self.flow.get_current_post_actions():
logger.debug(f"Executing post-actions for node {new_node}")
await self._execute_actions(self.flow.get_current_post_actions())
logger.debug(f"Transition to node {new_node} complete")
else:
logger.warning(
f"Received invalid function call '{function_name}' for node '{self.flow.current_node}'. "
f"Available functions are: {available_functions}"
)

163
src/pipecat/flows/state.py
View File

@@ -1,163 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from dataclasses import dataclass
from typing import Dict, List, Optional, Set
from loguru import logger
@dataclass
class NodeConfig:
"""Configuration for a single node in the flow.
A node represents a state in the conversation flow, containing all the
information needed for that particular point in the conversation.
Attributes:
messages: List of message dicts to be added to LLM context at this node.
Each message should have 'role' (system/user/assistant) and 'content'.
Messages are added in order, allowing for complex prompt building.
functions: List of available function definitions for this node
pre_actions: Optional list of actions to execute before LLM inference
post_actions: Optional list of actions to execute after LLM inference
"""
messages: List[dict]
functions: List[dict]
pre_actions: Optional[List[dict]] = None
post_actions: Optional[List[dict]] = None
class FlowState:
"""Manages the state and transitions between nodes in a conversation flow.
This class handles the state machine logic for conversation flows, where each node
represents a distinct state with its own messages, available functions, and optional
pre- and post-actions. It manages transitions between nodes based on function calls
and handles both regular and terminal functions.
Attributes:
nodes: Dictionary mapping node IDs to their configurations
current_node: ID of the currently active node
"""
def __init__(self, flow_config: dict):
"""Initialize the conversation flow.
Args:
flow_config: Dictionary containing the complete flow configuration,
must include 'initial_node' and 'nodes' keys
Raises:
ValueError: If required configuration keys are missing
"""
self.nodes: Dict[str, NodeConfig] = {}
self.current_node: str = flow_config["initial_node"]
self._load_config(flow_config)
def _load_config(self, config: dict):
"""Load and validate the flow configuration.
Args:
config: Dictionary containing the flow configuration
Raises:
ValueError: If required configuration keys are missing
"""
if "initial_node" not in config:
raise ValueError("Flow config must specify 'initial_node'")
if "nodes" not in config:
raise ValueError("Flow config must specify 'nodes'")
for node_id, node_config in config["nodes"].items():
self.nodes[node_id] = NodeConfig(
messages=node_config["messages"],
functions=node_config["functions"],
pre_actions=node_config.get("pre_actions"),
post_actions=node_config.get("post_actions"),
)
def get_current_messages(self) -> List[dict]:
"""Get the messages for the current node.
Returns:
List of message dictionaries for the current node
"""
return self.nodes[self.current_node].messages
def get_current_functions(self) -> List[dict]:
"""Get the available functions for the current node.
Returns:
List of function definitions available in the current node
"""
return self.nodes[self.current_node].functions
def get_current_pre_actions(self) -> Optional[List[dict]]:
"""Get the pre-actions for the current node.
Pre-actions are executed before updating the LLM context when
transitioning to this node.
Returns:
List of pre-actions to execute, or None if no pre-actions
"""
return self.nodes[self.current_node].pre_actions
def get_current_post_actions(self) -> Optional[List[dict]]:
"""Get the post-actions for the current node.
Post-actions are executed after updating the LLM context when
transitioning to this node.
Returns:
List of post-actions to execute, or None if no post-actions
"""
return self.nodes[self.current_node].post_actions
def get_available_function_names(self) -> Set[str]:
"""Get the names of available functions for the current node.
Returns:
Set of function names that can be called from the current node
"""
names = {f["function"]["name"] for f in self.nodes[self.current_node].functions}
logger.debug(f"Available function names for node {self.current_node}: {names}")
return names
def transition(self, function_name: str) -> Optional[str]:
"""Attempt to transition to a new node based on a function call.
Handles both regular transitions (where the function name matches a node)
and terminal functions (which execute but don't change nodes).
Args:
function_name: Name of the function that was called
Returns:
The ID of the new node after transition, or None if transition failed.
For terminal functions, returns the current node ID.
"""
available_functions = self.get_available_function_names()
logger.debug(f"Attempting transition from {self.current_node} to {function_name}")
if function_name in available_functions:
if function_name in self.nodes:
# Regular transition to a new node
previous_node = self.current_node
self.current_node = function_name
logger.info(f"Transitioned from {previous_node} to node: {self.current_node}")
return self.current_node
else:
# Handle terminal function calls (functions that don't lead to new nodes)
logger.info(f"Executed terminal function: {function_name}")
return self.current_node
return None
def get_current_node(self) -> str:
"""Get the current node ID."""
return self.current_node