Coverage for open_webui/utils/misc.py: 22%
695 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1from __future__ import annotations
3import collections.abc
4import hashlib
5import ipaddress
6import logging
7import re
8import threading
9import time
10import uuid
11from datetime import timedelta
12from functools import lru_cache
13from pathlib import Path
14from typing import Callable, Optional, Sequence, Union
16import aiohttp
17import mimeparse
18from open_webui.env import CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE
19from open_webui.utils.json_codec import JSONCodec
21log = logging.getLogger(__name__)
22SURROGATE_RE = re.compile('[\ud800-\udfff]')
25def deep_update(d, u):
26 for k, v in u.items():
27 if isinstance(v, collections.abc.Mapping):
28 d[k] = deep_update(d.get(k, {}), v)
29 else:
30 d[k] = v
31 return d
34def merge_model_params(base: dict, override: dict) -> dict:
35 params = {**base, **override}
36 base_custom = base.get('custom_params')
37 override_custom = override.get('custom_params')
38 if isinstance(base_custom, dict) and (override_custom is None or isinstance(override_custom, dict)):
39 params['custom_params'] = {**base_custom, **(override_custom or {})}
40 return params
43def get_response_error_detail(response: object) -> str:
44 status_code = getattr(response, 'status_code', None)
45 fallback = f'Provider returned HTTP {status_code}' if status_code else 'Provider returned an error'
47 try:
48 body = response.body
49 if not isinstance(body, str):
50 body = body.decode('utf-8', 'replace')
51 detail = JSONCodec.loads(body)
52 except Exception:
53 return fallback
55 while isinstance(detail, dict):
56 next_detail = None
57 for key in ('error', 'message', 'detail'):
58 if key in detail:
59 next_detail = detail[key]
60 break
61 if next_detail is None:
62 return str(detail)
63 detail = next_detail
65 return detail if isinstance(detail, str) else str(detail)
68def _strip_filter_entry(entry):
69 # Compose list-form env syntax passes surrounding quotes through verbatim
70 return (entry or '').strip().strip('"\'').strip()
73def get_allow_block_lists(filter_list):
74 allow_list = []
75 block_list = []
77 for raw_entry in filter_list or []:
78 entry = _strip_filter_entry(raw_entry)
79 is_blocked = entry.startswith('!')
80 if is_blocked:
81 entry = _strip_filter_entry(entry[1:])
82 if not entry:
83 continue
84 if is_blocked:
85 block_list.append(entry)
86 else:
87 allow_list.append(entry)
89 return allow_list, block_list
92def is_string_allowed(string: Union[str, Sequence[str]], filter_list: list[str] | None = None) -> bool:
93 """
94 Checks if a string is allowed based on the provided filter list.
95 :param string: The string or sequence of strings to check (e.g., domain or hostname).
96 :param filter_list: List of allowed/blocked strings. Strings starting with "!" are blocked.
97 :return: True if the string or sequence of strings is allowed, False otherwise.
98 """
99 if not filter_list:
100 return True
102 allow_list, block_list = get_allow_block_lists(filter_list)
103 strings = [string] if isinstance(string, str) else list(string)
105 # If allow list is non-empty, require domain to match one of them
106 if allow_list:
107 if not any(s.endswith(allowed) for s in strings for allowed in allow_list):
108 return False
110 # Block list always removes matches
111 if any(s.endswith(blocked) for s in strings for blocked in block_list):
112 return False
114 return True
117@lru_cache(maxsize=512)
118def as_network(pattern: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network | None:
119 """A filter entry read as an address range, or None when the entry names a host instead.
121 Surrounding whitespace and a trailing dot are stripped here rather than by each caller,
122 since ip_network rejects both and the callers do not normalise the same way.
123 """
124 try:
125 return ipaddress.ip_network((pattern or '').strip().lower().rstrip('.'), strict=False)
126 except ValueError:
127 return None
130def _host_matches_pattern(host: str, pattern: str) -> bool:
131 """Match a hostname against a filter entry on DNS label boundaries.
133 `pattern` matches `host` when equal or a parent domain of it, so `corp.com`
134 matches `api.corp.com` but not `evilcorp.com`. Avoids the raw-suffix confusion
135 of a plain endswith.
137 An entry that names an address or a CIDR range is matched by containment instead, so
138 `10.0.0.0/8` covers `10.1.2.3` and an address matches any spelling of itself in its own family.
139 """
140 host = (host or '').strip().lower().rstrip('.')
141 pattern = (pattern or '').strip().lower().rstrip('.')
142 if not host or not pattern:
143 return False
144 network = as_network(pattern)
145 if network is not None:
146 try:
147 return ipaddress.ip_address(host) in network
148 except ValueError:
149 return False # a hostname is never inside an address range
150 return host == pattern or host.endswith('.' + pattern)
153def is_host_allowed(host: Union[str, Sequence[str]], filter_list: list[str] | None = None) -> bool:
154 """Allow/block a hostname (or list of hostnames / resolved IPs) against a
155 WEB_FETCH_FILTER_LIST-style filter, matching on label boundaries.
157 Pass a parsed hostname, never a full URL: matching against a URL lets a path
158 component defeat the filter (e.g. ``https://blocked.example/x`` ends with ``/x``,
159 not the blocked host). Entries prefixed with ``!`` are blocked; the rest form an allowlist.
160 An entry naming an address or a CIDR range is matched by containment instead.
161 """
162 if not filter_list:
163 return True
165 allow_list, _ = get_allow_block_lists(filter_list)
166 hosts = [host] if isinstance(host, str) else list(host or [])
168 if allow_list:
169 if not any(_host_matches_pattern(h, allowed) for h in hosts for allowed in allow_list):
170 return False
172 return not is_host_blocked(hosts, filter_list)
175def is_host_blocked(host: Union[str, Sequence[str]], filter_list: list[str] | None = None) -> bool:
176 """Whether a host or resolved address matches a block entry, ignoring any allow entries."""
177 _, block_list = get_allow_block_lists(filter_list)
178 hosts = [host] if isinstance(host, str) else list(host or [])
179 return any(_host_matches_pattern(h, blocked) for h in hosts for blocked in block_list)
182def get_message_list(messages_map, message_id):
183 """
184 Reconstructs a list of messages in order up to the specified message_id.
186 :param message_id: ID of the message to reconstruct the chain
187 :param messages: Message history dict containing all messages
188 :return: List of ordered messages starting from the root to the given message
189 """
191 # Handle case where messages is None
192 if not messages_map:
193 return [] # Return empty list instead of None to prevent iteration errors
195 # Find the message by its id
196 current_message = messages_map.get(message_id)
198 if not current_message: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 return [] # Return empty list instead of None to prevent iteration errors
201 # Reconstruct the chain by following the parentId links
202 message_list = []
203 visited_message_ids = set()
205 # Track the map keys, not the messages' own 'id' field: a message may omit it
206 while current_message and message_id not in visited_message_ids:
207 visited_message_ids.add(message_id)
208 message_list.append(current_message)
210 message_id = current_message.get('parentId')
211 current_message = messages_map.get(message_id) if message_id else None
213 message_list.reverse()
214 return message_list
217def get_messages_content(messages: list[dict]) -> str:
218 return '\n'.join([f'{message["role"].upper()}: {get_content_from_message(message)}' for message in messages])
221def get_last_user_message_item(messages: list[dict]) -> dict | None:
222 for message in reversed(messages):
223 if message['role'] == 'user':
224 return message
225 return None
228def get_content_from_message(message: dict) -> str | None:
229 content = message.get('content')
230 if isinstance(content, list):
231 for item in content:
232 if isinstance(item, dict) and item.get('type') == 'text':
233 return item.get('text')
234 elif content:
235 return content
237 output_text = get_output_text(message.get('output'))
238 return output_text or (content if isinstance(content, str) else None)
241def get_output_text(output: list | None) -> str:
242 if not isinstance(output, list): 242 ↛ 245line 242 didn't jump to line 245 because the condition on line 242 was always true
243 return ''
245 texts = []
246 for item in output:
247 if not isinstance(item, dict) or item.get('type') != 'message':
248 continue
250 parts = item.get('content') or []
251 if not isinstance(parts, list):
252 continue
254 text = ''.join(
255 str(part.get('text')) for part in parts if isinstance(part, dict) and part.get('text') is not None
256 )
257 # isspace() avoids the full-string copy strip() would make
258 if text and not text.isspace():
259 texts.append(text)
261 return '\n'.join(texts)
264def reconcile_tool_pairs(messages: list[dict]) -> list[dict]:
265 """Drop unpaired tool_use / tool_result from a reconstructed conversation.
267 Stored output can be incomplete — a tool result may be missing (e.g. the
268 knowledge base was updated mid-chat, or the call was interrupted), or a
269 tool call may be missing while its result survived. Strict providers
270 (Anthropic, AWS Bedrock Converse) reject either direction of mismatch.
272 Well-formed output is unaffected: every id pairs, so nothing is stripped.
273 """
274 completed_tool_call_ids = {
275 message['tool_call_id'] for message in messages if message.get('role') == 'tool' and message.get('tool_call_id')
276 }
277 requested_tool_call_ids = {
278 tool_call['id']
279 for message in messages
280 for tool_call in message.get('tool_calls') or ()
281 if message.get('role') == 'assistant' and tool_call.get('id')
282 }
284 reconciled_messages = []
285 for message in messages:
286 role = message.get('role')
288 # Orphan tool result — no assistant ever claimed this call_id.
289 if role == 'tool' and message.get('tool_call_id') not in requested_tool_call_ids:
290 continue
292 # Non-assistant or no tool_calls — pass through unchanged.
293 if role != 'assistant' or not message.get('tool_calls'):
294 reconciled_messages.append(message)
295 continue
297 # Keep only tool_calls whose id received a tool-role response.
298 valid_tool_calls = [
299 tool_call for tool_call in message['tool_calls'] if tool_call.get('id') in completed_tool_call_ids
300 ]
302 if valid_tool_calls:
303 reconciled_messages.append({**message, 'tool_calls': valid_tool_calls})
304 continue
306 # All tool_calls were orphans — keep the message only if it
307 # carries meaningful text or reasoning content.
308 content = get_content_from_message(message) or ''
309 has_meaningful_content = content.strip() if isinstance(content, str) else content
310 if has_meaningful_content or message.get('reasoning_content'):
311 reconciled_messages.append({key: value for key, value in message.items() if key != 'tool_calls'})
313 return reconciled_messages
316def get_reasoning_details(payload: dict):
317 if not isinstance(payload, dict):
318 return None
320 provider_fields = payload.get('provider_specific_fields') or {}
321 provider_details = provider_fields.get('reasoning_details') if isinstance(provider_fields, dict) else None
322 return payload.get('reasoning_details') or provider_details
325def convert_output_to_messages(
326 output: list,
327 raw: bool = False,
328 reasoning_format: str | None = None,
329 flatten_tool_images: bool = False,
330) -> list[dict]:
331 """
332 Convert OR-aligned output items to OpenAI Chat Completion-format messages.
334 This reconstructs the full conversation from the stored Responses API-native
335 output items, including assistant messages with tool_calls arrays and tool
336 role messages.
338 Args:
339 output: List of OR-aligned output items (Responses API format).
340 raw: If True, include code interpreter blocks for LLM re-processing
341 follow-ups.
342 reasoning_format: How to include reasoning blocks in the output:
343 - None: skip reasoning (default, safe for strict providers).
344 - ``'thinking'``: set as ``thinking`` top-level field
345 (for native Ollama).
346 - ``'think_tags'``: wrap in ``<think>`` tags inside content
347 (for legacy providers that expect reasoning as tagged content).
348 - ``'reasoning_content'``: set as ``reasoning_content`` top-level field
349 (for llama.cpp, which routes it via the chat template).
350 flatten_tool_images: Move tool output images into a following user
351 message for Chat Completions providers.
352 """
353 if not output or not isinstance(output, list):
354 return []
356 messages = []
357 pending_tool_calls = []
358 pending_content = []
359 pending_reasoning = [] # Only populated for top-level structured reasoning fields.
360 pending_reasoning_details = []
361 pending_tool_image_urls = []
362 pending_tool_outputs = []
363 completed_call_ids = {
364 item.get('call_id')
365 for item in output
366 if item.get('type') == 'function_call'
367 and item.get('call_id')
368 and item.get('status') in {'completed', 'failed', 'rejected'}
369 }
370 result_call_ids = {
371 item.get('call_id') for item in output if item.get('type') == 'function_call_output' and item.get('call_id')
372 }
373 function_call_ids = completed_call_ids & result_call_ids
375 def flush_pending():
376 nonlocal pending_content, pending_tool_calls, pending_reasoning, pending_reasoning_details
377 if not pending_content and not pending_tool_calls and not pending_reasoning and not pending_reasoning_details:
378 return
380 message = {
381 'role': 'assistant',
382 'content': '\n'.join(pending_content) if pending_content else '',
383 **({'tool_calls': pending_tool_calls} if pending_tool_calls else {}),
384 }
386 if pending_reasoning:
387 if reasoning_format == 'thinking':
388 message['thinking'] = '\n'.join(pending_reasoning)
389 else:
390 message['reasoning_content'] = '\n'.join(pending_reasoning)
392 if pending_reasoning_details:
393 message['reasoning_details'] = pending_reasoning_details
395 messages.append(message)
396 pending_content = []
397 pending_tool_calls = []
398 pending_reasoning = []
399 pending_reasoning_details = []
401 def flush_tool_images():
402 nonlocal pending_tool_image_urls
403 if not pending_tool_image_urls:
404 return
406 messages.append(
407 {
408 'role': 'user',
409 'content': [
410 {
411 'type': 'text',
412 'text': 'Here are the images from the tool results above. Please analyze them.',
413 },
414 *[{'type': 'image_url', 'image_url': {'url': url}} for url in pending_tool_image_urls],
415 ],
416 }
417 )
418 pending_tool_image_urls = []
420 def flush_tool_outputs():
421 nonlocal pending_tool_outputs
422 if not pending_tool_outputs:
423 return
425 flush_pending()
426 for output_item in pending_tool_outputs:
427 output_parts = output_item.get('output', [])
428 content = ''
429 image_urls = []
430 for part in output_parts:
431 if part.get('type') == 'input_text':
432 output_text = part.get('text', '')
433 content += str(output_text) if not isinstance(output_text, str) else output_text
434 elif part.get('type') == 'input_image':
435 url = part.get('image_url', '')
436 if url:
437 image_urls.append(url)
439 if flatten_tool_images:
440 messages.append(
441 {
442 'role': 'tool',
443 'tool_call_id': output_item.get('call_id', ''),
444 'content': content,
445 }
446 )
447 pending_tool_image_urls.extend(image_urls)
448 elif image_urls:
449 messages.append(
450 {
451 'role': 'tool',
452 'tool_call_id': output_item.get('call_id', ''),
453 'content': [
454 {'type': 'input_text', 'text': content},
455 *[{'type': 'input_image', 'image_url': url} for url in image_urls],
456 ],
457 }
458 )
459 else:
460 messages.append(
461 {
462 'role': 'tool',
463 'tool_call_id': output_item.get('call_id', ''),
464 'content': content,
465 }
466 )
468 pending_tool_outputs = []
470 for item in output:
471 item_type = item.get('type', '')
472 if item_type not in {'function_call', 'function_call_output'}:
473 flush_tool_outputs()
474 flush_tool_images()
476 if item_type == 'message':
477 # Extract text from output_text content parts
478 content_parts = item.get('content', [])
479 text = ''
480 for part in content_parts:
481 if part.get('type') == 'output_text':
482 text += part.get('text', '')
483 if text:
484 pending_content.append(text)
486 elif item_type == 'function_call':
487 if item.get('call_id') not in function_call_ids:
488 continue
490 # Collect tool calls to batch into assistant message
491 arguments = item.get('arguments', '{}')
492 # Ensure arguments is always a JSON string
493 if not isinstance(arguments, str):
494 arguments = JSONCodec.dumps(arguments)
495 pending_tool_calls.append(
496 {
497 'id': item.get('call_id', ''),
498 'type': 'function',
499 'function': {
500 'name': item.get('name', ''),
501 'arguments': arguments,
502 },
503 }
504 )
506 elif item_type == 'function_call_output':
507 if item.get('call_id') not in function_call_ids:
508 continue
510 pending_tool_outputs.append(item)
512 elif item_type == 'reasoning':
513 reasoning_details = item.get('reasoning_details') if raw else None
514 if reasoning_details:
515 reasoning_details = reasoning_details if isinstance(reasoning_details, list) else [reasoning_details]
516 reasoning_details = [
517 detail
518 for detail in reasoning_details
519 if isinstance(detail, dict)
520 and (detail.get('format') != 'anthropic-claude-v1' or detail.get('signature'))
521 ]
522 if not reasoning_format and not reasoning_details:
523 continue
525 reasoning_text = ''
526 source_list = item.get('summary', []) or item.get('content', [])
527 for part in source_list:
528 if part.get('type') == 'output_text':
529 reasoning_text += part.get('text', '')
530 elif 'text' in part:
531 reasoning_text += part.get('text', '')
533 if reasoning_text:
534 if reasoning_format == 'think_tags':
535 # Legacy tag replay: embed in content with the item's original tags.
536 start_tag = item.get('start_tag', '<think>')
537 end_tag = item.get('end_tag', '</think>')
538 pending_content.append(f'{start_tag}{reasoning_text}{end_tag}')
539 elif reasoning_format in {'thinking', 'reasoning_content'}:
540 # Native providers: collect for their top-level reasoning field.
541 pending_reasoning.append(reasoning_text)
543 if reasoning_details:
544 pending_reasoning_details.extend(reasoning_details)
546 elif item_type == 'open_webui:code_interpreter':
547 # Always include code interpreter content so the LLM knows
548 # the code was already executed and doesn't retry.
549 code = item.get('code', '')
550 code_output = item.get('output', '')
552 if code:
553 pending_content.append(f'<code_interpreter>\n{code}\n</code_interpreter>')
555 if code_output:
556 if isinstance(code_output, dict):
557 stdout = code_output.get('stdout', '')
558 result = code_output.get('result', '')
559 output_text = stdout or result
560 else:
561 output_text = str(code_output)
562 if output_text:
563 pending_content.append(f'<code_interpreter_output>\n{output_text}\n</code_interpreter_output>')
565 elif item_type.startswith('open_webui:'):
566 # Skip other extension types
567 pass
569 # Flush remaining content/tool_calls
570 flush_tool_outputs()
571 flush_tool_images()
572 flush_pending()
574 return reconcile_tool_pairs(messages)
577def get_last_user_message(messages: list[dict]) -> str | None:
578 message = get_last_user_message_item(messages)
579 if message is None:
580 return None
581 return get_content_from_message(message)
584def set_last_user_message_content(content: str, messages: list[dict]) -> list[dict]:
585 """
586 Replace the text content of the last user message in-place.
587 Handles both plain-string and list-of-parts content formats.
588 """
589 for message in reversed(messages):
590 if message.get('role') == 'user':
591 if isinstance(message.get('content'), list):
592 for item in message['content']:
593 if item.get('type') == 'text':
594 item['text'] = content
595 break
596 else:
597 message['content'] = content
598 break
599 return messages
602def get_last_assistant_message_item(messages: list[dict]) -> dict | None:
603 for message in reversed(messages):
604 if message['role'] == 'assistant':
605 return message
606 return None
609def get_last_assistant_message(messages: list[dict]) -> str | None:
610 for message in reversed(messages):
611 if message['role'] == 'assistant':
612 return get_content_from_message(message)
613 return None
616def get_system_message(messages: list[dict]) -> dict | None:
617 for message in messages:
618 if message['role'] == 'system':
619 return message
620 return None
623def remove_system_message(messages: list[dict]) -> list[dict]:
624 return [message for message in messages if message['role'] != 'system']
627def pop_system_message(messages: list[dict]) -> tuple[dict | None, list[dict]]:
628 return get_system_message(messages), remove_system_message(messages)
631def merge_system_messages(messages: list[dict]) -> list[dict]:
632 """
633 Merge all system messages into one at position 0.
635 Some chat templates (e.g. Qwen) require exactly one system
636 message at the start. Multiple pipeline stages may each
637 insert their own system message; this function consolidates
638 them.
639 """
640 system_contents: list[str] = []
641 other_messages: list[dict] = []
643 for message in messages:
644 if message.get('role') == 'system':
645 content = get_content_from_message(message)
646 if content:
647 system_contents.append(content)
648 else:
649 other_messages.append(message)
651 if not system_contents:
652 return other_messages
654 merged = {'role': 'system', 'content': '\n'.join(system_contents)}
655 return [merged, *other_messages]
658def update_message_content(message: dict, content: str, append: bool = True) -> dict:
659 if isinstance(message['content'], list):
660 for item in message['content']:
661 if item['type'] == 'text':
662 if append:
663 item['text'] = f'{item["text"]}\n{content}'
664 else:
665 item['text'] = f'{content}\n{item["text"]}'
666 else:
667 if append:
668 message['content'] = f'{message["content"]}\n{content}'
669 else:
670 message['content'] = f'{content}\n{message["content"]}'
671 return message
674def replace_system_message_content(content: str, messages: list[dict]) -> dict:
675 for message in messages:
676 if message['role'] == 'system':
677 message['content'] = content
678 break
679 return messages
682def add_or_update_system_message(content: str, messages: list[dict], append: bool = False):
683 """
684 Adds a new system message at the beginning of the messages list
685 or updates the existing system message at the beginning.
687 :param msg: The message to be added or appended.
688 :param messages: The list of message dictionaries.
689 :return: The updated list of message dictionaries.
690 """
692 if messages and messages[0].get('role') == 'system':
693 messages[0] = update_message_content(messages[0], content, append)
694 else:
695 # Insert at the beginning
696 messages.insert(0, {'role': 'system', 'content': content})
698 return messages
701def add_or_update_user_message(content: str, messages: list[dict], append: bool = True):
702 """
703 Adds a new user message at the end of the messages list
704 or updates the existing user message at the end.
706 :param msg: The message to be added or appended.
707 :param messages: The list of message dictionaries.
708 :return: The updated list of message dictionaries.
709 """
711 if messages and messages[-1].get('role') == 'user':
712 messages[-1] = update_message_content(messages[-1], content, append)
713 else:
714 # Insert at the end
715 messages.append({'role': 'user', 'content': content})
717 return messages
720def prepend_to_first_user_message_content(content: str, messages: list[dict]) -> list[dict]:
721 for message in messages:
722 if message['role'] == 'user':
723 message = update_message_content(message, content, append=False)
724 break
725 return messages
728def append_or_update_assistant_message(content: str, messages: list[dict]):
729 """
730 Adds a new assistant message at the end of the messages list
731 or updates the existing assistant message at the end.
733 :param msg: The message to be added or appended.
734 :param messages: The list of message dictionaries.
735 :return: The updated list of message dictionaries.
736 """
738 if messages and messages[-1].get('role') == 'assistant':
739 messages[-1]['content'] = f'{messages[-1]["content"]}\n{content}'
740 else:
741 # Insert at the end
742 messages.append({'role': 'assistant', 'content': content})
744 return messages
747def strip_empty_content_blocks(messages: list[dict]) -> list[dict]:
748 """
749 Remove empty text content blocks from multimodal message content arrays.
751 Providers like Gemini and Claude reject messages where a text block has
752 an empty string. This can happen when a user sends only file/image
753 attachments without typing any text.
754 """
755 for message in messages:
756 content = message.get('content')
757 if isinstance(content, list):
758 cleaned = [
759 block
760 for block in content
761 if not (isinstance(block, dict) and block.get('type') == 'text' and not block.get('text', '').strip())
762 ]
763 if cleaned:
764 message['content'] = cleaned
765 return messages
768def openai_chat_message_template(model: str, message_id: str | None = None):
769 return {
770 'id': message_id if message_id else f'{model}-{str(uuid.uuid4())}',
771 'created': int(time.time()),
772 'model': model,
773 'choices': [{'index': 0, 'logprobs': None, 'finish_reason': None}],
774 }
777def openai_chat_chunk_message_template(
778 model: str,
779 content: str | None = None,
780 reasoning_content: str | None = None,
781 tool_calls: list[dict] | None = None,
782 usage: dict | None = None,
783 message_id: str | None = None,
784) -> dict:
785 template = openai_chat_message_template(model, message_id)
786 template['object'] = 'chat.completion.chunk'
788 template['choices'][0]['index'] = 0
789 template['choices'][0]['delta'] = {}
791 if content:
792 template['choices'][0]['delta']['content'] = content
794 if reasoning_content:
795 template['choices'][0]['delta']['reasoning_content'] = reasoning_content
797 if tool_calls:
798 template['choices'][0]['delta']['tool_calls'] = tool_calls
800 if not content and not reasoning_content and not tool_calls:
801 template['choices'][0]['finish_reason'] = 'stop'
803 if usage:
804 template['usage'] = usage
805 return template
808def openai_chat_completion_message_template(
809 model: str,
810 message: str | None = None,
811 reasoning_content: str | None = None,
812 tool_calls: list[dict] | None = None,
813 usage: dict | None = None,
814) -> dict:
815 template = openai_chat_message_template(model)
816 template['object'] = 'chat.completion'
817 if message is not None:
818 template['choices'][0]['message'] = {
819 'role': 'assistant',
820 'content': message,
821 **({'reasoning_content': reasoning_content} if reasoning_content else {}),
822 **({'tool_calls': tool_calls} if tool_calls else {}),
823 }
825 template['choices'][0]['finish_reason'] = 'tool_calls' if tool_calls else 'stop'
827 if usage:
828 template['usage'] = usage
829 return template
832def get_gravatar_url(email):
833 # Trim leading and trailing whitespace from
834 # an email address and force all characters
835 # to lower case
836 address = str(email).strip().lower()
838 # Create a SHA256 hash of the final string
839 hash_object = hashlib.sha256(address.encode())
840 hash_hex = hash_object.hexdigest()
842 # Grab the actual image URL
843 return f'https://www.gravatar.com/avatar/{hash_hex}?d=mp'
846# Give us each day the data we require, and forgive us our
847# technical debts as we forgive those who commit upstream.
848# Lead the bits not into corruption but deliver them from
849# entropy, for the checksum and the glory are forever.
850def calculate_sha256(file_path, chunk_size):
851 # Compute SHA-256 hash of a file efficiently in chunks
852 sha256 = hashlib.sha256()
853 with open(file_path, 'rb') as f:
854 while chunk := f.read(chunk_size):
855 sha256.update(chunk)
856 return sha256.hexdigest()
859def calculate_sha256_string(string):
860 # Create a new SHA-256 hash object
861 sha256_hash = hashlib.sha256()
862 # Update the hash object with the bytes of the input string
863 sha256_hash.update(string.encode('utf-8'))
864 # Get the hexadecimal representation of the hash
865 hashed_string = sha256_hash.hexdigest()
866 return hashed_string
869def validate_email_format(email: str) -> bool:
870 if email.endswith('@localhost'): 870 ↛ 871line 870 didn't jump to line 871 because the condition on line 870 was never true
871 return True
873 return bool(re.match(r'[^@]+@[^@]+\.[^@]+', email))
876def sanitize_filename(file_name):
877 # Convert to lowercase
878 lower_case_file_name = file_name.lower()
880 # Remove special characters using regular expression
881 sanitized_file_name = re.sub(r'[^\w\s]', '', lower_case_file_name)
883 # Replace spaces with dashes
884 final_file_name = re.sub(r'\s+', '-', sanitized_file_name)
886 return final_file_name
889def json_text_variants(value: str) -> list[str]:
890 """Both spellings ``value`` can take inside a serialized JSON column, unquoted.
892 Encoders disagree on non-ASCII — stdlib escapes it to ``\\uXXXX``, orjson writes it
893 raw — so a LIKE against the stored text has to accept either. ASCII collapses to one.
894 """
895 raw = JSONCodec.dumps(value, ensure_ascii=False)[1:-1]
896 escaped = JSONCodec.dumps(value, ensure_ascii=True)[1:-1]
897 return [raw] if raw == escaped else [raw, escaped]
900def sanitize_text_for_db(text: str) -> str:
901 """Remove null bytes and invalid UTF-8 surrogates from text for PostgreSQL storage."""
902 if not isinstance(text, str): 902 ↛ 903line 902 didn't jump to line 903 because the condition on line 902 was never true
903 return text
904 # Fast path: skip work when there are no null bytes or surrogate code points.
905 if '\x00' not in text and not SURROGATE_RE.search(text):
906 return text
907 return SURROGATE_RE.sub('', text.replace('\x00', ''))
910def _strip_null_bytes_deep(obj):
911 """Inner recursive walk — only called when null bytes are known to be present."""
912 if isinstance(obj, str):
913 return sanitize_text_for_db(obj)
914 elif isinstance(obj, dict):
915 cleaned = {}
916 for k, v in obj.items():
917 cleaned[sanitize_text_for_db(k) if isinstance(k, str) else k] = _strip_null_bytes_deep(v)
918 return cleaned
919 elif isinstance(obj, list):
920 return [_strip_null_bytes_deep(v) for v in obj]
921 return obj
924def sanitize_data_for_db(obj):
925 """Recursively sanitize all strings in a data structure for database storage.
927 Performs a fast pre-check: serializes the structure once and scans for
928 null bytes or invalid UTF-8 surrogates. If none are found, the
929 original object is returned immediately, skipping the expensive
930 recursive walk.
931 """
932 if isinstance(obj, str):
933 return sanitize_text_for_db(obj)
934 # Fast path: check for null bytes and surrogate code points in the serialized form.
935 # json.dumps is implemented in C and much faster than a Python-level
936 # recursive walk over every leaf string.
937 try:
938 serialized = JSONCodec.dumps(obj, ensure_ascii=False)
939 if '\\u0000' not in serialized:
940 serialized.encode('utf-8')
941 return obj
942 except (TypeError, ValueError, UnicodeEncodeError):
943 pass
944 return _strip_null_bytes_deep(obj)
947def sanitize_metadata(metadata: dict) -> dict:
948 """
949 Return a JSON-safe copy of a metadata dict for database storage.
951 The middleware metadata accumulates non-serializable Python objects
952 (e.g. callable tool functions, MCP client instances) that cause
953 PostgreSQL JSON inserts to fail. This helper strips those out while
954 preserving the primitive data needed for file-to-chat linking.
955 """
956 if not isinstance(metadata, dict): 956 ↛ 957line 956 didn't jump to line 957 because the condition on line 956 was never true
957 return metadata
959 def _sanitize(obj):
960 if isinstance(obj, (str, int, float, bool, type(None))):
961 return obj
962 if isinstance(obj, dict):
963 return {k: _sanitize(v) for k, v in obj.items() if not callable(v) and _is_serializable(v)}
964 if isinstance(obj, list): 964 ↛ 966line 964 didn't jump to line 966 because the condition on line 964 was always true
965 return [_sanitize(v) for v in obj if not callable(v) and _is_serializable(v)]
966 if callable(obj):
967 return None
968 # Last resort: try to see if it's serializable
969 try:
970 JSONCodec.dumps(obj)
971 return obj
972 except (TypeError, ValueError):
973 return None
975 def _is_serializable(obj):
976 """Quick check whether a value can survive JSON serialization."""
977 if isinstance(obj, (str, int, float, bool, type(None), dict, list)): 977 ↛ 979line 977 didn't jump to line 979 because the condition on line 977 was always true
978 return True
979 try:
980 JSONCodec.dumps(obj)
981 return True
982 except (TypeError, ValueError):
983 return False
985 return _sanitize(metadata)
988def extract_folders_after_data_docs(path):
989 # Convert the path to a Path object if it's not already
990 path = Path(path)
992 # Extract parts of the path
993 parts = path.parts
995 # Find the index of '/data/docs' in the path
996 try:
997 index_data_docs = parts.index('data') + 1
998 index_docs = parts.index('docs', index_data_docs) + 1
999 except ValueError:
1000 return []
1002 # Exclude the filename and accumulate folder names
1003 tags = []
1005 folders = parts[index_docs:-1]
1006 for idx, _ in enumerate(folders):
1007 tags.append('/'.join(folders[: idx + 1]))
1009 return tags
1012def parse_duration(duration: str) -> timedelta | None:
1013 if duration == '-1' or duration == '0': 1013 ↛ 1014line 1013 didn't jump to line 1014 because the condition on line 1013 was never true
1014 return None
1016 # Regular expression to find number and unit pairs
1017 pattern = r'(-?\d+(\.\d+)?)(ms|s|m|h|d|w)'
1018 matches = re.findall(pattern, duration)
1020 if not matches: 1020 ↛ 1021line 1020 didn't jump to line 1021 because the condition on line 1020 was never true
1021 raise ValueError('Invalid duration string')
1023 total_duration = timedelta()
1025 for number, _, unit in matches:
1026 number = float(number)
1027 if unit == 'ms': 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 total_duration += timedelta(milliseconds=number)
1029 elif unit == 's': 1029 ↛ 1030line 1029 didn't jump to line 1030 because the condition on line 1029 was never true
1030 total_duration += timedelta(seconds=number)
1031 elif unit == 'm': 1031 ↛ 1032line 1031 didn't jump to line 1032 because the condition on line 1031 was never true
1032 total_duration += timedelta(minutes=number)
1033 elif unit == 'h': 1033 ↛ 1034line 1033 didn't jump to line 1034 because the condition on line 1033 was never true
1034 total_duration += timedelta(hours=number)
1035 elif unit == 'd': 1035 ↛ 1036line 1035 didn't jump to line 1036 because the condition on line 1035 was never true
1036 total_duration += timedelta(days=number)
1037 elif unit == 'w': 1037 ↛ 1025line 1037 didn't jump to line 1025 because the condition on line 1037 was always true
1038 total_duration += timedelta(weeks=number)
1040 return total_duration
1043def parse_ollama_modelfile(model_text):
1044 parameters_meta = {
1045 'mirostat': int,
1046 'mirostat_eta': float,
1047 'mirostat_tau': float,
1048 'num_ctx': int,
1049 'repeat_last_n': int,
1050 'repeat_penalty': float,
1051 'temperature': float,
1052 'seed': int,
1053 'tfs_z': float,
1054 'num_predict': int,
1055 'top_k': int,
1056 'top_p': float,
1057 'num_keep': int,
1058 'presence_penalty': float,
1059 'frequency_penalty': float,
1060 'num_batch': int,
1061 'num_gpu': int,
1062 'use_mmap': bool,
1063 'use_mlock': bool,
1064 'num_thread': int,
1065 }
1067 data = {'base_model_id': None, 'params': {}}
1069 # Parse base model
1070 base_model_match = re.search(r'^FROM\s+(\w+)', model_text, re.MULTILINE | re.IGNORECASE)
1071 if base_model_match:
1072 data['base_model_id'] = base_model_match.group(1)
1074 # Parse template
1075 template_match = re.search(r'TEMPLATE\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE)
1076 if template_match:
1077 data['params'] = {'template': template_match.group(1).strip()}
1079 # Parse stops
1080 stops = re.findall(r'PARAMETER stop "(.*?)"', model_text, re.IGNORECASE)
1081 if stops:
1082 data['params']['stop'] = stops
1084 # Parse other parameters from the provided list
1085 for param, param_type in parameters_meta.items():
1086 param_match = re.search(rf'PARAMETER {param} (.+)', model_text, re.IGNORECASE)
1087 if param_match:
1088 value = param_match.group(1)
1090 try:
1091 if param_type is int:
1092 value = int(value)
1093 elif param_type is float:
1094 value = float(value)
1095 elif param_type is bool:
1096 value = value.lower() == 'true'
1097 except Exception as e:
1098 log.exception(f'Failed to parse parameter {param}: {e}')
1099 continue
1101 data['params'][param] = value
1103 # Parse adapter
1104 adapter_match = re.search(r'ADAPTER (.+)', model_text, re.IGNORECASE)
1105 if adapter_match:
1106 data['params']['adapter'] = adapter_match.group(1)
1108 # Parse system description
1109 system_desc_match = re.search(r'SYSTEM\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE)
1110 system_desc_match_single = re.search(r'SYSTEM\s+([^\n]+)', model_text, re.IGNORECASE)
1112 if system_desc_match:
1113 data['params']['system'] = system_desc_match.group(1).strip()
1114 elif system_desc_match_single:
1115 data['params']['system'] = system_desc_match_single.group(1).strip()
1117 # Parse messages
1118 messages = []
1119 message_matches = re.findall(r'MESSAGE (\w+) (.+)', model_text, re.IGNORECASE)
1120 for role, content in message_matches:
1121 messages.append({'role': role, 'content': content})
1123 if messages:
1124 data['params']['messages'] = messages
1126 return data
1129def convert_logit_bias_input_to_json(logit_bias_input) -> str | None:
1130 if not logit_bias_input:
1131 return None
1133 if isinstance(logit_bias_input, dict):
1134 return JSONCodec.dumps(logit_bias_input)
1136 logit_bias_pairs = logit_bias_input.split(',')
1137 logit_bias_json = {}
1138 for pair in logit_bias_pairs:
1139 token, bias = pair.split(':')
1140 token = str(token.strip())
1141 bias = int(bias.strip())
1142 bias = 100 if bias > 100 else -100 if bias < -100 else bias
1143 logit_bias_json[token] = bias
1144 return JSONCodec.dumps(logit_bias_json)
1147def freeze(value):
1148 """
1149 Freeze a value to make it hashable.
1150 """
1151 if isinstance(value, dict): 1151 ↛ 1153line 1151 didn't jump to line 1153 because the condition on line 1151 was always true
1152 return frozenset((k, freeze(v)) for k, v in value.items())
1153 elif isinstance(value, list):
1154 return tuple(freeze(v) for v in value)
1155 return value
1158def throttle(interval: float = 10.0):
1159 """
1160 Decorator to prevent a function from being called more than once within a specified duration.
1161 If the function is called again within the duration, it returns None. To avoid returning
1162 different types, the return type of the function should be T | None.
1164 :param interval: Duration in seconds to wait before allowing the function to be called again.
1165 Zero or negative disables throttling.
1166 """
1168 def decorator(func):
1169 if interval <= 0: 1169 ↛ 1170line 1169 didn't jump to line 1170 because the condition on line 1169 was never true
1170 return func
1172 last_calls = {}
1173 lock = threading.Lock()
1175 async def wrapper(*args, **kwargs):
1176 key = (args, freeze(kwargs))
1177 now = time.time()
1178 if now - last_calls.get(key, 0) < interval:
1179 return None
1180 with lock:
1181 if now - last_calls.get(key, 0) < interval: 1181 ↛ 1182line 1181 didn't jump to line 1182 because the condition on line 1181 was never true
1182 return None
1183 last_calls[key] = now
1184 return await func(*args, **kwargs)
1186 return wrapper
1188 return decorator
1191def strict_match_mime_type(supported: list[str] | str, header: str) -> str | None:
1192 """
1193 Strictly match the mime type with the supported mime types.
1195 :param supported: The supported mime types.
1196 :param header: The header to match.
1197 :return: The matched mime type or None if no match is found.
1198 """
1200 try:
1201 if isinstance(supported, str): 1201 ↛ 1202line 1201 didn't jump to line 1202 because the condition on line 1201 was never true
1202 supported = supported.split(',')
1204 supported = [s for s in supported if s.strip() and '/' in s]
1206 if len(supported) == 0:
1207 # Default to common types if none are specified
1208 supported = ['audio/*', 'video/webm']
1210 match = mimeparse.best_match(supported, header)
1211 if not match:
1212 return None
1214 _, _, match_params = mimeparse.parse_mime_type(match)
1215 _, _, header_params = mimeparse.parse_mime_type(header)
1216 for k, v in match_params.items():
1217 if header_params.get(k) != v:
1218 return None
1220 return match
1221 except Exception as e:
1222 log.exception(f'Failed to match mime type {header}: {e}')
1223 return None
1226def is_raster_image_content_type(content_type: str | None) -> bool:
1227 """Return True if the content type is an image that decodes as a bitmap; SVG is XML."""
1228 base_content_type = (content_type or '').split(';')[0].strip().lower()
1229 return base_content_type.startswith('image/') and base_content_type != 'image/svg+xml'
1232def extract_urls(text: str) -> list[str]:
1233 # Regex pattern to match URLs
1234 url_pattern = re.compile(r'(https?://[^\s]+)', re.IGNORECASE) # Matches http and https URLs
1235 return url_pattern.findall(text)
1238# We believe in one architect of all that is seen and served.
1239# Should this stream falter, it shall be raised again on the
1240# third retry. We look for the uptime of the world to come.
1241async def cleanup_response(
1242 response: aiohttp.ClientResponse | None,
1243 session: aiohttp.ClientSession | None,
1244):
1245 if response:
1246 if not response.closed:
1247 # aiohttp 3.9+ made ClientResponse.close() synchronous (returns None).
1248 # Older versions returned a coroutine. Handle both gracefully.
1249 result = response.close()
1250 if result is not None:
1251 await result
1252 if session:
1253 if not session.closed:
1254 result = session.close()
1255 if result is not None:
1256 await result
1259async def stream_wrapper(response, session, content_handler=None):
1260 """
1261 Wrap a stream to ensure cleanup happens even if streaming is interrupted.
1262 This is more reliable than BackgroundTask which may not run if client disconnects.
1263 """
1264 try:
1265 stream = content_handler(response.content) if content_handler else response.content
1266 async for chunk in stream:
1267 yield chunk
1268 finally:
1269 await cleanup_response(response, session)
1272def stream_chunks_handler(stream: aiohttp.StreamReader):
1273 """
1274 Assemble lines from raw chunks, so a line over aiohttp's reader limit no longer aborts the stream.
1275 When CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE is set, a line exceeding it is dropped.
1277 :param stream: The stream reader to handle.
1278 :return: An async generator that yields the stream one line at a time.
1279 """
1281 max_buffer_size = CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE
1282 if max_buffer_size is None or max_buffer_size <= 0:
1283 max_buffer_size = float('inf') # unset: no line is too long
1285 async def yield_safe_stream_chunks():
1286 buffer = bytearray() # bytearray, not bytes: `+=` on bytes reallocates, quadratic on long lines
1287 dropping_line_tail = False
1289 async for data, _ in stream.iter_chunks():
1290 if not data:
1291 continue
1293 buffer += data
1295 # Only split once a line completed: splitting every chunk re-copies the buffer, quadratic
1296 if b'\n' in data:
1297 *lines, rest = bytes(buffer).split(b'\n')
1298 buffer = bytearray(rest)
1300 for line in lines:
1301 if dropping_line_tail:
1302 dropping_line_tail = False
1303 elif len(line) > max_buffer_size:
1304 log.info('Dropped line over max buffer size: %s bytes', len(line))
1305 else:
1306 yield line + b'\n'
1308 # Oversized line still arriving: drop it instead of buffering the rest
1309 if len(buffer) > max_buffer_size:
1310 if not dropping_line_tail:
1311 log.info('Dropping line over max buffer size, buffered so far: %s bytes', len(buffer))
1312 dropping_line_tail = True
1313 buffer.clear()
1315 if buffer and not dropping_line_tail:
1316 yield bytes(buffer)
1318 return yield_safe_stream_chunks()