Rhinox13/chatapi

1from __future__ import annotations
2
3import time
4import uuid
5from typing import Any
6
7from .thinking import answer_text, has_thinking, split_thinking_parts
8
9
10def _estimate_tokens(text: str) -> int:
11    text = text.strip()
12    if not text:
13        return 0
14    ascii_tokens = len(text.split())
15    cjk_chars = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff")
16    return max(ascii_tokens, 1) + cjk_chars // 2
17
18
19def estimate_usage(input_text: str, output_text: str) -> dict[str, int]:
20    input_tokens = _estimate_tokens(input_text)
21    output_tokens = _estimate_tokens(output_text)
22    return {
23        "input_tokens": input_tokens,
24        "output_tokens": output_tokens,
25        "total_tokens": input_tokens + output_tokens,
26    }
27
28
29def build_openai_response(
30    *,
31    response_id: str,
32    model: str,
33    conversation_id: str,
34    assistant_text: str,
35    usage: dict[str, int] | None,
36    status: str = "completed",
37    output_items: list[dict[str, Any]] | None = None,
38    output_text: str | None = None,
39) -> dict[str, Any]:
40    created_at = int(time.time())
41    normalized_output_items = output_items
42    if normalized_output_items is None:
43        normalized_output_items = _build_default_output_items(assistant_text)
44        output_text_source = assistant_text if output_text is None else output_text
45        normalized_output_text = (
46            answer_text(output_text_source)
47            if has_thinking(output_text_source)
48            else output_text_source
49        )
50    else:
51        normalized_output_text = assistant_text if output_text is None else output_text
52    return {
53        "id": response_id,
54        "object": "response",
55        "created_at": created_at,
56        "status": status,
57        "model": model,
58        "conversation_id": conversation_id,
59        "output": normalized_output_items,
60        "output_text": normalized_output_text,
61        "usage": usage,
62    }
63
64
65def _build_default_output_items(assistant_text: str) -> list[dict[str, Any]]:
66    parts = split_thinking_parts(assistant_text)
67    if not any(part["type"] == "thinking" for part in parts):
68        return [
69            {
70                "id": f"msg_{uuid.uuid4().hex[:24]}",
71                "type": "message",
72                "role": "assistant",
73                "content": [
74                    {
75                        "type": "output_text",
76                        "text": assistant_text,
77                    }
78                ],
79            }
80        ]
81    if not parts:
82        parts = [{"type": "answer", "text": ""}]
83
84    output_items: list[dict[str, Any]] = []
85    answer_parts: list[str] = []
86    for part in parts:
87        text = part["text"]
88        if part["type"] == "thinking":
89            output_items.append(
90                {
91                    "id": f"rs_{uuid.uuid4().hex[:24]}",
92                    "type": "reasoning",
93                    "status": "completed",
94                    "content": [
95                        {
96                            "type": "reasoning_text",
97                            "text": text,
98                        }
99                    ],
100                    "summary": [
101                        {
102                            "type": "summary_text",
103                            "text": text,
104                        }
105                    ],
106                }
107            )
108        else:
109            answer_parts.append(text)
110
111    normalized_answer_text = "\n\n".join(part for part in answer_parts if part).strip()
112    if normalized_answer_text or not output_items:
113        output_items.append(
114            {
115                "id": f"msg_{uuid.uuid4().hex[:24]}",
116                "type": "message",
117                "status": "completed",
118                "role": "assistant",
119                "content": [
120                    {
121                        "type": "output_text",
122                        "annotations": [],
123                        "text": normalized_answer_text,
124                    }
125                ],
126            }
127        )
128    return output_items
129
130
131def build_openai_error(
132    message: str,
133    code: str = "bad_request",
134    status: int = 400,
135) -> tuple[dict[str, Any], int]:
136    return (
137        {
138            "error": {
139                "message": message,
140                "type": code,
141                "code": code,
142            }
143        },
144        status,
145    )
146