GPT LLM

GPT

Bases: ChatLLM[dict]

OpenAI GPT系列对话模型 | OpenAI GPT series chat model

partial_params property

partial_params: dict

Get partial parameters for OpenAI APIs | 获取部分通用参数，用于请求OpenAI接口 Returns: dict

model_post_init

model_post_init(__context: Any) -> None

重写 init 容易对mypy参数判断产生影响，因此使用 model_post_init替换 init 方法。这个方法的执行时机描述如下：

https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_post_init

Parameters:

Name	Type	Description	Default
__context	Any		required

Source code in tfrobot/brain/chain/llms/openai.py

def model_post_init(self, __context: Any) -> None:
    """
    重写 __init__ 容易对mypy参数判断产生影响，因此使用 model_post_init替换 __init__ 方法。这个方法的执行时机描述如下：

    https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_post_init

    Args:
        __context:
    """
    httpx_client = httpx_client_factory(
        self.proxy_host, self.proxy_port, self.proxy_user, self.proxy_pass, timeout=self.timeout
    )
    async_httpx_client = httpx_async_client_factory(
        self.proxy_host, self.proxy_port, self.proxy_user, self.proxy_pass, timeout=self.timeout
    )
    self._client = OpenAI(http_client=httpx_client, api_key=self.openai_api_key)
    self._async_client = AsyncOpenAI(http_client=async_httpx_client, api_key=self.openai_api_key)

reformat_request_msg_to_api

reformat_request_msg_to_api(msg: BaseLLMMessage) -> dict

将LLMUserMessage转换为GPT请求消息格式。GPT接口对于格式有自己的要求，与当前TFRobot协议并不完全一致。因此需要进行转换。

Parameters:

Name	Type	Description	Default
msg	BaseLLMMessage	BaseLLMMessage	required

Returns:

Name	Type	Description
dict	dict	GPT请求消息格式

Source code in tfrobot/brain/chain/llms/openai.py

def reformat_request_msg_to_api(self, msg: BaseLLMMessage) -> dict:
    """
    将LLMUserMessage转换为GPT请求消息格式。GPT接口对于格式有自己的要求，与当前TFRobot协议并不完全一致。因此需要进行转换。

    Args:
        msg (BaseLLMMessage): BaseLLMMessage

    Returns:
        dict: GPT请求消息格式
    """
    msg_dict = msg.model_dump(exclude_none=True)
    llm_meta = LLMeta(self.name)
    if isinstance(msg, LLMTFLMessage):
        # 如果是TFL模式的消息，目前将其转换为User类型消息。原因是Tool类型消息需要紧跟一个真实的工具调用，为此如果Mock工具信息得不偿失
        msg_dict["role"] = "user"
    if isinstance(msg, LLMUserMessage) and isinstance(msg.content, list):
        if isinstance(msg_dict.get("content"), list):
            for part in msg_dict["content"]:
                mime_type = part.pop("mime_type", "image/jpeg")
                if part.get("type") == "image_url":
                    if not llm_meta.capabilities.supports_vision:
                        part["text"] = (
                            part["image_url"].get("detail")
                            or "当前模型不支持图片输入，如果此图片内容异常重要，你可以提示用例切换模型。如果当前图片信息可有可元，不影响判断，可以忽略此消息。"
                        )
                        part["type"] = "text"  # 如果模型不支持图片，则需要转换类型，避免接口异常
                    elif not part.get("image_url", {}).get("url").startswith("http"):
                        # 本地图片
                        part["image_url"]["detail"] = (
                            part["image_url"].get("detail")
                            if part["image_url"].get("detail") in ["low", "high", "auto"]
                            else "auto"
                        )
                        if not is_base64(part.get("image_url", {}).get("url")):
                            # 本地图片（非网络）需要转换为base64 | Local images (non-network) need to be converted to base64
                            result = self.load_image_from_uri(part["image_url"]["url"])
                            if result:
                                img_data, mime_type = result
                                part["image_url"]["url"] = f"data:{mime_type};base64,{img_data.decode()}"
                            else:
                                # 如果加载失败，记录警告但继续处理 | If loading fails, log warning but continue
                                warnings.warn(f"Failed to load image from URI: {part['image_url']['url']}")
                        else:
                            # 对于已经转换为Base64的图片，需要格式化为：f"data:image/jpeg;base64,{base64_image}"
                            if not part["image_url"]["url"].startswith("data:"):
                                part["image_url"]["url"] = f"data:{mime_type};base64,{part['image_url']['url']}"
                    else:
                        # 远程图片 修改detail信息，其它继续延用
                        part["image_url"]["detail"] = (
                            part["image_url"].get("detail")
                            if part["image_url"].get("detail") in ["low", "high", "auto"]
                            else "auto"
                        )

                elif part.get("type") == "audio_url":
                    # 处理音频输入 | Handle audio input
                    # OpenAI要求音频使用input_audio格式，包含data和format字段
                    # OpenAI requires audio to use input_audio format with data and format fields
                    audio_url_data = part.pop("audio_url", {})
                    audio_url = audio_url_data.get("url", "")

                    if not llm_meta.capabilities.supports_audio:
                        part["type"] = "text"  # 如果不支持audio，则需要将其type类型转换为text，否则容易引起接口异常
                        part["text"] = (
                            audio_url_data.get("detail")
                            or "当前模型不支持音频输入，如果此音频内容异常重要，你可以提示用例切换模型。如果当前音频信息可有可元，不影响判断，可以忽略此消息。"
                        )

                    elif not is_base64(audio_url):
                        # 本地音频（非网络）需要转换为base64 | Local audio (non-network) needs to be converted to base64
                        result = self.load_audio_from_uri(audio_url)
                        if result:
                            audio_data, audio_mime_type = result
                            # 根据MIME类型确定格式 | Determine format based on MIME type
                            audio_format = "mp3" if "mp3" in audio_mime_type or "mpeg" in audio_mime_type else "wav"
                            # 转换为OpenAI的input_audio格式 | Convert to OpenAI's input_audio format
                            part["type"] = "input_audio"
                            part["input_audio"] = {
                                "data": audio_data.decode(),  # base64编码的音频数据 | base64 encoded audio data
                                "format": audio_format,
                            }
                        else:
                            # 如果加载失败，记录警告但继续处理 | If loading fails, log warning but continue
                            warnings.warn(f"Failed to load audio from URI: {audio_url}")
                    else:
                        # 对于已经是base64的音频数据 | For audio data that is already base64
                        audio_format = "mp3" if "mp3" in mime_type or "mpeg" in mime_type else "wav"
                        part["type"] = "input_audio"
                        part["input_audio"] = {"data": audio_url, "format": audio_format}
                elif part.get("type") == "video_url":
                    # OpenAI目前不支持视频，将视频的detail信息转换为文本
                    # OpenAI currently does not support video, convert video detail to text
                    assert (
                        not llm_meta.capabilities.supports_video
                    ), "目前OpenAI所有Chat类模型均不支持视频，如果看到此警告，请联系开发者解决。"
                    video_url_data = part.get("video_url", {})
                    video_detail = video_url_data.get("detail", "")
                    video_url = video_url_data.get("url", "")

                    # 构建提示文本 | Build prompt text
                    # 如果是base64数据，不要包含在文本中（太大）| If it's base64 data, don't include in text (too large)
                    if is_base64(video_url):
                        video_text = "[视频内容（base64编码，已省略）"
                    else:
                        video_text = f"[视频内容: {video_url}"

                    if video_detail:
                        video_text += f", 详情 | Detail: {video_detail}"
                    video_text += "]"

                    # 转换为文本部分 | Convert to text part
                    part["type"] = "text"
                    part["text"] = video_text
                    # 移除video_url字段 | Remove video_url field
                    part.pop("video_url", None)

                    warnings.warn(
                        "目前OpenAI Chat接口尚不支持Video，已将视频信息转换为文本提示。"
                        "请尝试更换其它模型比如Gemini | "
                        "OpenAI Chat API does not support video yet, converted video info to text prompt. "
                        "Try using other models like Gemini"
                    )
                elif part.get("type") == "pdf_url":
                    # 处理PDF文件输入 | Handle PDF file input
                    # OpenAI要求PDF使用file格式，包含file对象
                    # OpenAI requires PDF to use file format with file object
                    pdf_url_data = part.pop("pdf_url", {})
                    pdf_url = pdf_url_data.get("url", "")

                    if not llm_meta.capabilities.supports_vision:
                        # 如果模型不支持PDF，转换为文本提示 | If model doesn't support PDF, convert to text prompt
                        part["type"] = "text"
                        part["text"] = (
                            pdf_url_data.get("detail")
                            or "当前模型不支持PDF文件输入，如果此PDF内容异常重要，你可以提示用户切换模型或提取PDF文本内容。如果当前PDF信息可有可无，不影响判断，可以忽略此消息。"
                        )
                    elif not is_base64(pdf_url):
                        # 本地PDF文件（非网络）需要转换为base64 | Local PDF file (non-network) needs to be converted to base64
                        result = self.load_file_from_uri(pdf_url)
                        if result:
                            file_data, file_mime_type = result
                            # 转换为OpenAI的file格式，使用data URL格式 | Convert to OpenAI's file format using data URL format
                            part["type"] = "file"
                            part["file"] = {"file_data": f"data:{file_mime_type};base64,{file_data.decode()}"}
                            # filename字段使用 detail 或者 mime_type 填充
                            part["file"]["filename"] = pdf_url_data.get("detail") or file_mime_type
                        else:
                            # 如果加载失败，记录警告但继续处理 | If loading fails, log warning but continue
                            warnings.warn(f"Failed to load PDF from URI: {pdf_url}")
                    else:
                        # 对于已经是base64的PDF数据，需要格式化为data URL | For PDF data that is already base64, format as data URL
                        part["type"] = "file"
                        if not pdf_url.startswith("data:"):
                            part["file"] = {"file_data": f"data:application/pdf;base64,{pdf_url}"}
                        else:
                            part["file"] = {"file_data": pdf_url}
                        # filename字段使用 detail 或者 mime_type 填充
                        part["file"]["filename"] = pdf_url_data.get("detail") or "application/pdf"
    # OpenAI接口如果有设置用户名，那用户名需要满足OpenAI的正则要求，不允许出现一些特殊字符，因此需要进行转换
    if msg_dict.get("name"):
        msg_dict["name"] = replace_forbidden_chars(msg_dict["name"])
    return msg_dict

