camel.utils 包

本页内容

camel.utils 包#

子模块#

camel.utils.async_func 模块#

camel.utils.async_func.sync_funcs_to_async(funcs: list[FunctionTool]) → list[FunctionTool][来源]#

将一系列Python同步函数转换为Python异步函数。

Parameters:

funcs (list[FunctionTool]) – Python同步函数的列表，采用FunctionTool格式。

Returns:

Python异步函数列表: 以FunctionTool格式呈现。

Return type:

列表[FunctionTool]

camel.utils.commons 模块#

class camel.utils.commons.AgentOpsMeta(name, bases, dct)[来源]#

基类: type

元类，自动使用agentops_decorator装饰所有可调用属性，除了'get_tools'方法。

方法： __new__(cls, name, bases, dct):

创建一个带有装饰方法的新类。

class camel.utils.commons.BatchProcessor(max_workers: int | None = None, initial_batch_size: int | None = None, monitoring_interval: float = 5.0, cpu_threshold: float = 80.0, memory_threshold: float = 85.0)[来源]#

基类：object

基于系统负载动态调整批量处理大小并具备错误处理能力。

adjust_batch_size(success: bool, processing_time: float | None = None) → None[来源]#

根据成功/失败情况和系统资源调整批量大小。

Parameters:

success (bool) - 上一批次是否成功完成
processing_time (Optional[float]) – 处理最后一批数据所花费的时间。(默认: None)

get_performance_metrics() → Dict[str, Any][来源]#

获取当前性能指标。

Returns:

total_processed: 已处理的总批次数量
error_rate: 失败批次的百分比
avg_processing_time: 每批次的平均处理时间
current_batch_size: 当前批次大小
current_workers: 当前工作线程数量
current_cpu: 当前CPU使用率百分比
current_memory: 当前内存使用率百分比

Return type:

包含性能指标的字典

camel.utils.commons.agentops_decorator(func)[来源]#

如果ToolEvent可用，记录函数执行的装饰器。

Parameters:: func (可调用对象) – 需要被装饰的函数。
Returns:: 封装后的函数会记录其执行细节。
Return type:: 可调用

camel.utils.commons.api_keys_required(param_env_list: List[Tuple[str | None, str]]) → Callable[[F], F][来源]#

一个装饰器，用于检查环境变量或函数参数中是否提供了所需的API密钥。

Parameters:

param_env_list (List[Tuple[Optional[str], str]]) - 一个元组列表，其中每个元组包含一个函数参数名称（作为第一个元素，或为None）和保存API密钥的对应环境变量名称（作为第二个元素）。

Returns:

原始函数经过包装并添加了检查: 用于验证必需的API密钥。

Return type:

Callable[[F], F]

Raises:

ValueError – 如果缺少任何必需的API密钥，无论是来自函数参数还是环境变量。

示例

@api_keys_required([
    ('api_key_arg', 'API_KEY_1'),
    ('another_key_arg', 'API_KEY_2'),
    (None, 'API_KEY_3'),
])
def some_api_function(api_key_arg=None, another_key_arg=None):
    # Function implementation that requires API keys

camel.utils.commons.check_server_running(server_url: str) → bool[来源]#

检查URL指向的服务器的端口是否开放。

Parameters:: server_url (str) – 运行LLM推理服务的服务器URL。
Returns:: 端口是否开放以接收数据包（服务器正在运行）。
Return type:: 布尔值

camel.utils.commons.create_chunks(text: str, n: int) → List[str][来源]#

从提供的文本中返回连续的n大小块。将文本分割成大小为n的较小块。

Parameters:

text (str) – 需要分割的文本。
n (int) – 单个块的最大长度。

Returns:

分割文本的列表。

Return type:

List[str]

camel.utils.commons.dependencies_required(*required_modules: str) → Callable[[F], F][来源]#

一个装饰器，用于确保在函数执行前指定的Python模块可用。

Parameters:

required_modules (str) – 需要检查可用性的必需模块。

Returns:

原始函数增加了对: 必需模块依赖项的检查。

Return type:

Callable[[F], F]

Raises:

ImportError – 如果任何必需的模块不可用。

示例

