Spaces:
Sleeping
Sleeping
File size: 2,719 Bytes
09fa93e b0394f8 09fa93e b0394f8 09fa93e b0394f8 09fa93e 864f881 |
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 |
import json
import time
import typing
from agent_server.openai_schemas import ChatMessage
def normalize_content_to_text(content: typing.Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, (bytes, bytearray)):
try:
return content.decode("utf-8", errors="ignore")
except Exception:
return str(content)
if isinstance(content, list):
parts = []
for item in content:
if (
isinstance(item, dict)
and item.get("type") == "text"
and isinstance(item.get("text"), str)
):
parts.append(item["text"])
else:
try:
parts.append(json.dumps(item, ensure_ascii=False))
except Exception:
parts.append(str(item))
return "\n".join(parts)
if isinstance(content, dict):
try:
return json.dumps(content, ensure_ascii=False)
except Exception:
return str(content)
return str(content)
def messages_to_task(messages: typing.List[ChatMessage]) -> str:
system_parts = [
normalize_content_to_text(m.get("content", ""))
for m in messages
if m.get("role") == "system"
]
user_parts = [
normalize_content_to_text(m.get("content", ""))
for m in messages
if m.get("role") == "user"
]
assistant_parts = [
normalize_content_to_text(m.get("content", ""))
for m in messages
if m.get("role") == "assistant"
]
sys_txt = "\n".join([s for s in system_parts if s]).strip()
history = ""
if assistant_parts:
history = "\n\nPrevious assistant replies (for context):\n" + "\n---\n".join(
assistant_parts
)
last_user = user_parts[-1] if user_parts else ""
return f"{sys_txt}\nTask:\n{last_user}\n{history}".strip()
def openai_response(message_text: str, model_name: str) -> typing.Dict[str, typing.Any]:
now = int(time.time())
return {
"id": f"chatcmpl-smol-{now}",
"object": "chat.completion",
"created": now,
"model": model_name,
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": message_text},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
}
def sse_headers() -> dict:
return {
"Cache-Control": "no-cache, no-transform",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
def now_ts() -> int:
return int(time.time())
|