From ecec86dd32aa396dd5a383e9be5b0f16c4346c48 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Wed, 13 May 2026 15:55:21 +0900 Subject: [PATCH] refac --- backend/open_webui/tools/builtin.py | 233 +++++++++++++++++++++++ backend/open_webui/tools/knowledge_fs.py | 63 +++--- backend/open_webui/utils/tools.py | 6 +- 3 files changed, 276 insertions(+), 26 deletions(-) diff --git a/backend/open_webui/tools/builtin.py b/backend/open_webui/tools/builtin.py index a9e51a18fc..86ce210a5f 100644 --- a/backend/open_webui/tools/builtin.py +++ b/backend/open_webui/tools/builtin.py @@ -1733,12 +1733,175 @@ async def search_knowledge_files( # Hard cap for view_file / view_knowledge_file output MAX_VIEW_FILE_CHARS = 100_000 DEFAULT_VIEW_FILE_MAX_CHARS = 10_000 +MAX_GREP_RESULTS = 50 + + +async def grep_knowledge_files( + pattern: str, + file_id: Optional[str] = None, + case_insensitive: bool = False, + count_only: bool = False, + __request__: Request = None, + __user__: dict = None, + __model_knowledge__: Optional[list[dict]] = None, +) -> str: + """ + Search for exact text across knowledge files. Returns matching lines with line numbers. + Unlike query_knowledge_files (semantic/vector search), this performs exact string matching. + Automatically detects regex patterns (e.g. "error|warn", "version \\d+"). + + :param pattern: The text pattern to search for (regex auto-detected) + :param file_id: Optional file ID to search within a single file only + :param case_insensitive: If true, ignore case when matching (default: false) + :param count_only: If true, return only match counts per file (default: false) + :return: Matching lines with file IDs, filenames, and line numbers + """ + if __request__ is None: + return json.dumps({'error': 'Request context not available'}) + + if not __user__: + return json.dumps({'error': 'User context not available'}) + + if not pattern or not pattern.strip(): + return json.dumps({'error': 'Pattern is required'}) + + try: + from open_webui.models.files import Files + from open_webui.models.knowledge import Knowledges + from open_webui.tools.knowledge_fs import build_matcher + + user_id = __user__.get('id') + user_role = __user__.get('role', 'user') + user_group_ids = [group.id for group in await Groups.get_groups_by_member_id(user_id)] + + _matches, err = build_matcher(pattern, case_insensitive) + if err: + return json.dumps({'error': err}) + + # Collect files to search + files_to_search = [] + + if file_id: + # Single file mode — verify access + file = await Files.get_file_by_id(file_id) + if file: + from open_webui.utils.access_control.files import has_access_to_file + + if ( + file.user_id != user_id + and user_role != 'admin' + and not any( + item.get('type') == 'file' and item.get('id') == file_id + for item in (__model_knowledge__ or []) + ) + and not await has_access_to_file( + file_id=file_id, + access_type='read', + user=UserModel(**__user__), + ) + ): + return json.dumps({'error': 'File not found'}) + files_to_search.append(file) + elif __model_knowledge__: + # Scoped to model's attached knowledge + seen_ids = set() + for item in __model_knowledge__: + item_type = item.get('type') + item_id = item.get('id') + if item_type == 'file' and item_id not in seen_ids: + file = await Files.get_file_by_id(item_id) + if file: + files_to_search.append(file) + seen_ids.add(item_id) + elif item_type == 'collection': + knowledge = await Knowledges.get_knowledge_by_id(item_id) + if not knowledge: + continue + for fid in (knowledge.data or {}).get('file_ids', []): + if fid not in seen_ids: + file = await Files.get_file_by_id(fid) + if file: + files_to_search.append(file) + seen_ids.add(fid) + else: + # All accessible knowledge bases — use the same search pattern as list_knowledge_bases + result = await Knowledges.search_knowledge_bases( + user_id, + filter={ + 'query': '', + 'user_id': user_id, + 'group_ids': user_group_ids, + }, + skip=0, + limit=200, + ) + seen_ids = set() + for kb in result.items: + file_ids = [] + # Get files attached to this KB + files_from_kb = await Knowledges.get_files_by_id(kb.id) + if files_from_kb: + file_ids = [f.id for f in files_from_kb] + for fid in file_ids: + if fid not in seen_ids: + file = await Files.get_file_by_id(fid) + if file: + files_to_search.append(file) + seen_ids.add(fid) + + if not files_to_search: + return json.dumps({'error': 'No accessible files found'}) + + # Search + results = [] + total_matches = 0 + counts = [] + + for file in files_to_search: + content = '' + if file.data: + content = file.data.get('content', '') + if not content: + continue + + lines = content.split('\n') + file_matches = 0 + + for i, line in enumerate(lines, 1): + if _matches(line): + file_matches += 1 + total_matches += 1 + if not count_only and len(results) < MAX_GREP_RESULTS: + results.append(f'{file.id} {file.filename}:{i}: {line}') + + if file_matches > 0 and count_only: + counts.append(f'{file.id} {file.filename}: {file_matches}') + + if count_only: + if not counts: + return f'No matches for "{pattern}"' + return '\n'.join(counts) + f'\n[{total_matches} total matches]' + + if not results: + return f'No matches for "{pattern}"' + + output = '\n'.join(results) + if total_matches > MAX_GREP_RESULTS: + output += f'\n[{MAX_GREP_RESULTS} of {total_matches} matches shown — use file_id to narrow]' + return output + + except Exception as e: + log.exception(f'grep_knowledge_files error: {e}') + return json.dumps({'error': str(e)}) async def view_file( file_id: str, offset: int = 0, max_chars: int = DEFAULT_VIEW_FILE_MAX_CHARS, + line_numbers: bool = False, + start_line: Optional[int] = None, + end_line: Optional[int] = None, __request__: Request = None, __user__: dict = None, __model_knowledge__: Optional[list[dict]] = None, @@ -1749,6 +1912,9 @@ async def view_file( :param file_id: The ID of the file to retrieve :param offset: Character offset to start reading from (default: 0) :param max_chars: Maximum characters to return (default: 10000, hard cap: 100000) + :param line_numbers: If true, prefix each line with its 1-indexed line number + :param start_line: Optional 1-indexed start line (overrides offset/max_chars when set) + :param end_line: Optional 1-indexed end line (inclusive) :return: JSON with the file's id, filename, content, and pagination metadata if truncated """ if __request__ is None: @@ -1803,9 +1969,38 @@ async def view_file( content = file.data.get('content', '') total_chars = len(content) + + # Line-based addressing (overrides char-based offset/max_chars) + if start_line is not None: + all_lines = content.split('\n') + total_lines = len(all_lines) + s = max(1, int(start_line)) - 1 # 1-indexed to 0-indexed + e = min(total_lines, int(end_line) if end_line else s + 100) + selected = all_lines[s:e] + sliced = '\n'.join(f'{s + i + 1}: {line}' for i, line in enumerate(selected)) + is_truncated = e < total_lines + result = { + 'id': file.id, + 'filename': file.filename, + 'content': sliced, + 'updated_at': file.updated_at, + 'created_at': file.created_at, + 'total_lines': total_lines, + 'showing_lines': f'{s + 1}-{e}', + } + if is_truncated: + result['truncated'] = True + result['next_start_line'] = e + 1 + return json.dumps(result, ensure_ascii=False) + sliced = content[offset : offset + max_chars] is_truncated = (offset + len(sliced)) < total_chars + if line_numbers: + start_ln = content[:offset].count('\n') + 1 + lines = sliced.split('\n') + sliced = '\n'.join(f'{start_ln + i}: {line}' for i, line in enumerate(lines)) + result = { 'id': file.id, 'filename': file.filename, @@ -1832,6 +2027,9 @@ async def view_knowledge_file( file_id: str, offset: int = 0, max_chars: int = DEFAULT_VIEW_FILE_MAX_CHARS, + line_numbers: bool = False, + start_line: Optional[int] = None, + end_line: Optional[int] = None, __request__: Request = None, __user__: dict = None, ) -> str: @@ -1841,6 +2039,9 @@ async def view_knowledge_file( :param file_id: The ID of the file to retrieve :param offset: Character offset to start reading from (default: 0) :param max_chars: Maximum characters to return (default: 10000, hard cap: 100000) + :param line_numbers: If true, prefix each line with its 1-indexed line number + :param start_line: Optional 1-indexed start line (overrides offset/max_chars when set) + :param end_line: Optional 1-indexed end line (inclusive) :return: JSON with the file's id, filename, content, and pagination metadata if truncated """ if __request__ is None: @@ -1908,9 +2109,41 @@ async def view_knowledge_file( content = file.data.get('content', '') total_chars = len(content) + + # Line-based addressing (overrides char-based offset/max_chars) + if start_line is not None: + all_lines = content.split('\n') + total_lines = len(all_lines) + s = max(1, int(start_line)) - 1 + e = min(total_lines, int(end_line) if end_line else s + 100) + selected = all_lines[s:e] + sliced = '\n'.join(f'{s + i + 1}: {line}' for i, line in enumerate(selected)) + is_truncated = e < total_lines + result = { + 'id': file.id, + 'filename': file.filename, + 'content': sliced, + 'updated_at': file.updated_at, + 'created_at': file.created_at, + 'total_lines': total_lines, + 'showing_lines': f'{s + 1}-{e}', + } + if knowledge_info: + result['knowledge_id'] = knowledge_info['id'] + result['knowledge_name'] = knowledge_info['name'] + if is_truncated: + result['truncated'] = True + result['next_start_line'] = e + 1 + return json.dumps(result, ensure_ascii=False) + sliced = content[offset : offset + max_chars] is_truncated = (offset + len(sliced)) < total_chars + if line_numbers: + start_ln = content[:offset].count('\n') + 1 + lines = sliced.split('\n') + sliced = '\n'.join(f'{start_ln + i}: {line}' for i, line in enumerate(lines)) + result = { 'id': file.id, 'filename': file.filename, diff --git a/backend/open_webui/tools/knowledge_fs.py b/backend/open_webui/tools/knowledge_fs.py index 52c04dc9cf..b31f11497c 100644 --- a/backend/open_webui/tools/knowledge_fs.py +++ b/backend/open_webui/tools/knowledge_fs.py @@ -27,6 +27,42 @@ DEFAULT_TAIL_LINES = 10 MAX_GREP_MATCHES = 50 +# ============================================================================= +# SHARED REGEX UTILITIES — also used by builtin.py grep_knowledge_files +# ============================================================================= + + +def is_regex_pattern(pattern: str) -> bool: + """Detect if a pattern looks like regex (\|, .*, .+, \d, \w, \s, [...]).""" + return ('\|' in pattern or '.*' in pattern or '.+' in pattern + or '.?' in pattern or '\d' in pattern or '\w' in pattern + or '\s' in pattern or bool(re.search(r'\[.+\]', pattern))) + + +def normalize_regex(pattern: str) -> str: + """Normalize POSIX BRE patterns to Python regex (\| → |).""" + return pattern.replace('\\|', '|').replace('\|', '|') + + +def build_matcher(pattern: str, case_insensitive: bool = False, + use_regex: bool = False) -> tuple: + """Build a matcher function. Returns (match_fn, error_str_or_None).""" + if not use_regex and is_regex_pattern(pattern): + use_regex = True + + if use_regex: + normalized = normalize_regex(pattern) + try: + re_flags = re.IGNORECASE if case_insensitive else 0 + compiled = re.compile(normalized, re_flags) + except re.error as e: + return None, f'Invalid regex: {e}' + return (lambda line: bool(compiled.search(line))), None + else: + sp = pattern.lower() if case_insensitive else pattern + return (lambda line: sp in (line.lower() if case_insensitive else line)), None + + # ============================================================================= # COMMAND PARSING # ============================================================================= @@ -404,30 +440,9 @@ async def _kb_grep(args: list[str], flags: set[str], user: dict, count_only = 'c' in flags use_regex = 'E' in flags - # Auto-detect regex: models commonly use \| for alternation (POSIX BRE style), - # .* / .+ for wildcards, \d \w \s for character classes, or [...] brackets. - # If any of these are present, auto-promote to regex mode. - if not use_regex and ('\\|' in pattern or '.*' in pattern or '.+' in pattern - or '.?' in pattern or '\\d' in pattern or '\\w' in pattern - or '\\s' in pattern or re.search(r'\[.+\]', pattern)): - use_regex = True - - # Build matcher - if use_regex: - # Normalize \| to | (models trained on POSIX grep use \| for alternation) - normalized = pattern.replace('\\|', '|').replace('\|', '|') - try: - re_flags = re.IGNORECASE if case_insensitive else 0 - compiled = re.compile(normalized, re_flags) - except re.error as e: - return f'Invalid regex: {e}' - def _matches(line: str) -> bool: - return bool(compiled.search(line)) - else: - search_pattern = pattern.lower() if case_insensitive else pattern - def _matches(line: str) -> bool: - s = line.lower() if case_insensitive else line - return search_pattern in s + _matches, err = build_matcher(pattern, case_insensitive, use_regex) + if err: + return err # Grep on piped input if piped_input is not None: diff --git a/backend/open_webui/utils/tools.py b/backend/open_webui/utils/tools.py index 9fa478ea42..c18d1c4a5f 100644 --- a/backend/open_webui/utils/tools.py +++ b/backend/open_webui/utils/tools.py @@ -59,6 +59,7 @@ from open_webui.tools.builtin import ( fetch_url, generate_image, get_current_timestamp, + grep_knowledge_files, kb_exec, list_automations, list_knowledge, @@ -441,7 +442,7 @@ async def get_builtin_tools( if not model_knowledge: builtin_functions.append(query_knowledge_bases) elif model_knowledge: - builtin_functions.extend([list_knowledge, search_knowledge_files, query_knowledge_files]) + builtin_functions.extend([list_knowledge, search_knowledge_files, grep_knowledge_files, query_knowledge_files]) knowledge_types = {item.get('type') for item in model_knowledge} if 'file' in knowledge_types or 'collection' in knowledge_types: @@ -451,7 +452,8 @@ async def get_builtin_tools( else: builtin_functions.extend([ list_knowledge_bases, search_knowledge_bases, query_knowledge_bases, - search_knowledge_files, query_knowledge_files, view_knowledge_file, + grep_knowledge_files, search_knowledge_files, query_knowledge_files, + view_knowledge_file, ]) # Chats tools - search and fetch user's chat history