@dependencies_required('numpy', 'pandas')
def data_processing_function():
    # Function implementation...

camel.utils.commons.download_github_subdirectory(repo: str, subdir: str, data_dir: Path, branch='main')[来源]#

下载基准测试Github仓库的子目录。

此函数从GitHub仓库的指定子目录下载所有文件和子目录，并将它们保存到本地目录。

Parameters:

repo (str) – GitHub仓库名称，格式为"owner/repo"。
subdir (str) – 仓库内要下载的子目录路径。
data_dir (Path) – 文件将被保存的本地目录。
branch (str, optional) – 要使用的仓库分支。默认为"main"。

camel.utils.commons.download_tasks(task: 任务类型, folder_path: str) → None[来源]#

从指定URL下载任务相关文件并解压。

此函数从预定义的URL下载包含基于指定task类型任务的zip文件，将其保存到folder_path，然后将zip文件内容解压到同一文件夹。解压完成后，zip文件会被删除。

Parameters:

task (TaskType) - 一个枚举，表示要下载的任务类型。
folder_path (str) – zip文件将被下载并解压到的文件夹路径。

camel.utils.commons.func_string_to_callable(code: str)[来源]#

将函数代码字符串转换为可调用的函数对象。

Parameters:

code (str) – 作为字符串形式的函数代码。

Returns:

从代码字符串中提取的可调用函数对象: 代码字符串。

Return type:

可调用[…, 任意类型]

camel.utils.commons.generate_prompt_for_structured_output(response_format: Type[BaseModel] | None, user_message: str) → str[来源]#

该函数根据提供的Pydantic模型和用户消息生成提示。

Parameters:

response_format (Type[BaseModel]) – Pydantic模型类。
user_message (str) – 提示中要使用的用户消息。

Returns:

用于大语言模型的提示字符串。

Return type:

字符串

camel.utils.commons.get_first_int(string: str) → int | None[来源]#

返回给定字符串中找到的第一个整数。

如果没有找到整数，则返回 None。

Parameters:

字符串 (str) – 输入字符串。

Returns:

字符串中找到的第一个整数，如果没有找到则返回None: 未找到任何整数。

Return type:

整数或无

camel.utils.commons.get_prompt_template_key_words(template: str) → Set[str][来源]#

给定一个包含大括号{}的字符串模板，返回大括号内单词的集合。

Parameters:: template (str) – 一个包含大括号的字符串。
Returns:: 大括号内单词的列表。
Return type:: List[str]

示例

>>> get_prompt_template_key_words('Hi, {name}! How are you {status}?')
{'name', 'status'}

camel.utils.commons.get_pydantic_major_version() → int[来源]#

获取Pydantic的主版本号。

Returns:: 如果安装了Pydantic，则显示其主版本号，否则显示0。
Return type:: int

camel.utils.commons.get_pydantic_object_schema(pydantic_params: Type[BaseModel]) → Dict[来源]#

获取Pydantic模型的JSON模式。

Parameters:: pydantic_params (Type[BaseModel]) – 用于获取模式的Pydantic模型类。
Returns:: Pydantic模型的JSON模式。
Return type:: 字典

camel.utils.commons.get_system_information()[来源]#

收集有关操作系统的信息。

Returns:: 一个包含各种操作系统信息的字典。
Return type:: 字典

camel.utils.commons.get_task_list(task_response: str) → List[str][来源]#

解析Agent的响应并返回任务列表。

Parameters:: task_response (str) – Agent的字符串响应。
Returns:: 字符串任务列表。
Return type:: List[str]

camel.utils.commons.handle_http_error(response: Response) → str[来源]#

根据响应的状态码处理HTTP错误。

Parameters:: response (requests.Response) – 来自API调用的HTTP响应。
Returns:: 错误类型，基于状态码。
Return type:: 字符串

camel.utils.commons.is_docker_running() → bool[来源]#

检查Docker守护进程是否正在运行。

Returns:: 如果Docker守护进程正在运行则为True，否则为False。
Return type:: 布尔值

camel.utils.commons.is_module_available(module_name: str) → bool[来源]#

检查模块是否可用于导入。

