""" OpenAI 对话格式解析器 支持 OpenAI messages、content parts、tool calls/results 的结构化解析。 """ import json import logging import uuid from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger(__name__) class OpenAIMessageParser: """OpenAI 对话格式解析器""" def parse_json(self, json_data: Any) -> Dict[str, Any]: """ 解析 JSON 数据为对话视图模型 Args: json_data: JSON 数据（dict 或 list） Returns: 对话视图模型 { "conversation_id": str, "messages": List[dict], "raw_warnings": List[str], } """ conversation_id = str(uuid.uuid4())[:8] messages = [] raw_warnings = [] # 处理不同的输入格式 if isinstance(json_data, dict): # 检查是否有 messages 字段 if "messages" in json_data: conversation_id = json_data.get("conversation_id", conversation_id) raw_messages = json_data["messages"] if isinstance(raw_messages, list): messages, raw_warnings = self._parse_messages(raw_messages) else: raw_warnings.append("messages 字段不是数组") else: # 尝试将整个 dict 作为单条消息解析 msg, warnings = self._parse_single_message(json_data, 0) if msg: messages.append(msg) raw_warnings.extend(warnings) elif isinstance(json_data, list): # 直接作为消息列表解析 messages, raw_warnings = self._parse_messages(json_data) else: raw_warnings.append(f"不支持的 JSON 类型: {type(json_data).__name__}") return { "conversation_id": conversation_id, "messages": messages, "raw_warnings": raw_warnings, } def parse_jsonl(self, text: str) -> Dict[str, Any]: """ 解析 JSONL 格式 Args: text: JSONL 文本 Returns: 对话视图模型 """ conversation_id = str(uuid.uuid4())[:8] raw_messages = [] raw_warnings = [] for line_num, line in enumerate(text.strip().split("\n"), 1): line = line.strip() if not line: continue try: data = json.loads(line) raw_messages.append(data) except json.JSONDecodeError as e: raw_warnings.append(f"第 {line_num} 行 JSON 解析失败: {e}") messages, warnings = self._parse_messages(raw_messages) raw_warnings.extend(warnings) return { "conversation_id": conversation_id, "messages": messages, "raw_warnings": raw_warnings, } def _parse_messages(self, raw_messages: List[Any]) -> Tuple[List[dict], List[str]]: """ 解析消息列表 Args: raw_messages: 原始消息列表 Returns: (解析后的消息列表, 警告列表) """ messages = [] warnings = [] for idx, raw_msg in enumerate(raw_messages): if not isinstance(raw_msg, dict): warnings.append(f"消息 {idx} 不是对象，跳过") continue msg, msg_warnings = self._parse_single_message(raw_msg, idx) if msg: messages.append(msg) warnings.extend(msg_warnings) return messages, warnings def _parse_single_message(self, raw_msg: dict, index: int) -> Tuple[Optional[dict], List[str]]: """ 解析单条消息 Args: raw_msg: 原始消息 index: 消息索引 Returns: (解析后的消息, 警告列表) """ warnings = [] # 提取基本字段 role = raw_msg.get("role", "unknown") msg_id = raw_msg.get("id", f"msg_{index}") # 解析 content content_parts, content_warnings = self._parse_content(raw_msg.get("content")) warnings.extend(content_warnings) # 解析 tool_calls tool_calls = self._parse_tool_calls(raw_msg.get("tool_calls")) # 解析 tool results (从 tool_call_id 字段判断) tool_results = self._parse_tool_results(raw_msg) # 提取 metadata metadata = {} for key in ["name", "function_call", "tool_call_id", "finish_reason", "usage"]: if key in raw_msg: metadata[key] = raw_msg[key] return { "id": msg_id, "role": role, "content_parts": content_parts, "tool_calls": tool_calls, "tool_results": tool_results, "metadata": metadata, "raw_index": index, }, warnings def _parse_content(self, content: Any) -> Tuple[List[dict], List[str]]: """ 解析 content 字段 Args: content: content 字段值 Returns: (content parts 列表, 警告列表) """ parts = [] warnings = [] if content is None: return parts, warnings if isinstance(content, str): # 纯文本 if content.strip(): parts.append({ "type": "text", "text": content.strip(), }) elif isinstance(content, list): # 数组格式（OpenAI 多模态） for item in content: if isinstance(item, dict): part, item_warnings = self._parse_content_part(item) if part: parts.append(part) warnings.extend(item_warnings) elif isinstance(item, str): if item.strip(): parts.append({ "type": "text", "text": item.strip(), }) else: warnings.append(f"不支持的 content 类型: {type(content).__name__}") return parts, warnings def _parse_content_part(self, item: dict) -> Tuple[Optional[dict], List[str]]: """ 解析单个 content part Args: item: content part 对象 Returns: (解析后的 part, 警告列表) """ warnings = [] item_type = item.get("type", "") if item_type == "text": text = item.get("text", "") if text.strip(): return { "type": "text", "text": text.strip(), }, warnings elif item_type == "thinking": thinking = item.get("thinking", "") if thinking.strip(): return { "type": "thinking", "text": thinking.strip(), }, warnings elif item_type == "image_url": image_url = item.get("image_url", {}) url = image_url.get("url", "") if isinstance(image_url, dict) else "" return { "type": "image", "url": url, }, warnings elif item_type == "tool_use": # Anthropic 格式的 tool_use return { "type": "tool_use", "tool_name": item.get("name", "未知工具"), "tool_input": item.get("input", {}), "tool_use_id": item.get("id", ""), }, warnings elif item_type == "tool_result": # Anthropic 格式的 tool_result result_content = item.get("content", "") is_error = item.get("is_error", False) return { "type": "tool_result", "tool_use_id": item.get("tool_use_id", ""), "content": result_content, "is_error": is_error, }, warnings else: warnings.append(f"不支持的 content part 类型: {item_type}") return None, warnings def _parse_tool_calls(self, tool_calls: Any) -> List[dict]: """ 解析 tool_calls 字段（OpenAI 格式） Args: tool_calls: tool_calls 字段值 Returns: 解析后的 tool calls 列表 """ if not tool_calls or not isinstance(tool_calls, list): return [] result = [] for tc in tool_calls: if not isinstance(tc, dict): continue tc_id = tc.get("id", "") tc_type = tc.get("type", "function") function = tc.get("function", {}) if isinstance(function, dict): func_name = function.get("name", "") func_args = function.get("arguments", "") # 尝试解析 arguments JSON if isinstance(func_args, str): try: func_args = json.loads(func_args) except json.JSONDecodeError: pass result.append({ "id": tc_id, "type": tc_type, "function_name": func_name, "function_arguments": func_args, }) return result def _parse_tool_results(self, raw_msg: dict) -> List[dict]: """ 解析 tool results（从消息中提取） Args: raw_msg: 原始消息 Returns: tool results 列表 """ results = [] # OpenAI 格式：tool_call_id 字段 if "tool_call_id" in raw_msg: results.append({ "tool_call_id": raw_msg["tool_call_id"], "content": raw_msg.get("content", ""), "is_error": False, }) return results # 全局单例 _parser: Optional[OpenAIMessageParser] = None def get_openai_parser() -> OpenAIMessageParser: """获取 OpenAI 对话解析器全局单例""" global _parser if _parser is None: _parser = OpenAIMessageParser() return _parser