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

Add pre- and post-actions

Browse Source
This commit is contained in:
Mark Backman
2024-11-15 13:51:43 -05:00
parent 686165b95a
commit 5301f44b3b
3 changed files with 89 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -81,12 +81,12 @@ flow_config = {
},
}
],
"actions": [{"type": "tts_say", "text": "Let me help you order a pizza..."}],
"pre_actions": [{"type": "tts_say", "text": "Ok, let me pull up our pizza menu..."}],
},
"choose_sushi": {
"message": {
"role": "system",
"content": "The user has chosen sushi. Immediately ask them: 'How many sushi rolls would you like to order?' If they answer provide to the question of how many rolls, use the select_roll_count function.",
"content": "The user has chosen sushi. Immediately say: 'How many sushi rolls would you like to order?' If they answer provide to the question of how many rolls, use the select_roll_count function.",
},
"functions": [
{
@@ -109,7 +109,7 @@ flow_config = {
},
}
],
"actions": [{"type": "tts_say", "text": "Ok, one moment..."}],
"pre_actions": [{"type": "tts_say", "text": "Ok, one moment..."}],
},
},
}
@@ -164,7 +164,7 @@ async def main():
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
# Initialize flow manager
flow_manager = FlowManager(flow_config, task)
flow_manager = FlowManager(flow_config, task, tts)
# Register functions with LLM service
await flow_manager.register_functions(llm)

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

@@ -26,7 +26,8 @@ class FlowManager:
represents a state in the conversation. Each node has:
- A message for the LLM
- Available functions that can be called
- Optional actions to execute when entering the node
- 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
@@ -34,7 +35,7 @@ class FlowManager:
- Transitions between nodes via function calls
"""
def __init__(self, flow_config: dict, task):
def __init__(self, flow_config: dict, task, tts=None):
"""Initialize the flow manager.
Args:
@@ -45,6 +46,7 @@ class FlowManager:
self.flow = FlowState(flow_config)
self.initialized = False
self.task = task
self.tts = tts
self.action_handlers: Dict[str, Callable] = {}
# Register built-in actions
@@ -96,46 +98,6 @@ class FlowManager:
function_name = function["function"]["name"]
llm_service.register_function(function_name, handle_function_call)
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 actions associated with the new node
4. Updates the LLM context with new messages and available functions
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()
if function_name in available_functions:
new_node = self.flow.transition(function_name)
if new_node:
if self.flow.get_current_actions():
await self._execute_actions(self.flow.get_current_actions())
current_message = self.flow.get_current_message()
await self.task.queue_frame(LLMMessagesAppendFrame(messages=[current_message]))
await self.task.queue_frame(
LLMSetToolsFrame(tools=self.flow.get_current_functions())
)
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}"
)
def register_action(self, action_type: str, handler: Callable):
"""Register a handler for a specific action type.
@@ -175,5 +137,58 @@ class FlowManager:
logger.warning(f"No handler registered for action type: {action_type}")
async def _handle_tts_action(self, action: dict):
"""Built-in handler for tts_say actions"""
await self.task.queue_frame(TTSSpeakFrame(text=action["text"]))
"""Built-in handler for TTS actions"""
if self.tts:
# Direct call to TTS service to speak text
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_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()
if function_name in available_functions:
new_node = self.flow.transition(function_name)
if new_node:
# Execute pre-actions before updating LLM context
if 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_message = self.flow.get_current_message()
await self.task.queue_frame(LLMMessagesAppendFrame(messages=[current_message]))
await self.task.queue_frame(
LLMSetToolsFrame(tools=self.flow.get_current_functions())
)
# Execute post-actions after updating LLM context
if 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}"
)

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

@@ -20,12 +20,14 @@ class NodeConfig:
Attributes:
message: Dict containing role and content for the LLM at this node
functions: List of available function definitions for this node
actions: Optional list of actions to execute when entering this node
pre_actions: Optional list of actions to execute before LLM inference
post_actions: Optional list of actions to execute after LLM inference
"""
message: dict
functions: List[dict]
actions: Optional[List[dict]] = None
pre_actions: Optional[List[dict]] = None
post_actions: Optional[List[dict]] = None
class FlowState:
@@ -33,8 +35,8 @@ class FlowState:
This class handles the state machine logic for conversation flows, where each node
represents a distinct state with its own message, available functions, and optional
actions. It manages transitions between nodes based on function calls and handles
both regular and terminal functions.
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
@@ -73,7 +75,8 @@ class FlowState:
self.nodes[node_id] = NodeConfig(
message=node_config["message"],
functions=node_config["functions"],
actions=node_config.get("actions"),
pre_actions=node_config.get("pre_actions"),
post_actions=node_config.get("post_actions"),
)
def get_current_message(self) -> dict:
@@ -92,13 +95,27 @@ class FlowState:
"""
return self.nodes[self.current_node].functions
def get_current_actions(self) -> Optional[List[dict]]:
"""Get the actions for the current node.
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 actions to execute when entering the node, or None if no actions
List of pre-actions to execute, or None if no pre-actions
"""
return self.nodes[self.current_node].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.