Parameters:: module_name (str) – 要检查可用性的模块名称。
Returns:: 如果模块可以导入则为True，否则为False。
Return type:: 布尔值

camel.utils.commons.json_to_function_code(json_obj: Dict) → str[来源]#

根据JSON模式生成Python函数代码。

Parameters:: json_obj (dict) - 包含属性和必填字段的JSON模式对象，其json格式遵循openai工具模式
Returns:: 生成的Python函数代码作为字符串。
Return type:: 字符串

camel.utils.commons.print_text_animated(text, delay: float = 0.02, end: str = '')[来源]#

以动画效果显示给定的文本。

Parameters:

text (str) – 要打印的文本。
delay (float, optional) – 每个字符打印之间的延迟时间。 (默认值: 0.02)
end (str, optional) – 在每个文本字符后打印的结束字符。(默认: "")

camel.utils.commons.retry_on_error(max_retries: int = 3, initial_delay: float = 1.0) → Callable[来源]#

装饰器，用于在异常发生时以指数退避方式重试函数调用。

Parameters:

max_retries (int) - 最大重试次数
initial_delay (float) - 初始重试延迟时间（秒）

Returns:

带有重试逻辑的装饰函数

Return type:

可调用

camel.utils.commons.text_extract_from_web(url: str) → str[来源]#

从给定URL获取文本信息。

Parameters:: url (str) – 您想要搜索的网站。
Returns:: 所有文本均从网页中提取。
Return type:: 字符串

camel.utils.commons.to_pascal(snake: str) → str[来源]#

将snake_case字符串转换为PascalCase。

Parameters:: snake (str) – 需要转换的snake_case格式字符串。
Returns:: 转换后的PascalCase字符串。
Return type:: 字符串

camel.utils.commons.track_agent(*args, **kwargs)[来源]#: AgentOps的模拟追踪代理装饰器。

camel.utils.commons.with_timeout(timeout=None)[来源]#

为函数添加超时功能的装饰器。

以指定的超时值执行函数。如果执行时间超过限制，则返回超时消息。

Parameters:: timeout (float, optional) – 超时时长，单位为秒。如果为None，将尝试从实例的timeout属性获取超时时间。 (默认: None)

示例

>>> @with_timeout(5)
... def my_function():
...     return "Success"
>>> my_function()

>>> class MyClass:
...     timeout = 5
...     @with_timeout()
...     def my_method(self):
...         return "Success"

camel.utils.constants 模块#

class camel.utils.constants.Constants[来源]#

基类：object

一个包含CAMEL中使用常量的类。

DEFAULT_SIMILARITY_THRESHOLD = 0.7#

DEFAULT_TOP_K_RESULTS = 1#

FUNC_NAME_FOR_STRUCTURED_OUTPUT = 'return_json_response'#

VIDEO_DEFAULT_IMAGE_SIZE = 768#

VIDEO_DEFAULT_PLUG_PYAV = 'pyav'#

VIDEO_IMAGE_EXTRACTION_INTERVAL = 50#

camel.utils.token_counting 模块#

class camel.utils.token_counting.AnthropicTokenCounter(model: str)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用针对此类模型加载的特定分词器，计算所提供消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为令牌ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.token_counting.BaseTokenCounter[来源]#

基类: ABC

不同种类模型的token计数器的基类。

abstract count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

计算提供的消息列表中的令牌数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

abstract decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

abstract encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.token_counting.LiteLLMTokenCounter(model_type: UnifiedModelType)[来源]#

基类：BaseTokenCounter

calculate_cost_from_response(response: dict) → float[来源]#

计算给定完成响应的成本。

Parameters:: response (dict) – 来自LiteLLM的完成响应。
Returns:: 完成调用的费用（以美元计）。
Return type:: 浮点数

property completion_cost#

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用特定于此类型模型的标记器计算提供的消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 采用LiteLLM API格式的聊天历史消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为令牌ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

property token_counter#

class camel.utils.token_counting.MistralTokenCounter(model_type: ~<unknown>.ModelType)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用针对此类模型加载的特定分词器，计算所提供消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的总令牌数。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.token_counting.OpenAITokenCounter(model: UnifiedModelType)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

借助tiktoken包计算提供的消息列表中的令牌数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

