| """ |
| 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): |
| |
| 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: |
| |
| 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_parts, content_warnings = self._parse_content(raw_msg.get("content")) |
| warnings.extend(content_warnings) |
|
|
| |
| tool_calls = self._parse_tool_calls(raw_msg.get("tool_calls")) |
|
|
| |
| tool_results = self._parse_tool_results(raw_msg) |
|
|
| |
| 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): |
| |
| 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": |
| |
| 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": |
| |
| 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", "") |
|
|
| |
| 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 = [] |
|
|
| |
| 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 |
|
|