openassistant/app/core/runner/utils/message_util.py

"""
This module provides utility functions for working with messages in the OpenAI API.

Functions:
- new_message(role: str, content: str) -> dict: Creates a new message with the specified role and content.
- system_message(content: str) -> dict: Creates a system message with the specified content.
- user_message(content: str) -> dict: Creates a user message with the specified content.
- assistant_message(content: str) -> dict: Creates an assistant message with the specified content.
- tool_calls(tool_calls) -> dict: Creates a message with assistant tool calls.
- tool_call_result(id, content) -> dict: Creates a tool call result message with the specified ID and content.
- is_tool_call(message: ChatCompletionMessage) -> bool: Checks if a message is a tool call.
"""

from openai.types.chat import ChatCompletionMessage, ChatCompletionMessageToolCall
from openai.types.chat.chat_completion_message_tool_call import Function


def new_message(role: str, content: str, reasoning_content=None):
    if role != "user" and role != "system" and role != "assistant":
        raise ValueError(f"Invalid role {role}")
    if reasoning_content is None:
        return {"role": role, "content": content}
    else:
        return {"role": role, "content": content, "reasoning_content": reasoning_content}


def system_message(content: str):
    return new_message("system", content)


def user_message(content: str):
    return new_message("user", content)


def assistant_message(content: str):
    return new_message("assistant", content)


def tool_calls(tool_calls, reasoning_content=None):
    if reasoning_content is None:
        return {"role": "assistant", "tool_calls": tool_calls}
    else:
        return {"role": "assistant", "tool_calls": tool_calls, "reasoning_content": reasoning_content}


def tool_call_result(id, content):
    return {"role": "tool", "tool_call_id": id, "content": content}


def is_tool_call(message: ChatCompletionMessage) -> bool:
    return bool(message.tool_calls)


def merge_tool_call_delta(tool_calls, tool_call_delta):
    index = getattr(tool_call_delta, "index", None)
    if index is None:
        # 尝试通过 id 匹配已有工具调用
        if tool_call_delta.id:
            for i, tc in enumerate(tool_calls):
                if tc.id == tool_call_delta.id:
                    index = i
                    break
        # 若仍找不到，根据列表长度决定
        if index is None:
            index = 0 if len(tool_calls) == 0 else len(tool_calls) - 1

    if len(tool_calls) - 1 >= index:
        tool_call = tool_calls[index]
        # 确保 arguments 为字符串，避免 None 导致拼接失败
        tool_call.function.arguments = (tool_call.function.arguments or "") + (
            tool_call_delta.function.arguments or ""
        )
    else:
        tool_call = ChatCompletionMessageToolCall(
            id=tool_call_delta.id,
            function=Function(
                name=tool_call_delta.function.name,
                arguments=tool_call_delta.function.arguments or "",
            ),
            type=tool_call_delta.type,
        )
        tool_calls.append(tool_call)

    # 调试输出可保留，但建议移除或降级为 debug
    print("=" * 40)
    print(tool_calls)