construct_request_params

construct_request_params(current_input: BaseMessage, conversation: Optional[list[BaseMessage]] = None, elements: Optional[list[DocElement]] = None, knowledge: Optional[str] = None, tools: Optional[list[BaseTool]] = None, intermediate_msgs: Optional[list[BaseMessage]] = None, response_format: Optional[LLMResponseFormat] = None) -> dict

Construct chat request parameters | 构造聊天请求参数

最终的会话排序如下： 1. 首条System Message 2. 用户历史会话内容 3. 当前输入前的提示 4. 当前输入 5. 当前输入后的提示 6. 中间消息

Parameters:

Name	Type	Description	Default
current_input	BaseMessage	Current input | 当前输入	required
conversation	Optional[list[BaseMessage]]	Conversation history | 会话历史	None
elements	Optional[list[DocElement]]	Document elements | 文档元素	None
knowledge	Optional[str]	Related knowledge | 相关知识	None
tools	Optional[list[BaseTool]]	Available tool list | 可用的工具列表	None
intermediate_msgs	Optional[list[BaseMessage]]	Intermediate messages, it will be generated during llm work in chain, list tool's response or other system message, those messages will not save into memory, but usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中， 但在链式工作中非常有用	None
response_format	Optional[LLMResponseFormat]	Response format | 响应格式 如果指定了此参数，将会覆盖模型的默认响应格式	None

Returns:

Name	Type	Description
dict	dict	Request parameters | 请求参数

Source code in tfrobot/brain/chain/llms/openai.py

