refactor(utils): remove meesage trasformation between OpenAI, AgentScope

Author: TaoShuchang

ajet/utils/msg_converter.py

@@ -25,219 +25,6 @@
 from typing import List, Dict, Any, Union
 
 
-# =============================================================================
-# OpenAI -> AgentScope conversion
-# =============================================================================
-
-def openai_to_agentscope_single(msg: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Convert a single OpenAI format message to AgentScope format.
-
-    Args:
-        msg: Message dict in OpenAI format
-
-    Returns:
-        Message dict in AgentScope format
-    """
-    role = msg.get("role", "user")
-    content = msg.get("content", "")
-    tool_calls = msg.get("tool_calls", [])
-    tool_call_id = msg.get("tool_call_id", "")
-
-    if tool_calls:
-        # Assistant message contains tool_calls -> convert to ToolUseBlock format
-        content_blocks = []
-        # If there's text content, add TextBlock first
-        if content:
-            content_blocks.append({"type": "text", "text": content})
-        # Convert each tool_call to ToolUseBlock
-        for tc in tool_calls:
-            func_info = tc.get("function", {}) if isinstance(tc.get("function"), dict) else {}
-            tool_use_block = {
-                "type": "tool_use",
-                "id": tc.get("id", ""),
-                "name": func_info.get("name", ""),
-                "input": func_info.get("arguments", "{}")
-            }
-            # Try to parse arguments as dict
-            if isinstance(tool_use_block["input"], str):
-                try:
-                    tool_use_block["input"] = json.loads(tool_use_block["input"])
-                except:
-                    pass
-            content_blocks.append(tool_use_block)
-        return {
-            "name": "assistant",
-            "role": "assistant",
-            "content": content_blocks
-        }
-
-    elif role == "tool" and tool_call_id:
-        # Tool return result -> convert to ToolResultBlock format
-        tool_result_block = {
-            "type": "tool_result",
-            "id": tool_call_id,
-            "output": content
-        }
-        return {
-            "name": "tool",
-            "role": "user",  # tool_result in AgentScope is treated as user message
-            "content": [tool_result_block]
-        }
-
-    else:
-        # Normal message, keep original format
-        return {
-            "name": role,
-            "role": role,
-            "content": content
-        }
-
-
-def openai_to_agentscope(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """
-    Convert OpenAI format message list to AgentScope format.
-
-    Args:
-        messages: Message list in OpenAI format
-
-    Returns:
-        Message list in AgentScope format
-    """
-    return [openai_to_agentscope_single(msg) for msg in messages]
-
-
-def openai_to_agentscope_grouped(timelines: List[List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:
-    """
-    Convert timelines (multi-turn conversation steps) from OpenAI format to AgentScope format.
-
-    Args:
-        timelines: List of List of dict in OpenAI format
-
-    Returns:
-        Trajectory data in AgentScope format
-    """
-    return [[openai_to_agentscope_single(msg) for msg in step] for step in timelines]
-
-
-# =============================================================================
-# AgentScope -> OpenAI conversion
-# =============================================================================
-
-def agentscope_to_openai_single(msg: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Convert a single AgentScope format message to OpenAI format.
-
-    Args:
-        msg: Message dict in AgentScope format
-
-    Returns:
-        Message dict in OpenAI format
-    """
-    role = msg.get("role", "user")
-    content = msg.get("content", "")
-
-    # If content is string, return directly
-    if isinstance(content, str):
-        return {
-            "role": role,
-            "content": content
-        }
-
-    # If content is list (AgentScope block format)
-    if isinstance(content, list):
-        text_parts = []
-        tool_calls = []
-        tool_call_id = ""
-        tool_output = ""
-        is_tool_result = False
-
-        for item in content:
-            if not isinstance(item, dict):
-                continue
-
-            item_type = item.get("type", "")
-
-            if item_type == "text":
-                # TextBlock
-                text_parts.append(item.get("text", ""))
-
-            elif item_type == "tool_use":
-                # ToolUseBlock -> tool_calls
-                arguments = item.get("input", {})
-                if isinstance(arguments, dict):
-                    arguments = json.dumps(arguments, ensure_ascii=False)
-                tool_calls.append({
-                    "id": item.get("id", ""),
-                    "type": "function",
-                    "function": {
-                        "name": item.get("name", ""),
-                        "arguments": arguments
-                    }
-                })
-
-            elif item_type == "tool_result":
-                # ToolResultBlock -> tool response
-                is_tool_result = True
-                tool_call_id = item.get("id", "")
-                output = item.get("output", "")
-                if isinstance(output, str):
-                    tool_output += output
-                else:
-                    tool_output += str(output)
-
-        # Build OpenAI format based on parsing result
-        if is_tool_result and tool_call_id:
-            return {
-                "role": "tool",
-                "content": tool_output,
-                "tool_call_id": tool_call_id
-            }
-        elif tool_calls:
-            result = {
-                "role": "assistant",
-                "content": "".join(text_parts) if text_parts else "",
-                "tool_calls": tool_calls
-            }
-            return result
-        else:
-            return {
-                "role": role,
-                "content": "".join(text_parts) if text_parts else ""
-            }
-
-    # Otherwise, return as is
-    return {
-        "role": role,
-        "content": str(content) if content else ""
-    }
-
-
-def agentscope_to_openai(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """
-    Convert AgentScope format message list to OpenAI format.
-
-    Args:
-        messages: Message list in AgentScope format
-
-    Returns:
-        Message list in OpenAI format
-    """
-    return [agentscope_to_openai_single(msg) for msg in messages]
-
-
-def agentscope_to_openai_grouped(timelines: List[List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:
-    """
-    Convert timelines (multi-turn conversation steps) from AgentScope format to OpenAI format.
-
-    Args:
-        timelines: List of List of dict in AgentScope format
-
-    Returns:
-        Trajectory data in OpenAI format
-    """
-    return [[agentscope_to_openai_single(msg) for msg in step] for step in timelines]
-
 
 # =============================================================================
 # ExtendedMessage -> OpenAI conversion (backward compatible functions)