代理状态
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 方法来处理用户请求并确定下一个要处理请求的状态和代理,并协调代理执行必要的动作。