上下文
Context
对象是一个共享状态对象,它在 Session
中的所有 Rounds
之间存储对话状态。它用于维护对话的上下文以及对话的总体状态。
上下文属性
Context
对象的属性在 ContextNames
类中定义,该类是一个 Enum
。ContextNames
类指定了在整个会话中使用的各种上下文属性。下面是其定义:
class ContextNames(Enum):
"""
The context names.
"""
ID = "ID" # The ID of the session
MODE = "MODE" # The mode of the session
LOG_PATH = "LOG_PATH" # The folder path to store the logs
REQUEST = "REQUEST" # The current request
SUBTASK = "SUBTASK" # The current subtask processed by the AppAgent
PREVIOUS_SUBTASKS = "PREVIOUS_SUBTASKS" # The previous subtasks processed by the AppAgent
HOST_MESSAGE = "HOST_MESSAGE" # The message from the HostAgent sent to the AppAgent
REQUEST_LOGGER = "REQUEST_LOGGER" # The logger for the LLM request
LOGGER = "LOGGER" # The logger for the session
EVALUATION_LOGGER = "EVALUATION_LOGGER" # The logger for the evaluation
ROUND_STEP = "ROUND_STEP" # The step of all rounds
SESSION_STEP = "SESSION_STEP" # The step of the current session
CURRENT_ROUND_ID = "CURRENT_ROUND_ID" # The ID of the current round
APPLICATION_WINDOW = "APPLICATION_WINDOW" # The window of the application
APPLICATION_PROCESS_NAME = "APPLICATION_PROCESS_NAME" # The process name of the application
APPLICATION_ROOT_NAME = "APPLICATION_ROOT_NAME" # The root name of the application
CONTROL_REANNOTATION = "CONTROL_REANNOTATION" # The re-annotation of the control provided by the AppAgent
SESSION_COST = "SESSION_COST" # The cost of the session
ROUND_COST = "ROUND_COST" # The cost of all rounds
ROUND_SUBTASK_AMOUNT = "ROUND_SUBTASK_AMOUNT" # The amount of subtasks in all rounds
CURRENT_ROUND_STEP = "CURRENT_ROUND_STEP" # The step of the current round
CURRENT_ROUND_COST = "CURRENT_ROUND_COST" # The cost of the current round
CURRENT_ROUND_SUBTASK_AMOUNT = "CURRENT_ROUND_SUBTASK_AMOUNT" # The amount of subtasks in the current round
STRUCTURAL_LOGS = "STRUCTURAL_LOGS" # The structural logs of the session
每个属性都是一个字符串,表示会话上下文的特定方面,确保所有必要信息在应用程序中都是可访问和可管理的。
属性描述
属性 |
描述 |
ID |
会话的 ID。 |
MODE |
会话的模式。 |
LOG_PATH |
存储日志的文件夹路径。 |
REQUEST |
当前请求。 |
SUBTASK |
AppAgent 处理的当前子任务。 |
PREVIOUS_SUBTASKS |
AppAgent 处理的先前子任务。 |
HOST_MESSAGE |
HostAgent 发送给 AppAgent 的消息。 |
REQUEST_LOGGER |
LLM 请求的记录器。 |
LOGGER |
会话的记录器。 |
EVALUATION_LOGGER |
评估的记录器。 |
ROUND_STEP |
所有轮次的步骤。 |
SESSION_STEP |
当前会话的步骤。 |
CURRENT_ROUND_ID |
当前轮次的 ID。 |
APPLICATION_WINDOW |
应用程序的窗口。 |
APPLICATION_PROCESS_NAME |
应用程序的进程名称。 |
APPLICATION_ROOT_NAME |
应用程序的根名称。 |
CONTROL_REANNOTATION |
AppAgent 提供的控件的重新标注。 |
SESSION_COST |
会话的成本。 |
ROUND_COST |
所有轮次的成本。 |
ROUND_SUBTASK_AMOUNT |
所有轮次中的子任务数量。 |
CURRENT_ROUND_STEP |
当前轮次的步骤。 |
CURRENT_ROUND_COST |
当前轮次的成本。 |
CURRENT_ROUND_SUBTASK_AMOUNT |
当前轮次中的子任务数量。 |
STRUCTURAL_LOGS |
会话的结构化日志。 |
Context
对象的参考
维护会话和代理上下文的上下文类。
current_round_subtask_amount
属性
可写
add_to_structural_logs(data)
将数据添加到结构化日志。
源代码在 module/context.py
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291 | def add_to_structural_logs(self, data: Dict[str, Any]) -> None:
"""
Add data to the structural logs.
:param data: The data to add to the structural logs.
"""
round_key = data.get("Round", None)
subtask_key = data.get("SubtaskIndex", None)
if round_key is None or subtask_key is None:
return
remaining_items = {key: data[key] for key in data}
self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][subtask_key].append(
remaining_items
)
|
filter_structural_logs(round_key, subtask_key, keys)
过滤结构化日志。
参数 |
-
round_key (int ) –
-
subtask_key (int ) –
-
keys (Union[str, List[str]] ) –
|
返回 |
-
Union[List[Any], List[Dict[str, Any]]] –
|
源代码在 module/context.py
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313 | def filter_structural_logs(
self, round_key: int, subtask_key: int, keys: Union[str, List[str]]
) -> Union[List[Any], List[Dict[str, Any]]]:
"""
Filter the structural logs.
:param round_key: The round key.
:param subtask_key: The subtask key.
:param keys: The keys to filter.
:return: The filtered structural logs.
"""
structural_logs = self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][
subtask_key
]
if isinstance(keys, str):
return [log[keys] for log in structural_logs]
elif isinstance(keys, list):
return [{key: log[key] for key in keys} for log in structural_logs]
else:
raise TypeError(f"Keys should be a string or a list of strings.")
|
from_dict(context_dict)
从字典加载上下文。
参数 |
-
context_dict (Dict[str, Any] ) –
|
源代码在 module/context.py
340
341
342
343
344
345
346
347
348
349
350 | def from_dict(self, context_dict: Dict[str, Any]) -> None:
"""
Load the context from a dictionary.
:param context_dict: The dictionary of the context.
"""
for key in ContextNames:
if key.name in context_dict:
self._context[key.name] = context_dict.get(key.name)
# Sync the current round step and cost
self._sync_round_values()
|
get(key)
从上下文中获取值。
源代码在 module/context.py
167
168
169
170
171
172
173
174
175 | def get(self, key: ContextNames) -> Any:
"""
Get the value from the context.
:param key: The context name.
:return: The value from the context.
"""
# Sync the current round step and cost
self._sync_round_values()
return self._context.get(key.name)
|
set(key, value)
在上下文中设置值。
参数 |
-
key (ContextNames ) –
-
value (Any ) –
|
源代码在 module/context.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193 | def set(self, key: ContextNames, value: Any) -> None:
"""
Set the value in the context.
:param key: The context name.
:param value: The value to set in the context.
"""
if key.name in self._context:
self._context[key.name] = value
# Sync the current round step and cost
if key == ContextNames.CURRENT_ROUND_STEP:
self.current_round_step = value
if key == ContextNames.CURRENT_ROUND_COST:
self.current_round_cost = value
if key == ContextNames.CURRENT_ROUND_SUBTASK_AMOUNT:
self.current_round_subtask_amount = value
else:
raise KeyError(f"Key '{key}' is not a valid context name.")
|
to_dict(ensure_serializable=False)
将上下文转换为字典。
参数 |
-
ensure_serializable (bool , default: False ) –
|
源代码在 module/context.py
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338 | def to_dict(self, ensure_serializable: bool = False) -> Dict[str, Any]:
"""
Convert the context to a dictionary.
:param ensure_serializable: Ensure the context is serializable.
:return: The dictionary of the context.
"""
import copy
context_dict = copy.deepcopy(self._context)
if ensure_serializable:
for key in ContextNames:
if key.name in context_dict:
print_with_color(
f"Warn: The value of Context.{key.name} is not serializable.",
"yellow",
)
if not is_json_serializable(context_dict[key.name]):
context_dict[key.name] = None
return context_dict
|
update_dict(key, value)
向上下文键添加字典。值和上下文键都应该是字典。
参数 |
-
key (ContextNames ) –
-
value (Dict[str, Any] ) –
|
源代码在 module/context.py
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220 | def update_dict(self, key: ContextNames, value: Dict[str, Any]) -> None:
"""
Add a dictionary to a context key. The value and the context key should be dictionaries.
:param key: The context key to update.
:param value: The dictionary to add to the context key.
"""
if key.name in self._context:
context_value = self._context[key.name]
if isinstance(value, dict) and isinstance(context_value, dict):
self._context[key.name].update(value)
else:
raise TypeError(
f"Value for key '{key.name}' is {key.value}, requires a dictionary."
)
else:
raise KeyError(f"Key '{key.name}' is not a valid context name.")
|