agentscope.utils

Import modules in utils package.

agentscope.utils.setup_logger(path_log: str | None = None, level: Literal['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO') → None[源代码]

Setup loguru.logger and redirect stderr to logging.

参数:

path_log (str, defaults to “”) – The directory of log files.
level (str, defaults to “INFO”) – The logging level, which is one of the following: “TRACE”, “DEBUG”, “INFO”, “SUCCESS”, “WARNING”, “ERROR”, “CRITICAL”.

class agentscope.utils.MonitorBase[源代码]

基类：ABC

Base interface of Monitor

abstract register(metric_name: str, metric_unit: str | None = None, quota: float | None = None) → bool[源代码]

Register a metric to the monitor with value initialized to 0.

参数:

metric_name (str) – Name of the metric, must be unique.
metric_unit (Optional[str]) – Unit of the metric.
quota (Optional[str]) – The quota of the metric. An alert is triggered when metrics accumulate above this value.

返回:

whether the operation success.

返回类型:

bool

abstract exists(metric_name: str) → bool[源代码]

Determine whether a metric exists in the monitor.

参数:: metric_name (str) – Name of the metric.
返回:: Whether the metric exists.
返回类型:: bool

abstract add(metric_name: str, value: float) → bool[源代码]

Add value to a specific metric.

参数:

metric_name (str) – Name of the metric.
value (float) – Increased value.

返回:

whether the operation success.

返回类型:

bool

update(values: dict, prefix: str | None = None) → None[源代码]: Update multiple metrics at once.

abstract clear(metric_name: str) → bool[源代码]

Clear the values of a specific metric.

参数:: metric_name (str) – Name of the metric.
返回:: whether the operation success.
返回类型:: bool

abstract remove(metric_name: str) → bool[源代码]

Remove a specific metric from the monitor.

参数:: metric_name (str) – Name of the metric.
返回:: Whether the operation success.
返回类型:: bool

abstract get_value(metric_name: str) → float | None[源代码]

Get the value of a specific metric.

参数:: metric_name (str) – Name of the metric.
返回:: the value of the metric.
返回类型:: Optional[float]

abstract get_unit(metric_name: str) → str | None[源代码]

Get the unit of a specific metric.

参数:: metric_name (str) – Name of the metric.
返回:: The unit of the metric.
返回类型:: Optional[str]

abstract get_quota(metric_name: str) → float | None[源代码]

Get the quota of a specific metric.

参数:: metric_name (str) – Name of the metric.
返回:: The quota of the metric.
返回类型:: Optional[float]

abstract set_quota(metric_name: str, quota: float) → bool[源代码]

Set the quota of a specific metric

参数:

metric_name (str) – Name of the metric.
quota (float) – New quota of the metric.

返回:

whether the operation success.

返回类型:

bool

abstract get_metric(metric_name: str) → dict | None[源代码]

Get the specific metric

参数:

metric_name (str) – Name of the metric.

返回:

A dictionary of metric with following format:

{
    metric_value: [float],
    metric_unit: [str],
    quota: [float]
}

返回类型:

Optional[dict]

abstract get_metrics(filter_regex: str | None = None) → dict[源代码]

Get a dictionary of metrics.

参数:

filter_regex (Optional[str]) – Regular expression for filtering metric names, get all metrics if not provided.

返回:

a dictionary of metric with following format:

{
    metric_name_A: {
        metric_value: [float],
        metric_unit: [str],
        quota: [float]
    },
    metric_name_B: {
        ...
    },
    ...
}

返回类型:

dict

abstract register_budget(model_name: str, value: float, prefix: str | None = 'local') → bool[源代码]

Register model call budget to the monitor, the monitor will raise QuotaExceededError, when budget is exceeded.

参数:

model_name (str) – model that requires budget.
value (float) – the budget value.
prefix (Optional[str], default None) – used to distinguish multiple budget registrations. For multiple registrations with the same prefix, only the first time will take effect.

返回:

whether the operation success.

返回类型:

bool

exception agentscope.utils.QuotaExceededError(name: str)[源代码]

基类：Exception

An Exception used to indicate that a certain metric exceeds quota

__init__(name: str) → None[源代码]

Init a QuotaExceedError instance.

参数:: name (str) – name of the metric which exceeds quota.

class agentscope.utils.MonitorFactory[源代码]

基类：object

Factory of Monitor.

Get the singleton monitor using:

from agentscope.utils import MonitorFactory
monitor = MonitorFactory.get_monitor()

classmethod get_monitor(impl_type: str | None = None, db_path: str = 'agentscope.db') → MonitorBase[源代码]

Get the monitor instance.

参数:

impl_type (Optional[str], optional) – the type of monitor, currently supports sqlite only.
db_path (Optional[str], optional) – path to the sqlite db file.

返回:

the monitor instance.

返回类型:

MonitorBase

classmethod flush() → None[源代码]: Only for unittest usage. Don’t use this function in your code. Flush the monitor singleton.