File size: 3,055 Bytes
e0b68d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | #!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Simplified OpenAI message preprocessing utilities."""
import json
from typing import Any, Dict, List
# ---------------------------------------------------------------------------
# Content extraction helper
# ---------------------------------------------------------------------------
def _extract_text_from_content(
content: Any,
*,
include_structured: bool = False,
) -> str:
"""Extract text parts from OpenAI-compatible content payloads."""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: List[str] = []
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
parts.append(str(item.get("text", "")))
elif include_structured and isinstance(item, dict):
try:
parts.append(json.dumps(item, ensure_ascii=False))
except Exception:
parts.append(str(item))
return " ".join(part for part in parts if part).strip()
if content is None:
return ""
try:
return json.dumps(content, ensure_ascii=False)
except Exception:
return str(content)
# ---------------------------------------------------------------------------
# Last user text extraction
# ---------------------------------------------------------------------------
def extract_last_user_text(messages: List[Dict[str, Any]]) -> str:
"""Extract the last user text from the original OpenAI message history."""
for message in reversed(messages):
if message.get("role") != "user":
continue
content = _extract_text_from_content(message.get("content"))
if content:
return content
return ""
# ---------------------------------------------------------------------------
# Simplified message preprocessing
# ---------------------------------------------------------------------------
def preprocess_openai_messages(
messages: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Normalize OpenAI history into shapes accepted by the upstream service.
Simplified conversions:
- ``developer`` role -> ``system``
- Everything else passes through unchanged
Args:
messages: OpenAI-format message list (already model_dump'd).
Returns:
Normalized message list suitable for upstream consumption.
"""
normalized: List[Dict[str, Any]] = []
for message in messages:
if not isinstance(message, dict):
continue
role = message.get("role")
# developer -> system
if role == "developer":
converted = dict(message)
converted["role"] = "system"
normalized.append(converted)
continue
# Everything else passes through unchanged
normalized.append(dict(message))
return normalized
|