@overrides.override
def construct_request_params(
    self,
    current_input: BaseMessage,
    conversation: Optional[list[BaseMessage]] = None,
    elements: Optional[list[DocElement]] = None,
    knowledge: Optional[str] = None,
    tools: Optional[list[BaseTool]] = None,
    intermediate_msgs: Optional[list[BaseMessage]] = None,
    response_format: Optional[LLMResponseFormat] = None,
) -> dict:
    """
    Construct chat request parameters | 构造聊天请求参数

    最终的会话排序如下：
    1. 首条System Message
    2. 用户历史会话内容
    3. 当前输入前的提示
    4. 当前输入
    5. 当前输入后的提示
    6. 中间消息

    Args:
        current_input(BaseMessage): Current input | 当前输入
        conversation(Optional[list[BaseMessage]]): Conversation history | 会话历史
        elements(Optional[list[DocElement]]): Document elements | 文档元素
        knowledge(Optional[str]): Related knowledge | 相关知识
        tools(Optional[list[BaseTool]]): Available tool list | 可用的工具列表
        intermediate_msgs(Optional[list[BaseMessage]]): Intermediate messages, it will be generated during llm work
            in chain, list tool's response or other system message, those messages will not save into memory, but
            usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中，
            但在链式工作中非常有用
        response_format(Optional[LLMResponseFormat]): Response format | 响应格式
            如果指定了此参数，将会覆盖模型的默认响应格式

    Returns:
        dict: Request parameters | 请求参数
    """
    llm_model = LLMeta(self.name)
    # 构造请求参数
    partial_params = self.partial_params
    # 依据模板配置进行格式化
    prompt_ctx = self.construct_prompt_context(
        current_input, conversation, elements, knowledge, tools, intermediate_msgs, response_format
    )
    response_format = response_format or self.response_format
    if (
        llm_model.capabilities.supports_json_outputs
        and response_format
        and response_format != inspect.Parameter.empty
    ):
        # 优先使用参数覆盖，其次使用类配置
        partial_params["response_format"] = self._extract_res_format_dict(response_format)
    # 格式化首条System Message
    req_msgs = self.format_to_request_msgs(current_input, prompt_ctx)
    messages = req_msgs.to_list()
    # 构造请求参数
    params = {"messages": [self.reformat_request_msg_to_api(msg) for msg in messages]}
    params.update(**partial_params)
    # 下面这个代码在2025-3-19附近频繁变更过。一直在犹豫是否去掉 「and not self.used_tool_prompt」这个判断条件，但具体原因已经不可考。
    # 最终在2025-4-15决定再次添加这个判断条件。后续需要观察这个条件的增加与删除带来的影响。
    if prompt_ctx.tools and not self.used_tool_prompt:
        # 针对OpenAI，支持传入tools参数，可以通过此参数进行更好的工具调用，但如果使用了ToolPrompt，此参数会被ToolPrompt覆盖
        params["tools"] = [
            {
                "type": "function",
                "function": {
                    "description": cast(PromptTool, tool).description,
                    "name": cast(PromptTool, tool).name,
                    "parameters": cast(PromptTool, tool).json_params_schema
                    if cast(PromptTool, tool).json_params_schema
                    != {"type": "null"}  # OpenAI API要求如果无参数则不传此参数，但不支持 null 写法
                    else None,
                    "strict": self.tool_params_strict
                    if self.tool_params_strict is not inspect.Parameter.empty
                    else False,  # OpenAI 目前支持工具强制约束。但仅支持subset of JsonSchema。因此需要用户自己主动判断是否打开这个参数
                },
            }
            for tool in prompt_ctx.tools
        ]
    # 因为现在OpenAI API对入参结构有诸多限制，此函数对这些非通用限制进行适配，希望早日去掉
    params["messages"] = self.lying_to_openai_engineer(params["messages"])
    return params

construct_llm_result

construct_llm_result(chat_completion: ChatCompletion, params: dict) -> LLMResult

Construct LLM result | 构造LLM结果

Parameters:

Name	Type	Description	Default
chat_completion	ChatCompletion	Chat completion from OpenAI SDK | 聊天完成，来自OpenAI SDK的定义	required
params	dict	Request parameters | 请求参数	required

Returns:

Name	Type	Description
LLMResult	LLMResult	LLM result | LLM结果

Source code in tfrobot/brain/chain/llms/openai.py

def construct_llm_result(self, chat_completion: ChatCompletion, params: dict) -> LLMResult:
    """
    Construct LLM result | 构造LLM结果

    Args:
        chat_completion(ChatCompletion): Chat completion from OpenAI SDK | 聊天完成，来自OpenAI SDK的定义
        params(dict): Request parameters | 请求参数

    Returns:
        LLMResult: LLM result | LLM结果
    """
    if not chat_completion.usage:
        warnings.warn("Chat completion usage is empty, can't calculate cost")

        try:
            chat_completion.usage = calculate_tokens(
                chat_completion=chat_completion, prompt=params, model=self.name
            )
        except Exception as e:
            warnings.warn(f"Failed to calculate tokens usage: {e}")
            chat_completion.usage = CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)
    # 构建Generation
    generations: list[Generation] = []
    for choice in chat_completion.choices:
        g = Generation(text=choice.message.content, finish_reason=choice.finish_reason)
        if choice.message.function_call:
            g.function_call = FunctionCall(
                name=choice.message.function_call.name, parameters=choice.message.function_call.arguments
            )
        if choice.message.tool_calls:
            g.tool_calls = [
                ToolCall(
                    call_id=tool.id,
                    tool_type=tool.type,
                    function=FunctionCall(name=tool.function.name, parameters=tool.function.arguments),
                )
                for tool in choice.message.tool_calls
                if isinstance(tool, ChatCompletionMessageFunctionToolCall)
            ]
        generations.append(g)
    # 构建LLMResult 元数据信息
    meta: LLMResultMeta = {
        "id": chat_completion.id,
        "created": chat_completion.created,
        "model": self.name,  # 注意这里不能使用 chat_completion.model 因为比如 gpt-4o-mini 请求返回，会是其实际指向的model-name: gpt-4o-mini-2024-07-18 这会导致内部分析混乱
        "system_fingerprint": chat_completion.system_fingerprint,
        "eval_duration": None,
        "eval_count": None,
        "prompt_eval_duration": None,
        "load_duration": None,
        "total_duration": None,
        "done": None,
    }
    # 构建Usage
    usage: TokenUsage = {
        "prompt_tokens": chat_completion.usage.prompt_tokens,
        "completion_tokens": chat_completion.usage.completion_tokens,
        "total_tokens": chat_completion.usage.total_tokens,
        "completion_cost": chat_completion.usage.completion_tokens * self.output_price / 1000.0,
        "prompt_cost": chat_completion.usage.prompt_tokens * self.input_price / 1000.0,
        "total_cost": (
            chat_completion.usage.completion_tokens * self.output_price
            + chat_completion.usage.prompt_tokens * self.input_price
        )
        / 1000.0,
    }
    llm_res = LLMResult(generations=generations, usage=usage, meta_info=meta, prompt=copy.deepcopy(params))
    if self.used_tool_prompt:
        for prompt in self.all_prompts:
            if isinstance(prompt, ToolPrompt):
                prompt.extract_tool_call(llm_res)
    if self.enable_tfl_mode:
        for prompt in self.all_prompts:
            if isinstance(prompt, TFLPrompt):
                prompt.extract_tool_call(llm_res)
    return llm_res