camel.utils.token_counting.get_model_encoding(value_for_tiktoken: str)[来源]#

从tiktoken获取模型编码。

Parameters:: value_for_tiktoken – tiktoken的模型值。
Returns:: 模型编码。
Return type:: tiktoken.Encoding

模块内容#

class camel.utils.AgentOpsMeta(name, bases, dct)[来源]#

基类: type

元类，自动使用agentops_decorator装饰所有可调用属性，除了'get_tools'方法。

方法： __new__(cls, name, bases, dct):

创建一个带有装饰方法的新类。

class camel.utils.AnthropicTokenCounter(model: str)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用针对此类模型加载的特定分词器，计算所提供消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.BaseTokenCounter[来源]#

基类: ABC

不同种类模型的token计数器的基类。

abstract count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

计算提供的消息列表中的令牌数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

abstract decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

abstract encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.BatchProcessor(max_workers: int | None = None, initial_batch_size: int | None = None, monitoring_interval: float = 5.0, cpu_threshold: float = 80.0, memory_threshold: float = 85.0)[来源]#

基类：object

基于系统负载动态调整批量处理大小并具备错误处理能力。

adjust_batch_size(success: bool, processing_time: float | None = None) → None[来源]#

根据成功/失败情况和系统资源调整批量大小。

Parameters:

success (bool) - 上一批次是否成功完成
processing_time (Optional[float]) – 处理最后一批数据所花费的时间。(默认: None)

get_performance_metrics() → Dict[str, Any][来源]#

获取当前性能指标。

Returns:

total_processed: 已处理的总批次数量
error_rate: 失败批次的百分比
avg_processing_time: 每批次的平均处理时间
current_batch_size: 当前批次大小
current_workers: 当前工作线程数量
current_cpu: 当前CPU使用率百分比
current_memory: 当前内存使用率百分比

Return type:

包含性能指标的字典

class camel.utils.Constants[来源]#

基类：object

一个包含CAMEL中使用常量的类。

DEFAULT_SIMILARITY_THRESHOLD = 0.7#

DEFAULT_TOP_K_RESULTS = 1#

FUNC_NAME_FOR_STRUCTURED_OUTPUT = 'return_json_response'#

VIDEO_DEFAULT_IMAGE_SIZE = 768#

VIDEO_DEFAULT_PLUG_PYAV = 'pyav'#

VIDEO_IMAGE_EXTRACTION_INTERVAL = 50#

class camel.utils.DeduplicationResult(*, original_texts: List[str], unique_ids: List[int], unique_embeddings_dict: Dict[int, List[float]], duplicate_to_target_map: Dict[int, int])[来源]#

基类：BaseModel

去重后的结果。

original_texts#

原始文本。

Type:: List[str]

unique_ids#

一组唯一(不重复)的ID列表。

Type:: 整数列表

unique_embeddings_dict#

从每个唯一文本的索引到其嵌入向量的映射。

Type:: 字典[int, 列表[float]]

duplicate_to_target_map#

一个从重复文本索引到其所对应的原始文本索引的映射关系。

Type:: 字典[int, int]

duplicate_to_target_map: Dict[int, int]#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

original_texts: List[str]#

unique_embeddings_dict: Dict[int, List[float]]#

unique_ids: List[int]#

class camel.utils.LiteLLMTokenCounter(model_type: UnifiedModelType)[来源]#

基类：BaseTokenCounter

calculate_cost_from_response(response: dict) → float[来源]#

计算给定完成响应的成本。

Parameters:: response (dict) – 来自LiteLLM的完成响应。
Returns:: 完成调用的费用（以美元计）。
Return type:: 浮点数

property completion_cost#

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用特定于此类型模型的标记器计算提供的消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 采用LiteLLM API格式的聊天历史消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

property token_counter#

class camel.utils.MCPServer(function_names: list[str] | None = None, server_name: str | None = None)[来源]#

基类：object

make_wrapper(func: Callable[[...], Any]) → Callable[[...], Any][来源]#

class camel.utils.MistralTokenCounter(model_type: ~<unknown>.ModelType)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

使用针对此类模型加载的特定分词器，计算所提供消息列表中的标记数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的总令牌数。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

