Skip to main content

​
Overview

FlowsAgent extends BaseAgent with Pipecat Flows for structured conversation flows. It manages a FlowManager for node-based conversations with functions, transitions, and actions. On first activation the flow starts at build_initial_node(). Subsequent activations resume from build_resume_node().
Requires the flows extra: pip install pipecat-ai-subagents[flows]Imported from pipecat_subagents.agents.flows_agent:
from pipecat_subagents.agents.flows_agent import FlowsAgent

class MyFlowsAgent(FlowsAgent):
    def build_llm(self):
        return OpenAILLMService(api_key="...")

    def build_initial_node(self):
        return {"name": "start", ...}

​
Configuration

Inherits all parameters from BaseAgent.
​
name
str
required
Unique name for this agent.
​
bus
AgentBus
required
The AgentBus for inter-agent communication.
​
context_aggregator
Any
required
The context aggregator pair for managing LLM conversation context, forwarded to FlowManager.
​
context_strategy
Optional[ContextStrategyConfig]
default:"None"
Optional context strategy forwarded to FlowManager. See ContextStrategyConfig.
​
global_functions
Optional[List[FlowsFunctionSchema | FlowsDirectFunction]]
default:"None"
Optional list of functions available at every node, forwarded to FlowManager. Methods decorated with @tool are automatically included as global functions.
​
active
bool
default:"False"
Whether the agent starts active. Defaults to False.
​
bridged
Optional[tuple[str, ...]]
default:"()"
Bridge configuration. Defaults to an empty tuple (bridged, accepting all bridges).

​
Properties

Inherits all properties from BaseAgent.

​
flow_manager

agent.flow_manager -> Optional[FlowManager]
The FlowManager instance, available after the pipeline task is created.

​
Abstract Methods

​
build_llm

@abstractmethod
def build_llm(self) -> LLMService
Return the LLM service for this agent. Returns: An LLMService instance.

​
build_initial_node

@abstractmethod
def build_initial_node(self) -> NodeConfig
Return the initial flow node configuration. Called on first activation. Returns: A NodeConfig describing the first node of the flow.

​
Methods

​
build_resume_node

def build_resume_node(self) -> NodeConfig
Return the node to resume from when re-entering this agent. Called on subsequent activations (after the first). Override to resume from a specific point in the flow based on flow_manager.state. Defaults to restarting from the initial node. Returns: A NodeConfig for the resumption point.

​
on_activated

async def on_activated(self, args: Optional[dict]) -> None
Initialize or resume the flow on activation. On first call, initializes the flow at build_initial_node(). On subsequent calls, sets the node from build_resume_node().
ParameterTypeDescription
argsOptional[dict]Optional activation arguments