oai.completion

完成

class Completion(openai_Completion)

(openai<1) OpenAI 完成 API 的类。

它还支持：ChatCompletion，Azure OpenAI API。

设置缓存

@classmethod
def set_cache(cls,
              seed: Optional[int] = 41,
              cache_path_root: Optional[str] = ".cache")

设置缓存路径。

参数:

seed int, 可选 - 伪种子的整数标识符。对应于不同种子的结果将缓存到不同的位置。
cache_path str, Optional - 缓存的根路径。完整的缓存路径将会是 {cache_path_root}/{seed}。

清除缓存

@classmethod
def clear_cache(cls,
                seed: Optional[int] = None,
                cache_path_root: Optional[str] = ".cache")

清除缓存。

参数:

seed int, 可选 - 伪种子的整数标识符。如果省略，将清除cache_path_root下的所有缓存。
cache_path str, Optional - 缓存的根路径。完整的缓存路径将会是 {cache_path_root}/{seed}。

调优

@classmethod
def tune(cls,
         data: List[Dict],
         metric: str,
         mode: str,
         eval_func: Callable,
         log_file_name: Optional[str] = None,
         inference_budget: Optional[float] = None,
         optimization_budget: Optional[float] = None,
         num_samples: Optional[int] = 1,
         logging_level: Optional[int] = logging.WARNING,
         **config)

调整OpenAI API调用的参数。

TODO: 支持使用ray或spark进行并行调优。 TODO: 支持在测试中的agg_method。

参数:

data list - 数据点列表。
metric 字符串 - 需要优化的指标。
mode str - 优化模式，"min" 或 "max"。
eval_func Callable - 响应评估函数。该函数应接受一组响应和一个数据点作为输入，并返回一个指标字典。例如，

def eval_func(responses, **data):
    solution = data["solution"]
    success_list = []
    n = len(responses)
    for i in range(n):
        response = responses[i]
        succeed = is_equiv_chain_of_thought(response, solution)
        success_list.append(succeed)
    return {
        "expected_success": 1 - pow(1 - sum(success_list) / n, n),
        "success": any(s for s in success_list),
    }

log_file_name str, 可选 - 日志文件。
inference_budget float, 可选 - 推理预算，每实例的费用。
optimization_budget float, 可选 - 优化预算，总金额。
num_samples int, 可选 - 要评估的样本数量。-1 表示试验次数没有硬性限制，实际数量由 optimization_budget 决定。默认为 1。
logging_level 可选 - 日志级别。默认为 logging.WARNING。
**config dict - The search space to update over the default search. For prompt, please provide a string/Callable or a list of strings/Callables.
- 如果为聊天模型提供了提示，它将被转换为“用户”角色下的消息。
- 不要同时提供prompt和messages用于聊天模型，而是提供其中之一。
- 字符串模板将用于使用prompt.format(**data)为每个数据实例生成提示。
- 将使用可调用的模板来生成每个数据实例的提示，使用prompt(data)。对于停止条件，请提供一个字符串、字符串列表或字符串列表的列表。对于消息（仅限聊天模型），请提供消息列表（用于单个聊天前缀）或消息列表的列表（用于选择多个聊天前缀）。每条消息应该是一个包含 "role" 和 "content" 键的字典。"content" 的值可以是一个字符串/可调用模板。

dict - 优化后的超参数设置。
tune.ExperimentAnalysis - 调优结果。

创建

@classmethod
def create(cls,
           context: Optional[Dict] = None,
           use_cache: Optional[bool] = True,
           config_list: Optional[List[Dict]] = None,
           filter_func: Optional[Callable[[Dict, Dict], bool]] = None,
           raise_on_ratelimit_or_timeout: Optional[bool] = True,
           allow_format_str_template: Optional[bool] = False,
           **config)

为给定上下文生成补全。

参数:

context 字典, 可选 - 用于实例化提示的上下文。它需要包含提示模板或过滤函数使用的键。例如，prompt="Complete the following sentence: {prefix}, context={"prefix": "Today I feel"}。实际的提示将是： "完成以下句子：今天我感觉"。更多示例可以在templating中找到。
use_cache bool, Optional - 是否使用缓存的响应。
config_list 列表，可选的 - 用于尝试完成的配置列表。第一个不引发错误的配置将被使用。只需要提供与默认配置不同的部分。例如，

response = oai.Completion.create(
    config_list=[
        {
            "model": "gpt-4",
            "api_key": os.environ.get("AZURE_OPENAI_API_KEY"),
            "api_type": "azure",
            "base_url": os.environ.get("AZURE_OPENAI_API_BASE"),
            "api_version": "2024-02-01",
        },
        {
            "model": "gpt-3.5-turbo",
            "api_key": os.environ.get("OPENAI_API_KEY"),
            "api_type": "openai",
            "base_url": "https://api.openai.com/v1",
        },
        {
            "model": "llama-7B",
            "base_url": "http://127.0.0.1:8080",
            "api_type": "openai",
        }
    ],
    prompt="Hi",
)

filter_func 可调用的, 可选的 - 一个函数，接受上下文和响应并返回一个布尔值以指示响应是否有效。例如，

def yes_or_no_filter(context, config, response):
    return context.get("yes_or_no_choice", False) is False or any(
        text in ["Yes.", "No."] for text in oai.Completion.extract_text(response)
    )

