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

Class name updates and remove FrameProcessor base class

Browse Source
This commit is contained in:
Mark Backman
2024-11-15 13:19:05 -05:00
parent 1b74560f9d
commit 4e0ecdd673
5 changed files with 65 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,11 +14,11 @@ from loguru import logger
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
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.conversation_flow import ConversationFlowProcessor
from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -146,23 +146,6 @@ async def main():
}
]
# Register function handlers
async def handle_function_call(
function_name, tool_call_id, arguments, llm, context, result_callback
):
logger.info(f"Function called: {function_name} with arguments: {arguments}")
# Handle the state transition
await flow_processor.handle_transition(function_name)
# Send the acknowledgment
await result_callback("Acknowledged")
logger.info(f"Function call result sent: {function_name}")
# Register functions from all nodes
for node in flow_config["nodes"].values():
for function in node["functions"]:
function_name = function["function"]["name"]
llm.register_function(function_name, handle_function_call)
context = OpenAILLMContext(messages, initial_tools)
context_aggregator = llm.create_context_aggregator(context)
@@ -171,7 +154,6 @@ async def main():
transport.input(), # Transport user input
stt, # STT
context_aggregator.user(), # User responses
flow_processor, # Conversation flow management
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
@@ -181,17 +163,17 @@ async def main():
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
# Initialize conversation flow processor
flow_processor = ConversationFlowProcessor(flow_config, task)
# Initialize flow manager
flow_manager = FlowManager(flow_config, task)
# Register functions with LLM service
await flow_processor.register_functions(llm)
await flow_manager.register_functions(llm)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
# Initialize the flow processor
await flow_processor.initialize(messages)
await flow_manager.initialize(messages)
# Kick off the conversation using the context aggregator
await task.queue_frames([context_aggregator.user().get_context_frame()])

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

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

70
src/pipecat/processors/conversation_flow/processor.py → src/pipecat/flows/manager.py
View File

@@ -15,65 +15,66 @@ from pipecat.frames.frames import (
TTSSpeakFrame,
)
from .flow import ConversationFlow
from .state import FlowState
class ConversationFlowProcessor:
"""Processor that manages conversation flow based on function calls.
class FlowManager:
"""Manages conversation flows in a Pipecat pipeline.
This processor maintains conversation state and handles transitions between states
based on LLM function calls. Each state (node) has its own message, available
functions, and optional actions. The processor ensures the LLM context is updated
appropriately as the conversation progresses.
This manager handles the progression through a flow defined by nodes, where each node
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
The flow is defined by a configuration that specifies:
- Initial state
- Available states (nodes)
- Messages for each state
- Available functions for each state
- Optional actions for each state
- Initial node
- Available nodes and their configurations
- Transitions between nodes via function calls
"""
def __init__(self, flow_config: dict, task, **kwargs):
"""Initialize the conversation flow processor.
def __init__(self, flow_config: dict, task):
"""Initialize the flow manager.
Args:
flow_config: Dictionary containing the complete flow configuration,
including initial_node, nodes, and their configurations
including initial_node and node configurations
task: PipelineTask instance used to queue frames into the pipeline
"""
super().__init__()
self.flow = ConversationFlow(flow_config)
self.flow = FlowState(flow_config)
self.initialized = False
self.task = task
async def initialize(self, initial_messages: List[dict]):
"""Initialize the conversation with starting messages and functions.
"""Initialize the flow with starting messages and functions.
This method sets up the initial context for the conversation, combining
any system-level messages with the initial node's message 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
Note:
This must be called before the processor can handle any frames.
"""
if not self.initialized:
messages = initial_messages + [self.flow.get_current_message()]
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 conversation flow at node: {self.flow.current_node}")
logger.debug(f"Initialized flow at node: {self.flow.current_node}")
else:
logger.warning("Attempted to initialize ConversationFlowProcessor multiple times")
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 in the flow
configuration. When a function is called, it will automatically trigger the
appropriate state transition.
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
@@ -92,21 +93,22 @@ class ConversationFlowProcessor:
llm_service.register_function(function_name, handle_function_call)
async def handle_transition(self, function_name: str):
"""Handle state transition triggered by a function call.
"""Handle node transition triggered by a function call.
This method:
1. Validates the function call against available functions
2. Transitions to the new state if appropriate
3. Executes any actions associated with the new state
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(
"ConversationFlowProcessor must be initialized before handling transitions"
)
raise RuntimeError("FlowManager must be initialized before handling transitions")
available_functions = self.flow.get_available_function_names()
@@ -123,7 +125,7 @@ class ConversationFlowProcessor:
LLMSetToolsFrame(tools=self.flow.get_current_functions())
)
logger.debug(f"Transition to {new_node} complete")
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}'. "

25
src/pipecat/processors/conversation_flow/flow.py → src/pipecat/flows/state.py
View File

@@ -12,7 +12,10 @@ from loguru import logger
@dataclass
class NodeConfig:
"""Configuration for a single node in the conversation flow.
"""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:
message: Dict containing role and content for the LLM at this node
@@ -25,13 +28,13 @@ class NodeConfig:
actions: Optional[List[dict]] = None
class ConversationFlow:
"""Manages state transitions in a conversation flow.
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 state
(node) has its own message, available functions, and optional actions. It manages
transitions between states based on function calls and handles both regular and
terminal functions.
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.
Attributes:
nodes: Dictionary mapping node IDs to their configurations
@@ -108,7 +111,7 @@ class ConversationFlow:
return names
def transition(self, function_name: str) -> Optional[str]:
"""Attempt to transition based on a function call.
"""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).
@@ -117,8 +120,8 @@ class ConversationFlow:
function_name: Name of the function that was called
Returns:
The name of the new node after transition, or None if transition failed.
For terminal functions, returns the current node name.
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}")
@@ -130,7 +133,7 @@ class ConversationFlow:
logger.info(f"Transitioned to node: {self.current_node}")
return self.current_node
else:
# Handle terminal function calls
# 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

3
src/pipecat/processors/conversation_flow/__init__.py
View File

@@ -1,3 +0,0 @@
from .processor import ConversationFlowProcessor
__all__ = ["ConversationFlowProcessor"]