代理状态

State 类是 UFO 代理框架的基本组成部分。它表示代理的当前状态,并决定下一个要处理请求的动作和代理。每个代理都有一组特定的状态,这些状态定义了代理的行为和工作流程。

AgentStatus

代理的状态集在 AgentStatus 类中定义。

class AgentStatus(Enum):
    """
    The status class for the agent.
    """

    ERROR = "ERROR"
    FINISH = "FINISH"
    CONTINUE = "CONTINUE"
    FAIL = "FAIL"
    PENDING = "PENDING"
    CONFIRM = "CONFIRM"
    SCREENSHOT = "SCREENSHOT"

每个代理都实现自己的 AgentStatus 集来定义代理的状态。

AgentStateManager

AgentStateManager 类管理从字符串到相应状态类的状态映射。每个状态类都使用 register 装饰器向 AgentStateManager 注册,以将状态类与特定代理关联起来,例如:

@AgentStateManager.register
class SomeAgentState(AgentState):
    """
    The state class for the some agent.
    """

提示

您可以在 ufo/agents/states/app_agent_state.py 文件中找到如何为 AppAgent 注册状态类的示例。

下面是 AgentStateManager 类的基本结构:

class AgentStateManager(ABC, metaclass=SingletonABCMeta):
    """
    A abstract class to manage the states of the agent.
    """

    _state_mapping: Dict[str, Type[AgentState]] = {}

    def __init__(self):
        """
        Initialize the state manager.
        """

        self._state_instance_mapping: Dict[str, AgentState] = {}

    def get_state(self, status: str) -> AgentState:
        """
        Get the state for the status.
        :param status: The status string.
        :return: The state object.
        """

        # Lazy load the state class
        if status not in self._state_instance_mapping:
            state_class = self._state_mapping.get(status)
            if state_class:
                self._state_instance_mapping[status] = state_class()
            else:
                self._state_instance_mapping[status] = self.none_state

        state = self._state_instance_mapping.get(status, self.none_state)

        return state

    def add_state(self, status: str, state: AgentState) -> None:
        """
        Add a new state to the state mapping.
        :param status: The status string.
        :param state: The state object.
        """
        self.state_map[status] = state

    @property
    def state_map(self) -> Dict[str, AgentState]:
        """
        The state mapping of status to state.
        :return: The state mapping.
        """
        return self._state_instance_mapping

    @classmethod
    def register(cls, state_class: Type[AgentState]) -> Type[AgentState]:
        """
        Decorator to register the state class to the state manager.
        :param state_class: The state class to be registered.
        :return: The state class.
        """
        cls._state_mapping[state_class.name()] = state_class
        return state_class

    @property
    @abstractmethod
    def none_state(self) -> AgentState:
        """
        The none state of the state manager.
        """
        pass

AgentState

每个状态类都继承自 AgentState 类,并且必须实现 handle 方法来处理状态中的动作。此外,next_statenext_agent 方法用于确定下一个要处理转换的状态和代理。请参阅下面 UFO 中 State 类的参考。

基类:ABC

代理状态的抽象类。

agent_class() 抽象方法 类方法

代理的类。

返回
  • 类型[BasicAgent]

    代理的类。

源代码在 agents/states/basic.py
165
166
167
168
169
170
171
172
@classmethod
@abstractmethod
def agent_class(cls) -> Type[BasicAgent]:
    """
    The class of the agent.
    :return: The class of the agent.
    """
    pass

handle(agent, context=None) 抽象方法

处理当前步骤的代理。

参数
  • agent (BasicAgent) –

    要处理的代理。

  • context (可选['Context'], 默认值: None ) –

    代理和会话的上下文。

源代码在 agents/states/basic.py
122
123
124
125
126
127
128
129
@abstractmethod
def handle(self, agent: BasicAgent, context: Optional["Context"] = None) -> None:
    """
    Handle the agent for the current step.
    :param agent: The agent to handle.
    :param context: The context for the agent and session.
    """
    pass

is_round_end() 抽象方法

检查回合是否结束。

返回
  • 布尔值

    如果回合结束,则为 True,否则为 False。

源代码在 agents/states/basic.py
149
150
151
152
153
154
155
@abstractmethod
def is_round_end(self) -> bool:
    """
    Check if the round ends.
    :return: True if the round ends, False otherwise.
    """
    pass

is_subtask_end() 抽象方法

检查子任务是否结束。

返回
  • 布尔值

    如果子任务结束,则为 True,否则为 False。

源代码在 agents/states/basic.py
157
158
159
160
161
162
163
@abstractmethod
def is_subtask_end(self) -> bool:
    """
    Check if the subtask ends.
    :return: True if the subtask ends, False otherwise.
    """
    pass

name() 抽象方法 类方法

状态的类名。

返回
  • 字符串

    状态的类名。

源代码在 agents/states/basic.py
174
175
176
177
178
179
180
181
@classmethod
@abstractmethod
def name(cls) -> str:
    """
    The class name of the state.
    :return: The class name of the state.
    """
    return ""

next_agent(agent) 抽象方法

获取下一步的代理。

参数
  • agent (BasicAgent) –

    当前步骤的代理。

返回
  • BasicAgent

    下一步的代理。

源代码在 agents/states/basic.py
131
132
133
134
135
136
137
138
@abstractmethod
def next_agent(self, agent: BasicAgent) -> BasicAgent:
    """
    Get the agent for the next step.
    :param agent: The agent for the current step.
    :return: The agent for the next step.
    """
    return agent

next_state(agent) 抽象方法

获取下一步的状态。

参数
  • agent (BasicAgent) –

    当前步骤的代理。

返回
源代码在 agents/states/basic.py
140
141
142
143
144
145
146
147
@abstractmethod
def next_state(self, agent: BasicAgent) -> AgentState:
    """
    Get the state for the next step.
    :param agent: The agent for the current step.
    :return: The state for the next step.
    """
    pass

提示

HostAgentAppAgent 的状态机图显示在各自的文档中。

提示

Round 调用当前状态的 handlenext_statenext_agent 方法来处理用户请求并确定下一个要处理请求的状态和代理,并协调代理执行必要的动作。