GPT LLM

GPT

Bases: ChatLLM[dict, ChatCompletion, ChatCompletionChunk]

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):
            extra_parts: list[dict] = []  # PDF→图片降级时，一个 part 会展开为多个 image_url parts
            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 — 3-tier fallback strategy
                    # 级别1: supports_pdf → 原生 file 格式 | Level 1: native file format
                    # 级别2: supports_vision → PDF→图片转换 | Level 2: PDF-to-image conversion
                    # 级别3: 纯文本降级 | Level 3: text fallback
                    pdf_url_data = part.pop("pdf_url", {})
                    pdf_url = pdf_url_data.get("url", "")

                    if not llm_meta.capabilities.supports_pdf and not llm_meta.capabilities.supports_vision:
                        # 级别3: 模型既不支持PDF也不支持视觉，降级为文本
                        # Level 3: Model supports neither PDF nor vision, fallback to text
                        part["type"] = "text"
                        part["text"] = (
                            pdf_url_data.get("detail")
                            or "当前模型不支持PDF文件输入，如果此PDF内容异常重要，你可以提示用户切换模型或提取PDF文本内容。如果当前PDF信息可有可无，不影响判断，可以忽略此消息。"
                        )
                    elif llm_meta.capabilities.supports_pdf:
                        # 级别1: 原生PDF支持，使用OpenAI file格式
                        # Level 1: Native PDF support, use OpenAI file format
                        if not is_base64(pdf_url):
                            result = self.load_file_from_uri(pdf_url)
                            if result:
                                file_data, file_mime_type = result
                                part["type"] = "file"
                                part["file"] = {"file_data": f"data:{file_mime_type};base64,{file_data.decode()}"}
                                part["file"]["filename"] = pdf_url_data.get("detail") or file_mime_type
                            else:
                                warnings.warn(f"Failed to load PDF from URI: {pdf_url}")
                                part["type"] = "text"
                                part["text"] = (
                                    pdf_url_data.get("detail")
                                    or f"PDF文件加载失败: {pdf_url}，如果此PDF内容异常重要，你可以提示用户切换模型或提取PDF文本内容。"
                                )
                        else:
                            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}
                            part["file"]["filename"] = pdf_url_data.get("detail") or "application/pdf"
                    else:
                        # 级别2: 模型支持视觉但不支持原生PDF → 将PDF转换为图片
                        # Level 2: Model supports vision but not native PDF → convert PDF to images
                        try:
                            pdf_bytes = self._resolve_pdf_bytes(pdf_url)
                            image_base64_list = convert_pdf_to_images_base64(pdf_bytes, dpi=200, max_pages=10)

                            if image_base64_list:
                                # 第一张图片替换当前 part，其余追加到 extra_parts
                                # First image replaces current part, rest appended to extra_parts
                                part["type"] = "image_url"
                                part["image_url"] = {"url": image_base64_list[0], "detail": "auto"}
                                for img_b64 in image_base64_list[1:]:
                                    extra_parts.append(
                                        {"type": "image_url", "image_url": {"url": img_b64, "detail": "auto"}}
                                    )
                            else:
                                warnings.warn(f"PDF转图片失败: {pdf_url}")
                                part["type"] = "text"
                                part["text"] = (
                                    pdf_url_data.get("detail")
                                    or f"PDF文件转换失败: {pdf_url if not is_base64(pdf_url) else 'Base64编码-已省略'}"
                                )
                        except Exception as e:
                            warnings.warn(f"PDF文件处理失败: {e}")
                            part["type"] = "text"
                            part["text"] = (
                                pdf_url_data.get("detail")
                                or f"PDF文件处理失败: {pdf_url if not is_base64(pdf_url) else 'Base64编码-已省略'}，错误: {e}"
                            )
            if extra_parts:
                msg_dict["content"].extend(extra_parts)
    # OpenAI接口如果有设置用户名，那用户名需要满足OpenAI的正则要求，不允许出现一些特殊字符，因此需要进行转换
    if msg_dict.get("name"):
        msg_dict["name"] = replace_forbidden_chars(msg_dict["name"])
    return msg_dict

construct_llm_result

construct_llm_result(
    raw_completion: ChatCompletion,
    *,
    params: dict,
    response_format: Optional[LLMResponseFormat] = None
) -> LLMResult

Construct LLM result | 构造LLM结果

Parameters:

Name	Type	Description	Default
raw_completion	ChatCompletion	Chat completion from OpenAI SDK | 聊天完成，来自OpenAI SDK的定义	required
params	dict	Request parameters | 请求参数	required
response_format	Optional[LLMResponseFormat]	Response format | 响应格式（GPT 原生支持，当前未使用）	None

Returns:

Name	Type	Description
LLMResult	LLMResult	LLM result | LLM结果

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

def construct_llm_result(
    self, raw_completion: ChatCompletion, *, params: dict, response_format: Optional[LLMResponseFormat] = None
) -> LLMResult:
    """
    Construct LLM result | 构造LLM结果

    Args:
        raw_completion(ChatCompletion): Chat completion from OpenAI SDK | 聊天完成，来自OpenAI SDK的定义
        params(dict): Request parameters | 请求参数
        response_format(Optional[LLMResponseFormat]): Response format | 响应格式（GPT 原生支持，当前未使用）

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

        try:
            raw_completion.usage = calculate_tokens(chat_completion=raw_completion, prompt=params, model=self.name)
        except Exception as e:
            warnings.warn(f"Failed to calculate tokens usage: {e}")
            raw_completion.usage = CompletionUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)
    # 构建Generation
    generations: list[Generation] = []
    for choice in raw_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": raw_completion.id,
        "created": raw_completion.created,
        "model": self.name,  # 注意这里不能使用 raw_completion.model 因为比如 gpt-4o-mini 请求返回，会是其实际指向的model-name: gpt-4o-mini-2024-07-18 这会导致内部分析混乱
        "system_fingerprint": raw_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": raw_completion.usage.prompt_tokens,
        "completion_tokens": raw_completion.usage.completion_tokens,
        "total_tokens": raw_completion.usage.total_tokens,
        "completion_cost": raw_completion.usage.completion_tokens * self.output_price / 1000.0,
        "prompt_cost": raw_completion.usage.prompt_tokens * self.input_price / 1000.0,
        "total_cost": (
            raw_completion.usage.completion_tokens * self.output_price
            + raw_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

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)