merge_chat_completion_chunk staticmethod

merge_chat_completion_chunk(chunk_finally: ChatCompletionChunk, chunk: ChatCompletionChunk) -> None

将每一个chunk合并进入chunk_finally中 | Merge each chunk into chunk_finally

Parameters:

Name	Type	Description	Default
chunk_finally	ChatCompletionChunk	The final chunk | 最终的chunk	required
chunk	ChatCompletionChunk	The current chunk | 当前的chunk	required

Returns:

Type	Description
None	None

Source code in tfrobot/brain/chain/llms/openai.py

@staticmethod
def merge_chat_completion_chunk(chunk_finally: ChatCompletionChunk, chunk: ChatCompletionChunk) -> None:
    """
    将每一个chunk合并进入chunk_finally中 | Merge each chunk into chunk_finally

    Args:
        chunk_finally(ChatCompletionChunk): The final chunk | 最终的chunk
        chunk(ChatCompletionChunk): The current chunk | 当前的chunk

    Returns:
        None
    """
    # 当前函数中有些代码被标记为不需要测试覆盖，因为这些部分仅仅是有理论上存在的可能性，在目前的实际使用中没有遇到过。
    # 但是为了保证代码的完整性，这些部分仍然保留在代码中。
    # Some code in the current function is marked as not needing test coverage, because these parts are only
    # theoretically possible, and have not been encountered in the current actual use. But in order to ensure the
    # integrity of the code, these parts are still retained in the code.
    if chunk.usage:
        if not chunk_finally.usage:
            chunk_finally.usage = chunk.usage.model_copy()
        else:
            chunk_finally.usage.prompt_tokens += chunk.usage.prompt_tokens
            chunk_finally.usage.completion_tokens += chunk.usage.completion_tokens
            chunk_finally.usage.total_tokens += chunk.usage.total_tokens
    for choice in chunk.choices:
        original_choice: Optional[ChunkChoice] = next(
            filter(
                lambda x: x.index == choice.index,  # type: ignore
                chunk_finally.choices,
            ),
            None,
        )
        if not original_choice:
            # 如果没能找到index相同的内容，则进行追加
            chunk_finally.choices.append(choice)  # pragma: no cover
        else:
            # 如果有找到index相同的内容，则进行更新
            original_choice.finish_reason = choice.finish_reason
            original_choice.logprobs = choice.logprobs  # pragma: no cover
            if choice.delta.role:
                original_choice.delta.role = choice.delta.role  # pragma: no cover
            if choice.delta.content:
                if original_choice.delta.content is None:  # pragma: no cover
                    original_choice.delta.content = ""  # pragma: no cover
                original_choice.delta.content += choice.delta.content
            if choice.delta.function_call:
                original_choice.delta.function_call = choice.delta.function_call  # pragma: no cover
            if choice.delta.tool_calls:
                for tool_call in choice.delta.tool_calls:
                    if original_choice.delta.tool_calls is None:
                        original_choice.delta.tool_calls = [tool_call]  # pragma: no cover
                    if tool_call.index not in [x.index for x in original_choice.delta.tool_calls]:
                        original_choice.delta.tool_calls.append(tool_call)  # pragma: no cover
                    original_tool_call = next(
                        filter(lambda x: x.index == tool_call.index, original_choice.delta.tool_calls), None
                    )
                    if tool_call.function and tool_call.function.arguments is not None and original_tool_call:
                        if original_tool_call.function is None:
                            original_tool_call.function = ChoiceDeltaToolCallFunction()
                        original_tool_call.function.arguments = (
                            original_tool_call.function.arguments + tool_call.function.arguments
                            if original_tool_call.function.arguments is not None
                            else tool_call.function.arguments
                        )
                    if tool_call.function and tool_call.function.name is not None and original_tool_call:
                        assert (
                            original_tool_call.function is not None
                        ), "OriginalToolCall has no function struct"  # pragma: no cover
                        original_tool_call.function.name = tool_call.function.name  # pragma: no cover

complete

complete(current_input: BaseMessage, conversation: Optional[list[BaseMessage]] = None, elements: Optional[list[DocElement]] = None, knowledge: Optional[str] = None, tools: Optional[list[BaseTool]] = None, intermediate_msgs: Optional[list[BaseMessage]] = None, response_format: Optional[LLMResponseFormat] = None) -> LLMResult

Complete chat | 生成聊天结果

最终的会话排序如下： 1. 首条System Message 2. 用户历史会话内容 3. 当前输入前的提示 4. 当前输入 5. 当前输入后的提示 6. 中间消息

Parameters:

Name	Type	Description	Default
current_input	BaseMessage	User last input | 用户最后输入	required
conversation	Optional[list[BaseMessage]]	Conversation history | 会话历史	None
elements	Optional[list[DocElement]]	Document elements | 文档元素	None
knowledge	Optional[str]	Related knowledge | 相关知识	None
tools	Optional[list[BaseTool]]	Available tool list | 可用的工具列表	None
intermediate_msgs	Optional[list[BaseMessage]]	Intermediate messages, it will be generated during llm work in chain, list tool's response or other system message, those messages will not save into memory, but usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中， 但在链式工作中非常有用	None
response_format	Optional[LLMResponseFormat]	Response format | 响应格式 如果在此指定响应格式，会覆盖LLM的默认响应格式。	None

Returns:

Name	Type	Description
LLMResult	LLMResult	The result of completion | 完成的结果

Source code in tfrobot/brain/chain/llms/openai.py

