Delete request_converter.py

request_converter.py

@@ -1,185 +0,0 @@
-"""
-请求转换器模块
-处理OpenAI格式和Warp格式之间的转换
-"""
-from typing import List, Dict, Any, Optional, Tuple
-from dataclasses import dataclass
-
-from utils import Utils
-
-
-@dataclass
-class OpenAIMessage:
-    """OpenAI消息格式"""
-    role: str
-    content: str
-
-
-@dataclass 
-class OpenAIRequest:
-    """OpenAI请求格式"""
-    model: str
-    messages: List[OpenAIMessage]
-    stream: bool = False
-    temperature: float = 1.0
-    max_tokens: Optional[int] = None
-
-
-@dataclass
-class OpenAIChoice:
-    """OpenAI响应选择"""
-    index: int
-    message: Optional[OpenAIMessage] = None
-    delta: Optional[Dict[str, Any]] = None
-    finish_reason: Optional[str] = None
-
-
-@dataclass
-class OpenAIResponse:
-    """OpenAI响应格式"""
-    id: str
-    object: str
-    created: int
-    model: str
-    choices: List[OpenAIChoice]
-    usage: Optional[Dict[str, int]] = None
-
-
-class RequestConverter:
-    """OpenAI请求到Warp请求的转换器"""
-    
-    @staticmethod
-    def parse_openai_request(request_data: dict) -> OpenAIRequest:
-        """解析OpenAI请求格式"""
-        messages = []
-        for msg in request_data.get("messages", []):
-            content = msg.get("content", "")
-            
-            # 处理可能的列表格式内容
-            if isinstance(content, list):
-                text_content = ""
-                for item in content:
-                    if isinstance(item, dict) and item.get("type") == "text":
-                        text_content += item.get("text", "")
-                content = text_content
-            
-            messages.append(OpenAIMessage(
-                role=msg.get("role", "user"),
-                content=str(content)
-            ))
-        
-        return OpenAIRequest(
-            model=request_data.get("model", "gemini-2.0-flash"),
-            messages=messages,
-            stream=request_data.get("stream", False),
-            temperature=request_data.get("temperature", 1.0),
-            max_tokens=request_data.get("max_tokens")
-        )
-    
-    @staticmethod 
-    def process_message_history(messages: List[Dict]) -> Tuple[List[Dict], str]:
-        """处理消息历史，返回(合并后的消息组列表, 当前消息)"""
-        if not messages:
-            return [], "Hello"
-        
-        # 确保最后一句话是user  
-        if messages[-1].role != "user":
-            raise ValueError("最后一句话必须是user消息")
-
-        # 当前消息（最后一句）
-        current_message = messages[-1].content
-
-        if len(messages) == 1:
-            return [], current_message
-
-        # 处理除最后一句之外的所有消息，合并连续相同的role
-        merged_groups = []
-        i = 0
-        
-        while i < len(messages) - 1:  # 排除最后一句
-            current_role = messages[i].role
-            
-            # 如果是system，转换为user
-            if current_role == "system":
-                current_role = "user"
-            
-            # 收集当前内容
-            current_content = messages[i].content
-            
-            # 合并连续相同的role
-            j = i + 1
-            while j < len(messages) - 1:  # 排除最后一句
-                next_role = messages[j].role
-                if next_role == "system":
-                    next_role = "user"
-                
-                if next_role == current_role:
-                    # 合并内容
-                    current_content += "\n" + messages[j].content
-                    j += 1
-                else:
-                    break
-            
-            # 添加合并后的消息组
-            merged_groups.append({"role": current_role, "content": current_content})
-            i = j
-        
-        return merged_groups, current_message
-    
-    @staticmethod
-    def create_openai_response(content: str, model: str, request_id: str, 
-                             is_stream: bool = False, finish_reason: str = None) -> dict:
-        """创建OpenAI格式的响应"""
-        if is_stream:
-            choice = {
-                "index": 0,
-                "delta": {"content": content} if content else {},
-                "finish_reason": finish_reason
-            }
-            
-            return {
-                "id": request_id,
-                "object": "chat.completion.chunk",
-                "created": Utils.get_current_timestamp(),
-                "model": model,
-                "choices": [choice]
-            }
-        else:
-            choice = {
-                "index": 0,
-                "message": {
-                    "role": "assistant",
-                    "content": content
-                },
-                "finish_reason": "stop"
-            }
-            
-            return {
-                "id": request_id,
-                "object": "chat.completion",
-                "created": Utils.get_current_timestamp(),
-                "model": model,
-                "choices": [choice],
-                "usage": {
-                    "prompt_tokens": 100,
-                    "completion_tokens": len(content.split()) if content else 0,
-                    "total_tokens": 100 + (len(content.split()) if content else 0)
-                }
-            }
-    
-    @staticmethod
-    def create_error_response(error_message: str, model: str, request_id: str = None) -> dict:
-        """创建错误响应"""
-        if not request_id:
-            request_id = Utils.generate_request_id()
-            
-        return {
-            "id": request_id,
-            "object": "error",
-            "created": Utils.get_current_timestamp(),
-            "model": model,
-            "error": {
-                "message": error_message,
-                "type": "invalid_request_error"
-            }
-        }