class camel.utils.OpenAITokenCounter(model: UnifiedModelType)[来源]#

基类：BaseTokenCounter

count_tokens_from_messages(messages: List[OpenAIMessage]) → int[来源]#

借助tiktoken包计算提供的消息列表中的token数量。

Parameters:: messages (List[OpenAIMessage]) – 以OpenAI API格式存储聊天历史记录的消息列表。
Returns:: 消息中的令牌数量。
Return type:: int

decode(token_ids: List[int]) → str[来源]#

将令牌ID解码回文本。

Parameters:: token_ids (List[int]) – 要解码的token ID列表。
Returns:: 解码后的文本。
Return type:: 字符串

encode(text: str) → List[int][来源]#

将文本编码为token ID。

Parameters:: 文本 (str) – 需要编码的文本。
Returns:: 令牌ID列表。
Return type:: 整数列表

camel.utils.agentops_decorator(func)[来源]#

如果ToolEvent可用，记录函数执行的装饰器。

Parameters:: func (可调用对象) – 需要被装饰的函数。
Returns:: 封装后的函数会记录其执行细节。
Return type:: 可调用

camel.utils.api_keys_required(param_env_list: List[Tuple[str | None, str]]) → Callable[[F], F][来源]#

一个装饰器，用于检查环境变量或函数参数中是否提供了所需的API密钥。

Parameters:

Returns:

原始函数经过包装并添加了检查: 用于验证必需的API密钥。

Return type:

Callable[[F], F]

Raises:

ValueError – 如果缺少任何必需的API密钥，无论是来自函数参数还是环境变量。

示例

@api_keys_required([
    ('api_key_arg', 'API_KEY_1'),
    ('another_key_arg', 'API_KEY_2'),
    (None, 'API_KEY_3'),
])
def some_api_function(api_key_arg=None, another_key_arg=None):
    # Function implementation that requires API keys

camel.utils.check_server_running(server_url: str) → bool[来源]#

检查URL指向的服务器的端口是否开放。

Parameters:: server_url (str) – 运行LLM推理服务的服务器URL。
Returns:: 端口是否开放以接收数据包（服务器正在运行）。
Return type:: 布尔值

camel.utils.create_chunks(text: str, n: int) → List[str][来源]#

从提供的文本中返回连续的n大小块。将文本分割成大小为n的较小块。

Parameters:

text (str) – 需要分割的文本。
n (int) – 单个块的最大长度。

Returns:

分割文本的列表。

Return type:

List[str]

camel.utils.deduplicate_internally(texts: List[str], threshold: float = 0.65, embedding_instance: BaseEmbedding[str] | None = None, embeddings: List[List[float]] | None = None, strategy: Literal['top1', 'llm-supervise'] = 'top1', batch_size: int = 1000) → 去重结果[来源]#

基于余弦相似度对字符串列表进行去重。

您可以选择： 1) 通过embedding_instance提供一个CAMEL BaseEmbedding实例来让

该函数内部处理嵌入，或者

直接将预计算好的嵌入列表传递给embeddings。

如果同时提供了embedding_instance和embeddings，该函数将引发ValueError以避免歧义使用。

strategy 用于指定不同的策略，其中 'top1' 选择相似度最高的那个，而 'llm-supervise' 使用 LLM 来判断文本是否为重复项（尚未实现）。

Parameters:

texts (List[str]) – 需要进行去重的文本列表。
threshold (float, optional) – 用于判定两段文本是否为重复文本的相似度阈值。(默认值: 0.65)
embedding_instance (Optional[BaseEmbedding[str]], optional) – 用于自动嵌入的CAMEL嵌入实例。(默认: None)
embeddings (Optional[List[List[float]]], optional) – 预计算的texts嵌入向量。列表中的每个元素对应于texts中相同索引位置的文本嵌入。(默认值: None)
strategy (Literal["top1", "llm-supervise"], optional) – 用于去重的策略。(默认: "top1")
batch_size (int, optional) – 用于计算余弦相似度的批次大小。(默认: 1000)

Returns:

一个包含以下内容的对象：