@report_llm_metrics
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((APITimeoutError, APIConnectionError)),
    before_sleep=before_sleep_log(logger, logging.WARN),
)
@overrides.override
def complete(
    self,
    current_input: BaseMessage,
    conversation: Optional[list[BaseMessage]] = None,
    elements: Optional[list[DocElement]] = None,
    knowledge: Optional[str] = None,
    tools: Optional[list[BaseTool]] = None,
    intermediate_msgs: Optional[list[BaseMessage]] = None,
    response_format: Optional[LLMResponseFormat] = None,
) -> LLMResult:
    """
    Complete chat | 生成聊天结果

    最终的会话排序如下：
    1. 首条System Message
    2. 用户历史会话内容
    3. 当前输入前的提示
    4. 当前输入
    5. 当前输入后的提示
    6. 中间消息

    Args:
        current_input(BaseMessage): User last input | 用户最后输入
        conversation(Optional[list[BaseMessage]]): Conversation history | 会话历史
        elements(Optional[list[DocElement]]): Document elements | 文档元素
        knowledge(Optional[str]): Related knowledge | 相关知识
        tools(Optional[list[BaseTool]]): Available tool list | 可用的工具列表
        intermediate_msgs(Optional[list[BaseMessage]]): Intermediate messages, it will be generated during llm work
            in chain, list tool's response or other system message, those messages will not save into memory, but
            usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中，
            但在链式工作中非常有用
        response_format(Optional[LLMResponseFormat]): Response format | 响应格式
            如果在此指定响应格式，会覆盖LLM的默认响应格式。

    Returns:
        LLMResult: The result of completion | 完成的结果
    """
    params = self.construct_request_params(
        current_input, conversation, elements, knowledge, tools, intermediate_msgs, response_format
    )
    with tracer.start_as_current_span("openai-chat", kind=SpanKind.CLIENT) as span:
        if self.stream:
            span_ctx = LLMEventContext(
                scene="LLM",
                entity_id=str(id(self)),
                desc="before gpt streaming request",
                info=str(current_input.content),
                streaming=self.stream,
                token_usage=None,
                user_input=str(current_input.content),
                llm_model_name=self.name,
            )
            span.add_event(SpanEvent.BEFORE_LLM_STREAMING.value, span_ctx.model_dump())
            try:
                chat_completion_stream: Stream[ChatCompletionChunk] = cast(
                    Stream[ChatCompletionChunk], self._client.chat.completions.create(**params)
                )
            except BadRequestError as e:
                body = cast(dict, e.body)
                if body.get("code") == "context_length_exceeded":
                    # 从错误中提取 token 信息 | Extract token info from error
                    current_size, target_size = self.extract_context_size_from_error(e)
                    # 将 OpenAI 的上下文超长异常转换为 ContextTooLargeError，由 Chain 层捕获并处理
                    raise ContextTooLargeError(
                        message=f"OpenAI context length exceeded: {body.get('message')}",
                        current_size=current_size,
                        target_size=target_size,
                        model_name=self.name,
                    )
                else:
                    raise e
            except Exception as e:
                warnings.warn(f"Unexpected error occurred during request to OpenAI API: {e}")
                raise e
            assert isinstance(chat_completion_stream, Stream), "chat_completion is not Stream"
            chat_completion_finally = None
            for res_chunk in chat_completion_stream:
                if not chat_completion_finally:
                    chat_completion_finally = res_chunk.model_copy(deep=True)  # 在拿到首个返回后，将其深拷贝
                else:
                    # 在拿到首个返回之后，将后续的chunk返回合并到首个返回中
                    self.merge_chat_completion_chunk(chat_completion_finally, res_chunk)
                span_ctx = LLMEventContext(
                    scene="LLM",
                    entity_id=str(id(self)),
                    desc="gpt streaming response",
                    info=json.dumps(res_chunk.model_dump(), indent=2, ensure_ascii=False),
                    streaming=self.stream,
                    token_usage=None,
                    user_input=str(current_input.content),
                    llm_model_name=self.name,
                )
                span.add_event(SpanEvent.LLM_STREAMING.value, span_ctx.model_dump())
            span_ctx = LLMEventContext(
                scene="LLM",
                entity_id=str(id(self)),
                desc="after gpt streaming response",
                info=json.dumps(chat_completion_finally.model_dump(), indent=2, ensure_ascii=False)
                if chat_completion_finally
                else "None",
                streaming=self.stream,
                token_usage=None,
                user_input=str(current_input.content),
                llm_model_name=self.name,
            )
            span.add_event(SpanEvent.AFTER_LLM_STREAMING.value, span_ctx.model_dump())
            if chat_completion_finally:
                # 在OpenAI的接口中，需要将delta中的内容追加到message中，同时需要修改object字段为chat.completion
                dict_res = chat_completion_finally.model_dump()
                dict_res["object"] = "chat.completion"
                for choice in dict_res["choices"]:
                    delta = choice.pop("delta")
                    choice["message"] = delta
                chat_completion = ChatCompletion(**dict_res)
            else:
                raise ValueError("chat_completion_finally is None")  # pragma: no cover
        else:
            try:
                chat_completion = self._client.chat.completions.create(**params)
            except BadRequestError as e:
                body = cast(dict, e.body)
                if body.get("code") == "context_length_exceeded":
                    # 从错误中提取 token 信息 | Extract token info from error
                    current_size, target_size = self.extract_context_size_from_error(e)
                    # 将 OpenAI 的上下文超长异常转换为 ContextTooLargeError，由 Chain 层捕获并处理
                    raise ContextTooLargeError(
                        message=f"OpenAI context length exceeded: {body.get('message')}",
                        current_size=current_size,
                        target_size=target_size,
                        model_name=self.name,
                    )
                else:
                    raise e
            except Exception as e:
                warnings.warn(f"Unexpected error occurred during request to OpenAI API: {e}")
                raise e
    return self.construct_llm_result(chat_completion, params=params)

async_complete async

async_complete(current_input: BaseMessage, conversation: Optional[list[BaseMessage]] = None, elements: Optional[list[DocElement]] = None, knowledge: Optional[str] = None, tools: Optional[list[BaseTool]] = None, intermediate_msgs: Optional[list[BaseMessage]] = None, response_format: Optional[LLMResponseFormat] = None) -> LLMResult

Complete chat | 生成聊天结果

最终的会话排序如下： 1. 首条System Message 2. 用户历史会话内容 3. 当前输入前的提示 4. 当前输入 5. 当前输入后的提示 6. 中间消息

