Rhinox13/chatapi
0
1from __future__ import annotations
2
3import json
4import re
5import uuid
6from dataclasses import dataclass
7from typing import Any
8
9from ..core.auth import AuthContext
10from ..repositories import ConversationStore
11
12
13_IMAGE_URL_RE = re.compile(r"^(?:https?://[^\s\"']+)?/api/uploads/imgs/[A-Za-z0-9._-]+(?:\?.*)?$", re.IGNORECASE)
14
15
16@dataclass(frozen=True)
17class ResolvedConversation:
18 conversation: Any
19 source: str
20
21
22def normalize_message_text(value: str) -> str:
23 return value.replace("\r\n", "\n").replace("\\r\\n", "\n").replace("\\n", "\n")
24
25
26def build_protocol_response_id(request_format: str, fallback_request_id: str) -> str:
27 if request_format == "chat_completions":
28 return f"chatcmpl_{uuid.uuid4().hex}"
29 if request_format == "anthropic_messages":
30 return f"msg_{uuid.uuid4().hex[:24]}"
31 return fallback_request_id
32
33
34def response_input_payload(data: dict[str, Any]) -> Any:
35 if "input" in data:
36 return data["input"]
37 if "messages" in data:
38 return data["messages"]
39 return data
40
41
42def chat_input_payload(data: dict[str, Any]) -> Any:
43 return data.get("messages", [])
44
45
46def anthropic_input_payload(data: dict[str, Any]) -> Any:
47 payload: list[Any] = []
48 system_prompt = data.get("system")
49 if isinstance(system_prompt, str) and system_prompt.strip():
50 payload.append({"role": "system", "content": system_prompt})
51 elif isinstance(system_prompt, list) and system_prompt:
52 payload.append({"role": "system", "content": system_prompt})
53 messages = data.get("messages", [])
54 if isinstance(messages, list):
55 payload.extend(messages)
56 return payload
57
58
59def request_input_payload(data: dict[str, Any], request_format: str) -> Any:
60 if request_format == "chat_completions":
61 return chat_input_payload(data)
62 if request_format == "anthropic_messages":
63 return anthropic_input_payload(data)
64 return response_input_payload(data)
65
66
67def _is_image_reference_string(value: str) -> bool:
68 return value.startswith("data:image/") or bool(_IMAGE_URL_RE.match(value.strip()))
69
70
71def serialize_content(value: Any) -> str:
72 if isinstance(value, str):
73 return value
74 return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
75
76
77def canonical_json(value: Any) -> str:
78 return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
79
80
81def _assistant_request_content(item: dict[str, Any]) -> str:
82 tool_calls = item.get("tool_calls")
83 if isinstance(tool_calls, list) and tool_calls:
84 first_call = tool_calls[0] if isinstance(tool_calls[0], dict) else {}
85 function_payload = (
86 first_call.get("function")
87 if isinstance(first_call, dict) and isinstance(first_call.get("function"), dict)
88 else {}
89 )
90 tool_name = str(function_payload.get("name", "")).strip()
91 arguments = serialize_content(function_payload.get("arguments", ""))
92 if tool_name:
93 return f"{tool_name}({arguments})"
94 return serialize_content(tool_calls)
95 content = item.get("content")
96 if content is None:
97 return ""
98 return serialize_content(content)
99
100
101def extract_text_content(node: Any) -> str:
102 parts: list[str] = []
103
104 def visit(value: Any) -> None:
105 if value is None:
106 return
107 if isinstance(value, str):
108 if value.strip() and not _is_image_reference_string(value):
109 parts.append(value.strip())
110 return
111 if isinstance(value, list):
112 for item in value:
113 visit(item)
114 return
115 if isinstance(value, dict):
116 item_type = str(value.get("type", "")).strip()
117 if item_type in {"input_text", "output_text", "text"} and isinstance(
118 value.get("text"), str
119 ):
120 text = str(value.get("text", "")).strip()
121 if text:
122 parts.append(text)
123 return
124 if item_type == "tool_result":
125 visit(value.get("content"))
126 return
127 if isinstance(value.get("text"), str):
128 text = str(value.get("text", "")).strip()
129 if text:
130 parts.append(text)
131 return
132 if "content" in value:
133 visit(value.get("content"))
134
135 visit(node)
136 return "\n".join(parts).strip()
137
138
139def normalize_chatbox_history_content(content: str) -> str:
140 try:
141 parsed = json.loads(content)
142 except json.JSONDecodeError:
143 parsed = content
144 text = extract_text_content(parsed)
145 if text:
146 return text
147 if isinstance(parsed, str):
148 return parsed
149 return canonical_json(parsed)
150
151
152def extract_context_text(data: dict[str, Any], request_format: str) -> str:
153 input_payload = request_input_payload(data, request_format)
154 if isinstance(input_payload, str):
155 return "" if _is_image_reference_string(input_payload) else input_payload.strip()
156 chunks: list[str] = []
157
158 def visit(node: Any) -> None:
159 if node is None:
160 return
161 if isinstance(node, str):
162 if node.strip() and not _is_image_reference_string(node):
163 chunks.append(node.strip())
164 return
165 if isinstance(node, list):
166 for item in node:
167 visit(item)
168 return
169 if isinstance(node, dict):
170 if node.get("role") in {"user", "assistant", "system", "developer"}:
171 visit(node.get("content"))
172 return
173 if node.get("type") == "tool_result":
174 visit(node.get("content"))
175 return
176 if isinstance(node.get("text"), str):
177 chunks.append(str(node["text"]).strip())
178 return
179 if node.get("type") == "tool_use" and isinstance(node.get("input"), dict):
180 raw_input = canonical_json(node.get("input"))
181 if raw_input:
182 chunks.append(raw_input)
183 return
184 if isinstance(node.get("content"), (str, list, dict)):
185 visit(node.get("content"))
186 return
187 for value in node.values():
188 visit(value)
189
190 visit(input_payload)
191 if not chunks and isinstance(data.get("messages"), list):
192 visit(data.get("messages"))
193 return "\n".join(chunk for chunk in chunks if chunk).strip()
194
195
196def resolve_tool_name_for_call(
197 store: ConversationStore,
198 conversation_id: str | None,
199 owner: str,
200 call_id: str,
201) -> str:
202 if not conversation_id or not call_id:
203 return ""
204 try:
205 messages = store.get_messages(conversation_id, owner)
206 except ValueError:
207 return ""
208 for message in reversed(messages):
209 if str(message.metadata.get("tool_call_id", "")).strip() == call_id:
210 return str(message.metadata.get("tool_name", "")).strip()
211 return ""
212
213
214def extract_request_messages(
215 store: ConversationStore,
216 data: dict[str, Any],
217 *,
218 conversation_id: str | None,
219 owner: str,
220 request_format: str,
221) -> list[dict[str, Any]]:
222 payload = request_input_payload(data, request_format)
223 items = payload if isinstance(payload, list) else [payload]
224 extracted: list[dict[str, Any]] = []
225 for item in items:
226 if not isinstance(item, dict):
227 continue
228 role = str(item.get("role", "")).strip()
229 item_type = str(item.get("type", "")).strip()
230 if request_format == "anthropic_messages" and role == "user":
231 content_blocks = item.get("content")
232 if isinstance(content_blocks, list):
233 content = serialize_content(content_blocks)
234 if content:
235 extracted.append(
236 {
237 "role": "user",
238 "content": content,
239 "metadata": {
240 "turn": "user",
241 "status": "pending",
242 "source": request_format,
243 },
244 }
245 )
246 for content_block in content_blocks:
247 if not isinstance(content_block, dict):
248 continue
249 if str(content_block.get("type", "")).strip() != "tool_result":
250 continue
251 output = serialize_content(content_block.get("content"))
252 call_id = str(content_block.get("tool_use_id", "")).strip()
253 if output:
254 extracted.append(
255 {
256 "role": "tool",
257 "content": output,
258 "metadata": {
259 "source": request_format,
260 "response_mode": "tool_result",
261 "tool_call_id": call_id,
262 "tool_name": resolve_tool_name_for_call(
263 store,
264 conversation_id,
265 owner,
266 call_id,
267 ),
268 "output": output,
269 },
270 }
271 )
272 continue
273 if role == "user":
274 content = serialize_content(item.get("content"))
275 if content:
276 extracted.append(
277 {
278 "role": "user",
279 "content": content,
280 "metadata": {
281 "turn": "user",
282 "status": "pending",
283 "source": request_format,
284 },
285 }
286 )
287 continue
288 if role == "assistant":
289 content = _assistant_request_content(item)
290 if content:
291 metadata: dict[str, Any] = {
292 "source": request_format,
293 "turn": "assistant",
294 "history_imported": True,
295 }
296 tool_calls = item.get("tool_calls")
297 if isinstance(tool_calls, list) and tool_calls:
298 metadata["response_mode"] = "tool_call"
299 first_call = tool_calls[0] if isinstance(tool_calls[0], dict) else {}
300 function_payload = (
301 first_call.get("function")
302 if isinstance(first_call, dict) and isinstance(first_call.get("function"), dict)
303 else {}
304 )
305 metadata["tool_call_id"] = str(first_call.get("id", "")).strip()
306 metadata["tool_name"] = str(function_payload.get("name", "")).strip()
307 metadata["arguments"] = serialize_content(function_payload.get("arguments", ""))
308 extracted.append(
309 {
310 "role": "assistant",
311 "content": content,
312 "metadata": metadata,
313 }
314 )
315 continue
316 if request_format == "chat_completions" and role == "tool":
317 output = serialize_content(item.get("content"))
318 call_id = str(item.get("tool_call_id", "")).strip()
319 if output:
320 extracted.append(
321 {
322 "role": "tool",
323 "content": output,
324 "metadata": {
325 "source": request_format,
326 "response_mode": "tool_result",
327 "tool_call_id": call_id,
328 "tool_name": resolve_tool_name_for_call(
329 store,
330 conversation_id,
331 owner,
332 call_id,
333 ),
334 "output": output,
335 },
336 }
337 )
338 continue
339 if item_type == "function_call_output":
340 output = serialize_content(item.get("output", ""))
341 call_id = str(item.get("call_id", "")).strip()
342 if output:
343 extracted.append(
344 {
345 "role": "tool",
346 "content": output,
347 "metadata": {
348 "source": request_format,
349 "response_mode": "tool_result",
350 "tool_call_id": call_id,
351 "tool_name": resolve_tool_name_for_call(
352 store,
353 conversation_id,
354 owner,
355 call_id,
356 ),
357 "output": output,
358 },
359 }
360 )
361 return extracted
362
363
364def chatbox_comparable_messages_from_internal_messages(
365 messages: list[dict[str, Any]],
366) -> list[dict[str, str]]:
367 comparable: list[dict[str, str]] = []
368 for message in messages:
369 if not isinstance(message, dict):
370 continue
371 role = str(message.get("role", "")).strip()
372 if role not in {"user", "assistant", "tool"}:
373 continue
374 content = normalize_chatbox_history_content(str(message.get("content") or ""))
375 if content:
376 comparable.append(
377 {
378 "role": role,
379 "content": content,
380 }
381 )
382 return comparable
383
384
385def extract_chatbox_comparable_request_messages(
386 store: ConversationStore,
387 data: dict[str, Any],
388 *,
389 owner: str,
390 request_format: str,
391) -> list[dict[str, str]]:
392 return chatbox_comparable_messages_from_internal_messages(
393 extract_request_messages(
394 store,
395 data,
396 conversation_id=None,
397 owner=owner,
398 request_format=request_format,
399 )
400 )
401
402
403def resolve_conversation_by_history_strategies(
404 store: ConversationStore,
405 data: dict[str, Any],
406 owner: str,
407 request_format: str,
408):
409 chatbox_messages = extract_chatbox_comparable_request_messages(
410 store,
411 data,
412 owner=owner,
413 request_format=request_format,
414 )
415 return store.find_conversation_by_message_history(
416 owner,
417 chatbox_messages,
418 normalize_stored_content=normalize_chatbox_history_content,
419 )
420
421
422def extract_tool_result_call_ids(data: dict[str, Any], request_format: str) -> list[str]:
423 payload = request_input_payload(data, request_format)
424 items = payload if isinstance(payload, list) else [payload]
425 call_ids: list[str] = []
426 for item in items:
427 if not isinstance(item, dict):
428 continue
429 if request_format == "chat_completions":
430 if str(item.get("role", "")).strip() != "tool":
431 continue
432 call_id = str(item.get("tool_call_id", "")).strip()
433 if call_id:
434 call_ids.append(call_id)
435 continue
436 if request_format == "anthropic_messages":
437 content_blocks = item.get("content")
438 if not isinstance(content_blocks, list):
439 continue
440 for content_block in content_blocks:
441 if not isinstance(content_block, dict):
442 continue
443 if str(content_block.get("type", "")).strip() != "tool_result":
444 continue
445 call_id = str(content_block.get("tool_use_id", "")).strip()
446 if call_id:
447 call_ids.append(call_id)
448 continue
449 if str(item.get("type", "")).strip() != "function_call_output":
450 continue
451 call_id = str(item.get("call_id", "")).strip()
452 if call_id:
453 call_ids.append(call_id)
454 return call_ids
455
456
457def resolve_conversation_for_request(
458 store: ConversationStore,
459 data: dict[str, Any],
460 owner: str,
461 request_format: str,
462) -> tuple[ResolvedConversation | None, tuple[str, int] | None]:
463 explicit_conversation_id = str(data.get("conversation_id", "")).strip()
464 if explicit_conversation_id:
465 conversation = store.get_conversation(explicit_conversation_id, owner)
466 if conversation is None:
467 return None, ("conversation not found", 404)
468 locked_format = str(conversation.metadata.get("request_format", "")).strip().lower().replace("-", "_")
469 if locked_format and locked_format != request_format:
470 return None, ("conversation protocol is already locked", 409)
471 return ResolvedConversation(conversation=conversation, source="explicit_id"), None
472
473 for call_id in extract_tool_result_call_ids(data, request_format):
474 conversation = store.find_conversation_by_tool_call_id(owner, call_id)
475 if conversation is not None:
476 return ResolvedConversation(conversation=conversation, source="tool_call_id"), None
477
478 conversation = resolve_conversation_by_history_strategies(
479 store,
480 data,
481 owner,
482 request_format,
483 )
484 if conversation is not None:
485 return ResolvedConversation(conversation=conversation, source="history"), None
486
487 return None, None
488
489
490def build_message_debug_metadata(
491 *,
492 auth: AuthContext,
493 request_format: str,
494 request_data: dict[str, Any],
495 input_text: str,
496 input_payload: Any,
497 request_id: str,
498 resolved_model: str,
499 response_id: str | None = None,
500) -> dict[str, Any]:
501 tool_schemas = request_data.get("tools")
502 if request_format == "anthropic_messages" and isinstance(tool_schemas, list):
503 tool_schemas = [
504 {
505 "type": "function",
506 "function": {
507 "name": item.get("name"),
508 "description": item.get("description", ""),
509 "parameters": item.get("input_schema", {}),
510 },
511 }
512 if isinstance(item, dict)
513 else item
514 for item in tool_schemas
515 ]
516 return {
517 "provider": request_format,
518 "model": resolved_model,
519 "request_format": request_format,
520 "request_debug": {
521 "request_id": request_id,
522 "response_id": response_id or "",
523 "model": resolved_model,
524 "request_format": request_format,
525 "api_key_name": auth.request_api_key_name(),
526 "request_keys": sorted(request_data.keys()),
527 "input_text": input_text,
528 "input_payload": input_payload,
529 "tool_schemas": tool_schemas if isinstance(tool_schemas, list) else [],
530 "request_body": request_data,
531 "headers": auth.request_headers_snapshot(),
532 },
533 }
534