original_texts: 原始文本。
unique_ids: 去重后的唯一标识符。
unique_embeddings_dict: 一个字典，将（唯一）文本ID映射到其嵌入向量。
duplicate_to_target_map: 一个字典，将重复文本的ID映射到它被视为重复的目标文本ID。

Return type:

DeduplicationResult

Raises:

NotImplementedError – 如果策略不是"top1"。
ValueError - 如果既没有提供embeddings也没有提供embedding_instance，或者同时提供了两者。
ValueError – 如果embeddings的长度与texts的长度不匹配。

示例

>>> from camel.embeddings.openai_embedding import OpenAIEmbedding
>>> # Suppose we have 5 texts, some of which may be duplicates
>>> texts = [
...     "What is AI?",
...     "Artificial Intelligence is about machines",
...     "What is AI?",
...     "Deep Learning is a subset of AI",
...     "What is artificial intelligence?"
... ]
>>> # or any other BaseEmbedding instance
>>> embedding_model = OpenAIEmbedding()
>>> result = deduplicate_internally(
...     texts=texts,
...     threshold=0.7,
...     embedding_instance=embedding_model
... )
>>> print("Unique ids:")
>>> for uid in result.unique_ids:
...     print(texts[uid])
Unique ids:
What is AI?
Artificial Intelligence is about machines
Deep Learning is a subset of AI
What is artificial intelligence?

>>> print("Duplicate map:")
>>> print(result.duplicate_to_target_map)
{2: 0}
# This indicates the text at index 2 is considered
# a duplicate of index 0.

camel.utils.dependencies_required(*required_modules: str) → Callable[[F], F][来源]#

一个装饰器，用于确保在函数执行前指定的Python模块可用。

Parameters:

required_modules (str) – 需要检查可用性的必需模块。

Returns:

原始函数增加了对: 必需模块依赖项的检查。

Return type:

Callable[[F], F]

Raises:

ImportError – 如果任何必需的模块不可用。

示例

@dependencies_required('numpy', 'pandas')
def data_processing_function():
    # Function implementation...

camel.utils.download_github_subdirectory(repo: str, subdir: str, data_dir: Path, branch='main')[来源]#

下载基准测试Github仓库的子目录。

此函数从GitHub仓库的指定子目录下载所有文件和子目录，并将它们保存到本地目录。

Parameters:

repo (str) – GitHub仓库名称，格式为"owner/repo"。
subdir (str) – 仓库内要下载的子目录路径。
data_dir (Path) – 文件将被保存的本地目录。
branch (str, optional) – 要使用的仓库分支。默认为"main"。

camel.utils.download_tasks(task: 任务类型, folder_path: str) → None[来源]#

从指定URL下载任务相关文件并解压。

Parameters:

task (TaskType) - 一个枚举，表示要下载的任务类型。
folder_path (str) – zip文件将被下载并解压到的文件夹路径。

camel.utils.func_string_to_callable(code: str)[来源]#

将函数代码字符串转换为可调用的函数对象。

Parameters:

code (str) – 作为字符串形式的函数代码。

Returns:

从代码字符串中提取的可调用函数对象: 代码字符串。

Return type:

可调用[…, 任意类型]

camel.utils.get_first_int(string: str) → int | None[来源]#

返回给定字符串中找到的第一个整数。

如果没有找到整数，则返回 None。

Parameters:

字符串 (str) – 输入字符串。

Returns:

字符串中找到的第一个整数，如果没有找到则返回None: 未找到任何整数。

Return type:

整数或无

camel.utils.get_model_encoding(value_for_tiktoken: str)[来源]#

从tiktoken获取模型编码。

Parameters:: value_for_tiktoken – tiktoken的模型值。
Returns:: 模型编码。
Return type:: tiktoken.Encoding

camel.utils.get_prompt_template_key_words(template: str) → Set[str][来源]#

给定一个包含大括号{}的字符串模板，返回大括号内单词的集合。

Parameters:: template (str) – 一个包含大括号的字符串。
Returns:: 大括号内单词的列表。
Return type:: List[str]

示例

>>> get_prompt_template_key_words('Hi, {name}! How are you {status}?')
{'name', 'status'}

camel.utils.get_pydantic_major_version() → int[来源]#