OpenAI遇到超过上下文限制的异常时会抛出错误： openai.BadRequestError Error code: 400 - {'error': {'message': "This model's maximum context length is 16385 tokens. However, your messages resulted in 768010 tokens. Please reduce the length of the messages.", 'type': 'invalid_request_error', 'param': 'messages', 'code': 'context_length_exceeded'}}

Parameters:

Name	Type	Description	Default
current_input	BaseMessage	User last input | 用户最后输入	required
conversation	Optional[list[BaseMessage]]	Conversation history | 会话历史	None
elements	Optional[list[DocElement]]	Document elements | 文档元素	None
knowledge	Optional[str]	Related knowledge | 相关知识	None
tools	Optional[list[BaseTool]]	Available tool list | 可用的工具列表	None
intermediate_msgs	Optional[list[BaseMessage]]	Intermediate messages, it will be generated during llm work in chain, list tool's response or other system message, those messages will not save into memory, but usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中， 但在链式工作中非常有用	None
response_format	Optional[LLMResponseFormat]	Response format | 响应格式 如果在此指定响应格式，会覆盖LLM的默认响应格式。	None

Returns:

Name	Type	Description
LLMResult	LLMResult	The result of completion | 完成的结果

Source code in tfrobot/brain/chain/llms/openai.py

@report_llm_metrics
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((APITimeoutError, APIConnectionError)),
    before_sleep=before_sleep_log(logger, logging.WARN),
)
@overrides.override
async def async_complete(
    self,
    current_input: BaseMessage,
    conversation: Optional[list[BaseMessage]] = None,
    elements: Optional[list[DocElement]] = None,
    knowledge: Optional[str] = None,
    tools: Optional[list[BaseTool]] = None,
    intermediate_msgs: Optional[list[BaseMessage]] = None,
    response_format: Optional[LLMResponseFormat] = None,
) -> LLMResult:
    """
    Complete chat | 生成聊天结果

    最终的会话排序如下：
    1. 首条System Message
    2. 用户历史会话内容
    3. 当前输入前的提示
    4. 当前输入
    5. 当前输入后的提示
    6. 中间消息

    OpenAI遇到超过上下文限制的异常时会抛出错误：
    openai.BadRequestError
    Error code: 400 - {'error': {'message': "This model's maximum context length is 16385 tokens. However,
    your messages resulted in 768010 tokens. Please reduce the length of the messages.",
    'type': 'invalid_request_error', 'param': 'messages', 'code': 'context_length_exceeded'}}

    Args:
        current_input(BaseMessage): User last input | 用户最后输入
        conversation(Optional[list[BaseMessage]]): Conversation history | 会话历史
        elements(Optional[list[DocElement]]): Document elements | 文档元素
        knowledge(Optional[str]): Related knowledge | 相关知识
        tools(Optional[list[BaseTool]]): Available tool list | 可用的工具列表
        intermediate_msgs(Optional[list[BaseMessage]]): Intermediate messages, it will be generated during llm work
            in chain, list tool's response or other system message, those messages will not save into memory, but
            usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中，
            但在链式工作中非常有用
        response_format(Optional[LLMResponseFormat]): Response format | 响应格式
            如果在此指定响应格式，会覆盖LLM的默认响应格式。

    Returns:
        LLMResult: The result of completion | 完成的结果
    """
    params = self.construct_request_params(
        current_input, conversation, elements, knowledge, tools, intermediate_msgs, response_format
    )
    with tracer.start_as_current_span("openai-chat", kind=SpanKind.CLIENT) as span:
        if self.stream:
            span_ctx = LLMEventContext(
                scene="LLM",
                entity_id=str(id(self)),
                desc="Asyncio | before llm streaming request",
                info=str(current_input.content),
                streaming=self.stream,
                token_usage=None,
                user_input=str(current_input.content),
                llm_model_name=self.name,
            )
            span.add_event(SpanEvent.BEFORE_LLM_STREAMING.value, span_ctx.model_dump())
            try:
                chat_completion_stream: AsyncStream[ChatCompletionChunk] = cast(
                    AsyncStream[ChatCompletionChunk], await self._async_client.chat.completions.create(**params)
                )
            except BadRequestError as e:
                body = cast(dict, e.body)
                if body.get("code") == "context_length_exceeded":
                    # 从错误中提取 token 信息 | Extract token info from error
                    current_size, target_size = self.extract_context_size_from_error(e)
                    # 将 OpenAI 的上下文超长异常转换为 ContextTooLargeError，由 Chain 层捕获并处理
                    raise ContextTooLargeError(
                        message=f"OpenAI context length exceeded: {body.get('message')}",
                        current_size=current_size,
                        target_size=target_size,
                        model_name=self.name,
                    )
                else:
                    raise e
            except Exception as e:
                warnings.warn(f"Unexpected error occurred during request to OpenAI API: {e}")
                raise e
            assert isinstance(chat_completion_stream, AsyncStream), "chat_completion is not Stream"
            chat_completion_finally = None
            async for res_chunk in chat_completion_stream:
                if not chat_completion_finally:
                    chat_completion_finally = res_chunk.model_copy(deep=True)  # 在拿到首个返回后，将其深拷贝
                else:
                    # 在拿到首个返回之后，将后续的chunk返回合并到首个返回中
                    self.merge_chat_completion_chunk(chat_completion_finally, res_chunk)
                span_ctx = LLMEventContext(
                    scene="LLM",
                    entity_id=str(id(self)),
                    desc="Asyncio | llm streaming response",
                    info=json.dumps(res_chunk.model_dump(), indent=2, ensure_ascii=False),
                    streaming=self.stream,
                    token_usage=None,
                    user_input=str(current_input.content),
                    llm_model_name=self.name,
                )
                span.add_event(SpanEvent.LLM_STREAMING.value, span_ctx.model_dump())
            span_ctx = LLMEventContext(
                scene="LLM",
                entity_id=str(id(self)),
                desc="Asyncio | after llm streaming response",
                info=json.dumps(chat_completion_finally.model_dump(), indent=2, ensure_ascii=False)
                if chat_completion_finally
                else "None",
                streaming=self.stream,
                token_usage=None,
                user_input=str(current_input.content),
                llm_model_name=self.name,
            )
            span.add_event(SpanEvent.AFTER_LLM_STREAMING.value, span_ctx.model_dump())
            if chat_completion_finally:
                # 在OpenAI的接口中，需要将delta中的内容追加到message中，同时需要修改object字段为chat.completion
                dict_res = chat_completion_finally.model_dump()
                dict_res["object"] = "chat.completion"
                for choice in dict_res["choices"]:
                    delta = choice.pop("delta")
                    choice["message"] = delta
                chat_completion = ChatCompletion(**dict_res)
            else:
                raise ValueError("chat_completion_finally is None")  # pragma: no cover
        else:
            try:
                chat_completion = await self._async_client.chat.completions.create(**params)
            except BadRequestError as e:
                body = cast(dict, e.body)
                if body.get("code") == "context_length_exceeded":
                    # 从错误中提取 token 信息 | Extract token info from error
                    current_size, target_size = self.extract_context_size_from_error(e)
                    # 将 OpenAI 的上下文超长异常转换为 ContextTooLargeError，由 Chain 层捕获并处理
                    raise ContextTooLargeError(
                        message=f"OpenAI context length exceeded: {body.get('message')}",
                        current_size=current_size,
                        target_size=target_size,
                        model_name=self.name,
                    )
                else:
                    raise e
            except Exception as e:
                warnings.warn(f"Unexpected error occurred during request to OpenAI API: {e}")
                raise e
    return self.construct_llm_result(chat_completion, params=params)

