上下文

Context 对象是一个共享状态对象,它在 Session 中的所有 Rounds 之间存储对话状态。它用于维护对话的上下文以及对话的总体状态。

上下文属性

Context 对象的属性在 ContextNames 类中定义,该类是一个 EnumContextNames 类指定了在整个会话中使用的各种上下文属性。下面是其定义:

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_cost 属性 可写

获取当前轮次成本。

current_round_step 属性 可写

获取当前轮次步骤。

current_round_subtask_amount 属性 可写

获取当前轮次子任务索引。

add_to_structural_logs(data)

将数据添加到结构化日志。

参数
  • data (Dict[str, Any]) –

    要添加到结构化日志的数据。

源代码在 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)

从上下文中获取值。

参数
  • key (ContextNames) –

    上下文名称。

返回
  • Any

    从上下文中获取的值。

源代码在 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 ) –

    确保上下文可序列化。

返回
  • Dict[str, Any]

    上下文的字典。

源代码在 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.")