获取Pydantic的主版本号。

Returns:: 如果安装了Pydantic，则显示其主版本号，否则显示0。
Return type:: int

camel.utils.get_pydantic_model(input_data: str | Type[BaseModel] | Callable) → Type[BaseModel][来源]#

A multi-purpose function that can be used as a normal function,: 一个类装饰器，或函数装饰器。

参数: input_data (Union[str, type, Callable]):

如果提供的是字符串，它应该是一个JSON编码的字符串
该字符串将被转换为BaseModel。

如果提供了函数，它将被装饰，使得
其参数被转换为BaseModel。

如果提供了BaseModel类，将直接返回它。

Returns:

BaseModel类将被用于: 结构化输入数据。

Return type:

类型[BaseModel]

camel.utils.get_pydantic_object_schema(pydantic_params: Type[BaseModel]) → Dict[来源]#

获取Pydantic模型的JSON模式。

Parameters:: pydantic_params (Type[BaseModel]) – 用于获取模式的Pydantic模型类。
Returns:: Pydantic模型的JSON模式。
Return type:: 字典

camel.utils.get_system_information()[来源]#

收集有关操作系统的信息。

Returns:: 一个包含各种操作系统信息的字典。
Return type:: 字典

camel.utils.get_task_list(task_response: str) → List[str][来源]#

解析Agent的响应并返回任务列表。

Parameters:: task_response (str) – Agent的字符串响应。
Returns:: 字符串任务列表。
Return type:: List[str]

camel.utils.handle_http_error(response: Response) → str[来源]#

根据响应的状态码处理HTTP错误。

Parameters:: response (requests.Response) – 来自API调用的HTTP响应。
Returns:: 错误类型，基于状态码。
Return type:: 字符串

camel.utils.is_docker_running() → bool[来源]#

检查Docker守护进程是否正在运行。

Returns:: 如果Docker守护进程正在运行则为True，否则为False。
Return type:: 布尔值

camel.utils.json_to_function_code(json_obj: Dict) → str[来源]#

根据JSON模式生成Python函数代码。

Parameters:: json_obj (dict) - 包含属性和必填字段的JSON模式对象，其json格式遵循openai工具模式
Returns:: 生成的Python函数代码作为字符串。
Return type:: 字符串

camel.utils.print_text_animated(text, delay: float = 0.02, end: str = '')[来源]#

以动画效果显示给定的文本。

Parameters:

text (str) – 要打印的文本。
delay (float, optional) – 每个字符打印之间的延迟时间。 (默认值: 0.02)
end (str, optional) – 在每个文本字符后打印的结束字符。(默认: "")

camel.utils.retry_on_error(max_retries: int = 3, initial_delay: float = 1.0) → Callable[来源]#

装饰器，用于在异常发生时以指数退避方式重试函数调用。

Parameters:

max_retries (int) - 最大重试次数
initial_delay (float) - 初始重试延迟时间（秒）

Returns:

带有重试逻辑的装饰函数

Return type:

可调用

camel.utils.text_extract_from_web(url: str) → str[来源]#

从给定URL获取文本信息。

Parameters:: url (str) – 您想要搜索的网站。
Returns:: 所有文本均从网页中提取。
Return type:: 字符串

camel.utils.to_pascal(snake: str) → str[来源]#

将snake_case字符串转换为PascalCase。

Parameters:: snake (str) – 需要转换的snake_case格式字符串。
Returns:: 转换后的PascalCase字符串。
Return type:: 字符串

camel.utils.track_agent(*args, **kwargs)[来源]#: AgentOps的模拟追踪代理装饰器。

camel.utils.with_timeout(timeout=None)[来源]#

为函数添加超时功能的装饰器。

以指定的超时值执行函数。如果执行时间超过限制，则返回超时消息。

Parameters:: timeout (float, optional) – 超时时长，单位为秒。如果为None，将尝试从实例的timeout属性获取超时时间。 (默认: None)

示例

>>> @with_timeout(5)
... def my_function():
...     return "Success"
>>> my_function()

>>> class MyClass:
...     timeout = 5
...     @with_timeout()
...     def my_method(self):
...         return "Success"