chatflow.state
State management for agents, teams and termination conditions.
- class agentopera.chatflow.state.BaseState(*, type: str = 'BaseState', version: str = '1.0.0')[source]
Bases:
BaseModelBase class for all saveable state
- type: str
- version: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.AssistantAgentState(*, type: str = 'AssistantAgentState', version: str = '1.0.0', llm_context: ~typing.Mapping[str, ~typing.Any] = <factory>)[source]
Bases:
BaseStateState for an assistant agent.
- llm_context: Mapping[str, Any]
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.BaseGroupChatManagerState(*, type: str = 'BaseGroupChatManagerState', version: str = '1.0.0', message_thread: ~typing.List[~agentopera.chatflow.messages.ToolCallRequestEvent | ~agentopera.chatflow.messages.ToolCallExecutionEvent | ~agentopera.chatflow.messages.MemoryQueryEvent | ~agentopera.chatflow.messages.UserInputRequestedEvent | ~agentopera.chatflow.messages.ModelClientStreamingChunkEvent | ~agentopera.chatflow.messages.ThoughtEvent | ~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>, current_turn: int = 0)[source]
Bases:
BaseStateBase state for all group chat managers.
- message_thread: List[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent | TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage | VercelMessage]
- current_turn: int
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.ChatAgentContainerState(*, type: str = 'ChatAgentContainerState', version: str = '1.0.0', agent_state: ~typing.Mapping[str, ~typing.Any] = <factory>, message_buffer: ~typing.List[~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>)[source]
Bases:
BaseStateState for a container of chat agents.
- agent_state: Mapping[str, Any]
- message_buffer: List[TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage | VercelMessage]
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.RoundRobinManagerState(*, type: str = 'RoundRobinManagerState', version: str = '1.0.0', message_thread: ~typing.List[~agentopera.chatflow.messages.ToolCallRequestEvent | ~agentopera.chatflow.messages.ToolCallExecutionEvent | ~agentopera.chatflow.messages.MemoryQueryEvent | ~agentopera.chatflow.messages.UserInputRequestedEvent | ~agentopera.chatflow.messages.ModelClientStreamingChunkEvent | ~agentopera.chatflow.messages.ThoughtEvent | ~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>, current_turn: int = 0, next_speaker_index: int = 0)[source]
Bases:
BaseGroupChatManagerStateState for
RoundRobinGroupChatmanager.- next_speaker_index: int
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.SelectorManagerState(*, type: str = 'SelectorManagerState', version: str = '1.0.0', message_thread: ~typing.List[~agentopera.chatflow.messages.ToolCallRequestEvent | ~agentopera.chatflow.messages.ToolCallExecutionEvent | ~agentopera.chatflow.messages.MemoryQueryEvent | ~agentopera.chatflow.messages.UserInputRequestedEvent | ~agentopera.chatflow.messages.ModelClientStreamingChunkEvent | ~agentopera.chatflow.messages.ThoughtEvent | ~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>, current_turn: int = 0, previous_speaker: str | None = None)[source]
Bases:
BaseGroupChatManagerStateState for
SelectorGroupChatmanager.- previous_speaker: str | None
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.SwarmManagerState(*, type: str = 'SwarmManagerState', version: str = '1.0.0', message_thread: ~typing.List[~agentopera.chatflow.messages.ToolCallRequestEvent | ~agentopera.chatflow.messages.ToolCallExecutionEvent | ~agentopera.chatflow.messages.MemoryQueryEvent | ~agentopera.chatflow.messages.UserInputRequestedEvent | ~agentopera.chatflow.messages.ModelClientStreamingChunkEvent | ~agentopera.chatflow.messages.ThoughtEvent | ~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>, current_turn: int = 0, current_speaker: str = '')[source]
Bases:
BaseGroupChatManagerStateState for
Swarmmanager.- current_speaker: str
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.MagenticOneOrchestratorState(*, type: str = 'MagenticOneOrchestratorState', version: str = '1.0.0', message_thread: ~typing.List[~agentopera.chatflow.messages.ToolCallRequestEvent | ~agentopera.chatflow.messages.ToolCallExecutionEvent | ~agentopera.chatflow.messages.MemoryQueryEvent | ~agentopera.chatflow.messages.UserInputRequestedEvent | ~agentopera.chatflow.messages.ModelClientStreamingChunkEvent | ~agentopera.chatflow.messages.ThoughtEvent | ~agentopera.chatflow.messages.TextMessage | ~agentopera.chatflow.messages.MultiModalMessage | ~agentopera.chatflow.messages.StopMessage | ~agentopera.chatflow.messages.ToolCallSummaryMessage | ~agentopera.chatflow.messages.HandoffMessage | ~agentopera.engine.types.models.types.VercelMessage] = <factory>, current_turn: int = 0, task: str = '', facts: str = '', plan: str = '', n_rounds: int = 0, n_stalls: int = 0)[source]
Bases:
BaseGroupChatManagerStateState for
MagneticOneGroupChatorchestrator.- task: str
- facts: str
- plan: str
- n_rounds: int
- n_stalls: int
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.TeamState(*, type: str = 'TeamState', version: str = '1.0.0', agent_states: ~typing.Mapping[str, ~typing.Any] = <factory>, team_id: str = '')[source]
Bases:
BaseStateState for a team of agents.
- agent_states: Mapping[str, Any]
- team_id: str
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class agentopera.chatflow.state.SocietyOfMindAgentState(*, type: str = 'SocietyOfMindAgentState', version: str = '1.0.0', inner_team_state: ~typing.Mapping[str, ~typing.Any] = <factory>)[source]
Bases:
BaseStateState for a Society of Mind agent.
- inner_team_state: Mapping[str, Any]
- type: str
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].