代理状态
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_state
和 next_agent
方法用于确定下一个要处理转换的状态和代理。请参阅下面 UFO 中 State
类的参考。
基类:ABC
代理状态的抽象类。
agent_class()
抽象方法
类方法
代理的类。
返回 |
|
---|
源代码在 agents/states/basic.py
中
165 166 167 168 169 170 171 172 |
|
handle(agent, context=None)
抽象方法
处理当前步骤的代理。
参数 |
|
---|
源代码在 agents/states/basic.py
中
122 123 124 125 126 127 128 129 |
|
is_round_end()
抽象方法
检查回合是否结束。
返回 |
|
---|
源代码在 agents/states/basic.py
中
149 150 151 152 153 154 155 |
|
is_subtask_end()
抽象方法
检查子任务是否结束。
返回 |
|
---|
源代码在 agents/states/basic.py
中
157 158 159 160 161 162 163 |
|
name()
抽象方法
类方法
状态的类名。
返回 |
|
---|
源代码在 agents/states/basic.py
中
174 175 176 177 178 179 180 181 |
|
next_agent(agent)
抽象方法
获取下一步的代理。
参数 |
|
---|
返回 |
|
---|
源代码在 agents/states/basic.py
中
131 132 133 134 135 136 137 138 |
|
next_state(agent)
抽象方法
获取下一步的状态。
参数 |
|
---|
返回 |
|
---|
源代码在 agents/states/basic.py
中
140 141 142 143 144 145 146 147 |
|
提示
HostAgent
和 AppAgent
的状态机图显示在各自的文档中。
提示
Round
调用当前状态的 handle
、next_state
和 next_agent
方法来处理用户请求并确定下一个要处理请求的状态和代理,并协调代理执行必要的动作。