Skip to content

core / state / agent_state

core.state.agent_state

Global runtime state for a single-user, single-agent process.

AgentState dataclass

Authoritative runtime state for the agent.

Source code in core\state\agent_state.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class AgentState:
    """Authoritative runtime state for the agent."""

    conversation_state: Optional[str] = None
    current_task: Optional[Task] = None
    event_stream: Optional[str] = None
    gui_mode: bool = False
    agent_properties: AgentProperties = AgentProperties(current_task_id="", action_count=0, current_step_index=0)

    def update_conversation_state(self, new_state: str) -> None:
        self.conversation_state = new_state

    def update_current_task(self, new_task: Optional[Task]) -> None:
        self.current_task = new_task

    def update_event_stream(self, new_event_stream: Optional[str]) -> None:
        self.event_stream = new_event_stream

    def update_gui_mode(self, gui_mode: bool) -> None:
        self.gui_mode = gui_mode

    def refresh(
        self,
        *,
        conversation_state: Optional[str] = None,
        current_task: Optional[Task] = None,
        event_stream: Optional[str] = None,
        gui_mode: Optional[bool] = None,
    ) -> None:
        """Update only fields that changed."""
        self.conversation_state = conversation_state
        self.current_task = current_task
        self.event_stream = event_stream
        self.gui_mode = gui_mode

    def set_agent_property(self, key, value):
        """
        Sets a global agent property (not specific to any task).
        """
        self.agent_properties.set_property(key, value)

    def get_agent_property(self, key, default=None):
        """
        Retrieves a global agent property.
        """
        return self.agent_properties.get_property(key, default)

    def get_agent_properties(self):
        """
        Retrieves all global agent properties.
        """
        return self.agent_properties.to_dict()

refresh(*, conversation_state=None, current_task=None, event_stream=None, gui_mode=None)

Update only fields that changed.

Source code in core\state\agent_state.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
def refresh(
    self,
    *,
    conversation_state: Optional[str] = None,
    current_task: Optional[Task] = None,
    event_stream: Optional[str] = None,
    gui_mode: Optional[bool] = None,
) -> None:
    """Update only fields that changed."""
    self.conversation_state = conversation_state
    self.current_task = current_task
    self.event_stream = event_stream
    self.gui_mode = gui_mode

set_agent_property(key, value)

Sets a global agent property (not specific to any task).

Source code in core\state\agent_state.py 
45
46
47
48
49
def set_agent_property(self, key, value):
    """
    Sets a global agent property (not specific to any task).
    """
    self.agent_properties.set_property(key, value)

get_agent_property(key, default=None)

Retrieves a global agent property.

Source code in core\state\agent_state.py 
51
52
53
54
55
def get_agent_property(self, key, default=None):
    """
    Retrieves a global agent property.
    """
    return self.agent_properties.get_property(key, default)

get_agent_properties()

Retrieves all global agent properties.

Source code in core\state\agent_state.py 
57
58
59
60
61
def get_agent_properties(self):
    """
    Retrieves all global agent properties.
    """
    return self.agent_properties.to_dict()