lying_to_openai_engineer staticmethod

lying_to_openai_engineer(messages: list[dict]) -> list[dict]

Lying to OpenAI engineer | 向OpenAI工程师撒谎

OpenAI的API工程中对消息结构进行了校验，比如要求所有的ToolReturn消息必须紧跟Assistant-ToolCall消息。 同时如果一个工具是yield迭代器的模式向外返回，后面的消息会被检查不满足上述条件从而触发BadRequestError(http-400）

此方法专门将请求消息做改造，使得OpenAI的工程接口无法检测到错误，从而可以继续调用。

我们希望随着OpenAI API的迭代，可以尽快去掉这个方法

Update At 2024-11-18

目前来看应该可以优化这个方法了，根据测试用例：tests/integration_tests/brain/chain/llms/test_mutil_tool_calls.py 11-18的更新内容 多ToolCalls调用可以不再添加中间消息，而是紧跟Assistant后进行展开。因此这个方法可以优化。但仍然需要注意，展开的ToolReturn消息中间不可以掺杂其它消息。

Parameters:

Name	Type	Description	Default
messages	list[dict]	Messages | 消息列表	required

Returns:

Type	Description
list[dict]	list[dict]: Modified messages | 修改后的消息列表

Source code in tfrobot/brain/chain/llms/openai.py

@staticmethod
def lying_to_openai_engineer(messages: list[dict]) -> list[dict]:
    """
    Lying to OpenAI engineer | 向OpenAI工程师撒谎

    OpenAI的API工程中对消息结构进行了校验，比如要求所有的ToolReturn消息必须紧跟Assistant-ToolCall消息。
    同时如果一个工具是yield迭代器的模式向外返回，后面的消息会被检查不满足上述条件从而触发BadRequestError(http-400）

    此方法专门将请求消息做改造，使得OpenAI的工程接口无法检测到错误，从而可以继续调用。

    我们希望随着OpenAI API的迭代，可以尽快去掉这个方法

    # Update At 2024-11-18

    目前来看应该可以优化这个方法了，根据测试用例：tests/integration_tests/brain/chain/llms/test_mutil_tool_calls.py 11-18的更新内容
    多ToolCalls调用可以不再添加中间消息，而是紧跟Assistant后进行展开。因此这个方法可以优化。但仍然需要注意，展开的ToolReturn消息中间不可以掺杂其它消息。

    Args:
        messages(list[dict]): Messages | 消息列表

    Returns:
        list[dict]: Modified messages | 修改后的消息列表
    """
    tool_call_cache = {}
    new_messages = []
    i = 0
    n = len(messages)
    while i < n:
        msg: dict = messages[i]
        if msg["role"] == "assistant" and msg.get("tool_calls"):
            # Assistant message with tool_calls
            tool_calls = msg["tool_calls"]
            for tc in tool_calls:
                tool_call_cache[tc["id"]] = tc
            tool_call_ids = [tc["id"] for tc in tool_calls]
            tool_returns = []
            interleaving_messages = []
            i += 1
            # Collect following messages
            while i < n:
                next_msg = messages[i]
                if next_msg["role"] in ["tool", "function"]:
                    if next_msg.get("tool_call_id") in tool_call_ids:
                        tool_returns.append(next_msg)
                    else:
                        break
                    i += 1
                elif next_msg["role"] == "assistant" and next_msg.get("tool_calls"):
                    # Next assistant message with tool_calls; process in next iteration
                    break
                else:
                    # Collect interleaving messages to append after tool returns
                    interleaving_messages.append(next_msg)
                    i += 1
            # Identify missing tool_returns and insert mock messages
            returned_tool_call_ids = [tr["tool_call_id"] for tr in tool_returns]
            missing_tool_call_ids = set(tool_call_ids) - set(returned_tool_call_ids)
            for missing_id in missing_tool_call_ids:
                mock_tool_return = {
                    "role": "tool",
                    "tool_call_id": missing_id,
                    "content": (
                        "The tool did not return properly. It may have encountered an error "
                        "or failed to execute successfully. Please try again later or verify the outcome through "
                        "other means."
                    ),
                }
                tool_returns.append(mock_tool_return)
            # Order tool_returns to match the order of tool_calls
            tool_return_dict = {tr["tool_call_id"]: tr for tr in tool_returns}
            ordered_tool_returns = [tool_return_dict[tc_id] for tc_id in tool_call_ids]
            # Append assistant message and its tool_returns
            new_messages.append(msg)
            new_messages.extend(ordered_tool_returns)
            # Append any interleaving messages after the tool_returns
            new_messages.extend(interleaving_messages)
        elif msg["role"] in ["tool", "function"]:
            # 一般所有的工具消息均被assistant处理逻辑回收了，但有一种情况会到这里，Yield模式，消息被拆分成多个消息返回，每段都用一个call_id，而上述逻辑仅回收第一个。因此要利用tool_call_cache来Mock一个Assistant消息，形成一个有效调用。
            # Mock an assistant message
            mock_assistant = {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    tool_call_cache.get(
                        msg["tool_call_id"],
                        {
                            "id": msg["tool_call_id"],
                            "type": "function",
                            "function": {"name": "unknown", "arguments": "{}"},
                        },
                    )
                ],
            }
            new_messages.append(mock_assistant)
            new_messages.append(msg)
            i += 1
        else:
            # Append messages without modification
            new_messages.append(msg)
            i += 1
    return new_messages