raise_on_ratelimit_or_timeout bool, 可选 - 当所有配置失败时是否抛出RateLimitError或Timeout。当设置为False时，所有配置失败时将返回-1。
allow_format_str_template bool, 可选 - 是否允许在配置中使用格式字符串模板。
**config - Configuration for the openai API call. This is used as parameters for calling openai API. The "prompt" or "messages" parameter can contain a template (str or Callable) which will be instantiated with the context. Besides the parameters for the openai API call, it can also contain:
- max_retry_period (int): 允许重试失败请求的总时间（以秒为单位）。
- retry_wait_time (int): 在重试失败请求之前等待的时间间隔（以秒为单位）。
- cache_seed (int) 用于缓存。这在实现“受控随机性”完成时非常有用。

来自OpenAI API的响应，带有额外的字段。

cost: 总成本。当提供了config_list时，响应将包含更多字段：
config_id: 用于生成响应的config_list中配置的索引。
pass_filter: 响应是否通过过滤函数。如果没有提供过滤器，则为None。

测试

@classmethod
def test(cls,
         data,
         eval_func=None,
         use_cache=True,
         agg_method="avg",
         return_responses_and_per_instance_result=False,
         logging_level=logging.WARNING,
         **config)

评估使用OpenAI API调用的配置生成的响应。

参数:

data 列表 - 测试数据点的列表。
eval_func 可调用 - 每个数据实例的响应评估函数。该函数应接受一组响应和一个数据点作为输入，并返回一个包含指标的字典。您需要提供一个有效的可调用 eval_func；或者不提供（设置为 None），但在调用 tune 函数（其中提供了 eval_func）后调用测试函数。在后一种情况下，我们将使用通过 tune 函数提供的 eval_func。默认为 None。

def eval_func(responses, **data):
    solution = data["solution"]
    success_list = []
    n = len(responses)
    for i in range(n):
        response = responses[i]
        succeed = is_equiv_chain_of_thought(response, solution)
        success_list.append(succeed)
    return {
        "expected_success": 1 - pow(1 - sum(success_list) / n, n),
        "success": any(s for s in success_list),
    }

use_cache 布尔值, 可选 - 是否使用缓存的响应。默认为 True。
agg_method str, 可调用对象或可调用对象字典 - 每个指标的聚合方法（跨多个实例）。默认为 'avg'。一个字符串类型的 agg_method 示例：

agg_method = 'median'

一个可调用对象中的示例 agg_method：

agg_method = np.median

在可调用字典中的一个示例 agg_method：

agg_method={'median_success': np.median, 'avg_success': np.mean}

return_responses_and_per_instance_result bool - 是否在聚合结果之外还返回响应和每个实例的结果。
logging_level 可选 - 日志级别。默认为 logging.WARNING。
**config dict - 传递给openai API调用create()的参数。

当在测试或调整中没有提供有效的eval_func时，返回None；否则，如果return_responses_and_per_instance_result为True，则返回包含聚合结果、响应和每个实例结果的字典；否则，返回包含聚合结果的字典（不返回响应和每个实例结果）。

成本

@classmethod
def cost(cls, response: dict)

计算API调用的成本。

参数:

response dict - 来自 OpenAI API 的响应。

以美元计算的成本。如果模型不支持则为0。

提取文本

@classmethod
def extract_text(cls, response: dict) -> List[str]

从完成或聊天响应中提取文本。

参数:

response dict - 来自OpenAI API的响应。

响应的文本列表。

提取文本或函数调用

@classmethod
def extract_text_or_function_call(cls, response: dict) -> List[str]

从完成或聊天响应中提取文本或函数调用。

参数:

response dict - 来自 OpenAI API 的响应。

响应中的文本或函数调用列表。

已记录的历史记录

@classmethod
@property
def logged_history(cls) -> Dict

返回书籍管理字典。

print_usage_summary

@classmethod
def print_usage_summary(cls) -> Dict

返回使用情况摘要。

start_logging

@classmethod
def start_logging(cls,
                  history_dict: Optional[Dict] = None,
                  compact: Optional[bool] = True,
                  reset_counter: Optional[bool] = True)

开始记账。

参数:

history_dict Dict - 用于记录的字典。如果没有提供，将会创建一个新的字典。
compact bool - 是否保持历史记录字典的紧凑性。紧凑的历史记录每个对话只包含一个键，其值是一个类似字典的结构：

{
    "create_at": [0, 1],
    "cost": [0.1, 0.2],
}

其中 "created_at" 是 API 调用的索引，表示所有调用的顺序，"cost" 是每次调用的成本。此示例显示会话基于两个 API 调用。紧凑格式对于压缩会话历史记录非常有用。如果 compact 为 False，历史记录字典将包含所有 API 调用：键是 API 调用的索引，值是一个类似以下的字典：

{
    "request": request_dict,
    "response": response_dict,
}

其中 request_dict 是发送到 OpenAI API 的请求，而 response_dict 是响应。对于包含两个 API 调用的对话，非压缩的历史字典将如下所示：

{
    0: {
        "request": request_dict_0,
        "response": response_dict_0,
    },
    1: {
        "request": request_dict_1,
        "response": response_dict_1,
    },

第一个请求的消息加上响应等于第二个请求的消息。对于多次对话，非紧凑的历史字典具有二次方大小，而紧凑的历史字典具有线性大小。

reset_counter bool - 是否重置API调用次数的计数器。

stop_logging

@classmethod
def stop_logging(cls)

结束账目记录。

ChatCompletion

class ChatCompletion(Completion)

(openai<1) 一个用于OpenAI API ChatCompletion的类。与Completion共享相同的API。

完成​

设置缓存​

清除缓存​

调优​

创建​

测试​

成本​

提取文本​

提取文本或函数调用​

已记录的历史记录​

print_usage_summary​

start_logging​

stop_logging​

ChatCompletion​

完成