GPTWithTools

Bases: GPT

GPT基础类可以适配多数情况，但因为其工具是由外部传入，并且在注册到neural后可以自由扩展。因此带来了非常高的工具自由度。在以下场景下，我们需要一个更加简单的工具调用方式：

1. 比如Git操作流程，我们当前仅需要执行git diff操作，但通过召回工具集，git diff与git commit等工具均被召回，此时大模型非常容易在git diff后自动运行git commit。但需求场景中不允许此时调 用，git commit需要人工触发。此时我们需要一个更加简单的工具调用方式，只需要调用git diff，而不需要调用git commit。故而可以在GPTWithTools中进行简单的配置，只调用git diff，而不调用 git commit。

construct_request_params

construct_request_params(current_input: BaseMessage, conversation: Optional[list[BaseMessage]] = None, elements: Optional[list[DocElement]] = None, knowledge: Optional[str] = None, tools: Optional[list[BaseTool]] = None, intermediate_msgs: Optional[list[BaseMessage]] = None, response_format: Optional[LLMResponseFormat] = None) -> dict

Construct chat request parameters | 构造聊天请求参数

最终的会话排序如下： 1. 首条System Message 2. 用户历史会话内容 3. 当前输入前的提示 4. 当前输入 5. 当前输入后的提示 6. 中间消息

Parameters:

Name	Type	Description	Default
current_input	BaseMessage	Current input | 当前输入	required
conversation	Optional[list[BaseMessage]]	Conversation history | 会话历史	None
elements	Optional[list[DocElement]]	Document elements | 文档元素	None
knowledge	Optional[str]	Related knowledge | 相关知识	None
tools	Optional[list[BaseTool]]	Available tool list | 可用的工具列表	None
intermediate_msgs	Optional[list[BaseMessage]]	Intermediate messages, it will be generated during llm work in chain, list tool's response or other system message, those messages will not save into memory, but usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中， 但在链式工作中非常有用	None
response_format	Optional[LLMResponseFormat]	Response format | 响应格式 如果在此指定响应格式，会覆盖LLM的默认响应格式。	None

Returns:

Name	Type	Description
dict	dict	Request parameters | 请求参数

Source code in tfrobot/brain/chain/llms/openai.py

@overrides.override
def construct_request_params(
    self,
    current_input: BaseMessage,
    conversation: Optional[list[BaseMessage]] = None,
    elements: Optional[list[DocElement]] = None,
    knowledge: Optional[str] = None,
    tools: Optional[list[BaseTool]] = None,
    intermediate_msgs: Optional[list[BaseMessage]] = None,
    response_format: Optional[LLMResponseFormat] = None,
) -> dict:
    """
    Construct chat request parameters | 构造聊天请求参数

    最终的会话排序如下：
    1. 首条System Message
    2. 用户历史会话内容
    3. 当前输入前的提示
    4. 当前输入
    5. 当前输入后的提示
    6. 中间消息

    Args:
        current_input(BaseMessage): Current input | 当前输入
        conversation(Optional[list[BaseMessage]]): Conversation history | 会话历史
        elements(Optional[list[DocElement]]): Document elements | 文档元素
        knowledge(Optional[str]): Related knowledge | 相关知识
        tools(Optional[list[BaseTool]]): Available tool list | 可用的工具列表
        intermediate_msgs(Optional[list[BaseMessage]]): Intermediate messages, it will be generated during llm work
            in chain, list tool's response or other system message, those messages will not save into memory, but
            usefully during chain work | 中间消息，它将在链式工作中生成，列表工具的响应或其他系统消息，这些消息不会保存到记忆体中，
            但在链式工作中非常有用
        response_format(Optional[LLMResponseFormat]): Response format | 响应格式
            如果在此指定响应格式，会覆盖LLM的默认响应格式。

    Returns:
        dict: Request parameters | 请求参数
    """
    super_res = super().construct_request_params(
        current_input, conversation, elements, knowledge, tools, intermediate_msgs, response_format
    )
    if tools := super_res.get("tools"):
        for i in range(len(tools) - 1, -1, -1):
            tool_name = tools[i]["function"]["name"]
            if self.exclude_tools and tool_name in self.exclude_tools:
                tools.pop(i)
                continue
            if self.available_tools is not None and tool_name not in self.available_tools:
                tools.pop(i)
        if not super_res.get("tools"):
            del super_res["tools"]
            super_res["tool_choice"] = None  # 明确表示不使用任何工具，避免OpenAI接口报错
    return super_res

calculate_tokens

calculate_tokens(chat_completion: ChatCompletion, prompt: dict, model: str) -> CompletionUsage

Calculate the number of tokens in the chat completion | 计算聊天完成中的token数量

Parameters:

Name	Type	Description	Default
chat_completion	ChatCompletion	Chat completion from OpenAI SDK | 来自OpenAI SDK的聊天完成	required
prompt	dict	Prompt | 提示	required
model	str	Model name | 模型名称	required

Returns:

Name	Type	Description
int	CompletionUsage	Number of tokens | token数量

Source code in tfrobot/brain/chain/llms/openai.py

def calculate_tokens(chat_completion: ChatCompletion, prompt: dict, model: str) -> CompletionUsage:
    """
    Calculate the number of tokens in the chat completion | 计算聊天完成中的token数量

    Args:
        chat_completion(ChatCompletion): Chat completion from OpenAI SDK | 来自OpenAI SDK的聊天完成
        prompt(dict): Prompt | 提示
        model(str): Model name | 模型名称

    Returns:
        int: Number of tokens | token数量
    """
    enc = tiktoken.encoding_for_model(model)
    prompt_tokens = len(enc.encode(str(prompt)))
    completion_tokens = len(enc.encode(json.dumps(chat_completion.model_dump())))
    total_tokens = prompt_tokens + completion_tokens
    return CompletionUsage(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, total_tokens=total_tokens)