mirror of
https://github.com/open-webui/open-webui.git
synced 2026-08-02 22:20:49 -05:00
6be11d4fc9
Fourteen modules import `json` without using it. Ruff flags every one with F401, and a word-boundary search for `json` in each file matches only the import line itself, including inside strings, comments and annotations. Two exclusions, both deliberate. Migration files are left alone: the import is equally dead there, but those files are frozen history and not worth the churn. `models/chats.py` has the same dead import and is handled in its own change, so it is skipped here to avoid two changes touching the same line. No behaviour change.
1183 lines
40 KiB
Python
1183 lines
40 KiB
Python
"""
|
|
Knowledge Base Filesystem Interface.
|
|
|
|
Provides a filesystem-like command interface (ls, cat, grep, find, etc.)
|
|
for AI models to interact with knowledge bases using commands they already know.
|
|
|
|
Re-exported through builtin.py for consistent imports.
|
|
"""
|
|
|
|
import contextvars
|
|
import logging
|
|
import re
|
|
import shlex
|
|
import time
|
|
from contextlib import contextmanager
|
|
from typing import Optional
|
|
|
|
import regex
|
|
from fastapi import Request
|
|
|
|
from open_webui.env import (
|
|
KB_EXEC_MAX_GREP_FILES,
|
|
KB_EXEC_MAX_OUTPUT_CHARS,
|
|
KNOWLEDGE_GREP_MAX_MATCHES,
|
|
)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
DEFAULT_HEAD_LINES = 10
|
|
DEFAULT_TAIL_LINES = 10
|
|
|
|
# Matching time allowed per tool call. Backtracking cost is exponential in the length of the
|
|
# matched text, so capping the pattern or the line does not bound it.
|
|
MATCH_BUDGET_SECONDS = 2.0
|
|
|
|
|
|
class MatchBudgetExceeded(Exception):
|
|
"""A tool call spent its whole matching budget, so the caller reports it."""
|
|
|
|
|
|
class MatchBudget:
|
|
"""Matching time remaining, counted only inside search() so awaits do not consume it."""
|
|
|
|
def __init__(self):
|
|
self.remaining = MATCH_BUDGET_SECONDS
|
|
|
|
|
|
# Scoped to the running task, so one budget covers every matcher a command builds without
|
|
# threading it through each handler.
|
|
_active_budget: contextvars.ContextVar[MatchBudget | None] = contextvars.ContextVar('kb_match_budget', default=None)
|
|
|
|
|
|
@contextmanager
|
|
def match_budget():
|
|
"""Bound the matching time of one tool call rather than of each search it runs."""
|
|
token = _active_budget.set(MatchBudget())
|
|
try:
|
|
yield
|
|
finally:
|
|
_active_budget.reset(token)
|
|
|
|
|
|
# =============================================================================
|
|
# 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 = regex.IGNORECASE if case_insensitive else 0
|
|
compiled = regex.compile(normalized, re_flags)
|
|
except regex.error as e:
|
|
return None, f'Invalid regex: {e}'
|
|
|
|
budget = _active_budget.get() or MatchBudget()
|
|
|
|
def matches(line: str) -> bool:
|
|
started = time.monotonic()
|
|
try:
|
|
# A negative timeout disables it, so an exhausted budget must not reach search().
|
|
if budget.remaining <= 0:
|
|
raise TimeoutError
|
|
return bool(compiled.search(line, timeout=budget.remaining))
|
|
except TimeoutError:
|
|
raise MatchBudgetExceeded(f'Search exceeded {MATCH_BUDGET_SECONDS:g}s, narrow the pattern') from None
|
|
finally:
|
|
budget.remaining -= time.monotonic() - started
|
|
|
|
return matches, 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
|
|
# =============================================================================
|
|
|
|
|
|
def _parse_pipeline(command: str) -> list[list[str]]:
|
|
"""Split command on pipes, then tokenize each segment."""
|
|
# Split on | but not inside quotes
|
|
segments = []
|
|
current = []
|
|
in_single = False
|
|
in_double = False
|
|
buf = []
|
|
|
|
for ch in command:
|
|
if ch == "'" and not in_double:
|
|
in_single = not in_single
|
|
buf.append(ch)
|
|
elif ch == '"' and not in_single:
|
|
in_double = not in_double
|
|
buf.append(ch)
|
|
elif ch == '|' and not in_single and not in_double:
|
|
segments.append(''.join(buf).strip())
|
|
buf = []
|
|
else:
|
|
buf.append(ch)
|
|
|
|
remaining = ''.join(buf).strip()
|
|
if remaining:
|
|
segments.append(remaining)
|
|
|
|
result = []
|
|
for seg in segments:
|
|
if not seg:
|
|
continue
|
|
try:
|
|
tokens = shlex.split(seg)
|
|
except ValueError:
|
|
# Fallback for malformed quotes
|
|
tokens = seg.split()
|
|
if tokens:
|
|
result.append(tokens)
|
|
|
|
return result
|
|
|
|
|
|
def _extract_flags(tokens: list[str]) -> tuple[set[str], list[str]]:
|
|
"""Extract single-char flags (e.g. -i, -l, -c, -n, -la) from tokens.
|
|
|
|
Returns (flags_set, remaining_args).
|
|
"""
|
|
flags = set()
|
|
args = []
|
|
for token in tokens:
|
|
if token.startswith('-') and len(token) > 1 and not token[1:].isdigit():
|
|
# Could be -ilc (combined) or -20 (number, skip)
|
|
for ch in token[1:]:
|
|
flags.add(ch)
|
|
else:
|
|
args.append(token)
|
|
return flags, args
|
|
|
|
|
|
def _extract_numeric_flag(tokens: list[str]) -> tuple[Optional[int], list[str]]:
|
|
"""Extract a numeric flag like -20 from tokens. Returns (number, remaining)."""
|
|
num = None
|
|
remaining = []
|
|
for token in tokens:
|
|
if num is None and re.match(r'^-\d+$', token):
|
|
num = int(token[1:])
|
|
else:
|
|
remaining.append(token)
|
|
return num, remaining
|
|
|
|
|
|
# =============================================================================
|
|
# DIRECTORY TREE & PATH RESOLUTION
|
|
# =============================================================================
|
|
|
|
|
|
async def _build_directory_tree(knowledge_id: str) -> dict:
|
|
"""Build an in-memory directory tree for a KB. Returns {dirs, files, path_to_dir_id, dir_id_to_path}."""
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
all_dirs = await Knowledges.get_all_directories(knowledge_id)
|
|
files_with_dirs = await Knowledges.get_files_with_directory_ids(knowledge_id)
|
|
|
|
# Build dir_id -> dir info map
|
|
dir_map = {}
|
|
for d in all_dirs:
|
|
dir_map[d.id] = {'name': d.name, 'parent_id': d.parent_id, 'id': d.id}
|
|
|
|
# Compute full path for each directory
|
|
dir_id_to_path = {}
|
|
|
|
def _get_dir_path(dir_id):
|
|
if dir_id in dir_id_to_path:
|
|
return dir_id_to_path[dir_id]
|
|
d = dir_map.get(dir_id)
|
|
if not d:
|
|
return ''
|
|
if d['parent_id'] and d['parent_id'] in dir_map:
|
|
parent_path = _get_dir_path(d['parent_id'])
|
|
path = f'{parent_path}/{d["name"]}' if parent_path else d['name']
|
|
else:
|
|
path = d['name']
|
|
dir_id_to_path[dir_id] = path
|
|
return path
|
|
|
|
for d_id in dir_map:
|
|
_get_dir_path(d_id)
|
|
|
|
path_to_dir_id = {v: k for k, v in dir_id_to_path.items()}
|
|
|
|
# Build file list with paths
|
|
files = []
|
|
for file_model, directory_id in files_with_dirs:
|
|
if directory_id and directory_id in dir_id_to_path:
|
|
file_path = f'{dir_id_to_path[directory_id]}/{file_model.filename}'
|
|
else:
|
|
file_path = file_model.filename
|
|
files.append(
|
|
{
|
|
'id': file_model.id,
|
|
'filename': file_model.filename,
|
|
'path': file_path,
|
|
'directory_id': directory_id,
|
|
'size': file_model.meta.get('size') if file_model.meta else None,
|
|
'type': file_model.meta.get('content_type') if file_model.meta else None,
|
|
'updated_at': file_model.updated_at,
|
|
}
|
|
)
|
|
|
|
return {
|
|
'dirs': dir_map,
|
|
'files': files,
|
|
'path_to_dir_id': path_to_dir_id,
|
|
'dir_id_to_path': dir_id_to_path,
|
|
}
|
|
|
|
|
|
def _resolve_path(path: str, tree: dict) -> str | None:
|
|
"""Resolve a directory path string to a dir_id. Returns None if not found."""
|
|
path = path.strip('/')
|
|
return tree['path_to_dir_id'].get(path)
|
|
|
|
|
|
def _get_files_in_dir(tree: dict, dir_id: str | None) -> list[dict]:
|
|
"""Get files directly in a directory (None = root)."""
|
|
return [f for f in tree['files'] if f['directory_id'] == dir_id]
|
|
|
|
|
|
def _get_subdirs(tree: dict, parent_id: str | None) -> list[dict]:
|
|
"""Get immediate child directories."""
|
|
return sorted([d for d in tree['dirs'].values() if d['parent_id'] == parent_id], key=lambda d: d['name'])
|
|
|
|
|
|
def _get_files_under_dir(tree: dict, dir_id: str) -> list[dict]:
|
|
"""Get all files recursively under a directory."""
|
|
# Collect this dir + all descendant dir IDs
|
|
target_ids = {dir_id}
|
|
changed = True
|
|
while changed:
|
|
changed = False
|
|
for d in tree['dirs'].values():
|
|
if d['parent_id'] in target_ids and d['id'] not in target_ids:
|
|
target_ids.add(d['id'])
|
|
changed = True
|
|
return [f for f in tree['files'] if f['directory_id'] in target_ids]
|
|
|
|
|
|
# =============================================================================
|
|
# FILE RESOLUTION & ACCESS CONTROL
|
|
# =============================================================================
|
|
|
|
|
|
async def _get_accessible_kb_ids(
|
|
user: dict, model_knowledge: list[dict] | None, knowledge_id: str | None = None
|
|
) -> list[tuple[str, str, str]]:
|
|
"""Get list of (kb_id, kb_name, kb_description) the user can access."""
|
|
from open_webui.models.access_grants import AccessGrants
|
|
from open_webui.models.groups import Groups
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
user_id = user.get('id')
|
|
user_role = user.get('role', 'user')
|
|
user_group_ids = [g.id for g in await Groups.get_groups_by_member_id(user_id)]
|
|
|
|
async def _has_access(kb):
|
|
return (
|
|
user_role == 'admin'
|
|
or kb.user_id == user_id
|
|
or await AccessGrants.has_access(
|
|
user_id=user_id,
|
|
resource_type='knowledge',
|
|
resource_id=kb.id,
|
|
permission='read',
|
|
user_group_ids=set(user_group_ids),
|
|
)
|
|
)
|
|
|
|
result = []
|
|
|
|
if model_knowledge:
|
|
attached_kb_ids = set()
|
|
for item in model_knowledge:
|
|
if item.get('type') == 'collection':
|
|
attached_kb_ids.add(item.get('id'))
|
|
if knowledge_id:
|
|
if knowledge_id not in attached_kb_ids:
|
|
return []
|
|
attached_kb_ids = {knowledge_id}
|
|
for kb_id in attached_kb_ids:
|
|
kb = await Knowledges.get_knowledge_by_id(kb_id)
|
|
if kb and await _has_access(kb):
|
|
result.append((kb.id, kb.name, kb.description or ''))
|
|
elif knowledge_id:
|
|
kb = await Knowledges.get_knowledge_by_id(knowledge_id)
|
|
if kb and await _has_access(kb):
|
|
result.append((kb.id, kb.name, kb.description or ''))
|
|
else:
|
|
search = await Knowledges.search_knowledge_bases(
|
|
user_id,
|
|
filter={'query': '', 'user_id': user_id, 'group_ids': user_group_ids},
|
|
skip=0,
|
|
limit=50,
|
|
)
|
|
for kb in search.items:
|
|
result.append((kb.id, kb.name, kb.description or ''))
|
|
|
|
return result
|
|
|
|
|
|
async def _get_accessible_files(
|
|
user: dict, model_knowledge: list[dict] | None, knowledge_id: str | None = None
|
|
) -> list[dict]:
|
|
"""Get all files the user can access, with KB metadata and directory_id (no path computation)."""
|
|
from open_webui.models.files import Files
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
kb_ids = await _get_accessible_kb_ids(user, model_knowledge, knowledge_id)
|
|
files = []
|
|
|
|
for kb_id, kb_name, _ in kb_ids:
|
|
kb_files = await Knowledges.get_files_with_directory_ids(kb_id)
|
|
for file_model, dir_id in kb_files:
|
|
files.append(
|
|
{
|
|
'id': file_model.id,
|
|
'filename': file_model.filename,
|
|
'directory_id': dir_id,
|
|
'size': file_model.meta.get('size') if file_model.meta else None,
|
|
'type': file_model.meta.get('content_type') if file_model.meta else None,
|
|
'updated_at': file_model.updated_at,
|
|
'knowledge_id': kb_id,
|
|
'knowledge_name': kb_name,
|
|
}
|
|
)
|
|
|
|
# Also handle directly attached files (not in any KB)
|
|
if model_knowledge:
|
|
attached_file_ids = set()
|
|
for item in model_knowledge:
|
|
if item.get('type') == 'file':
|
|
attached_file_ids.add(item.get('id'))
|
|
for fid in attached_file_ids:
|
|
f = await Files.get_file_by_id(fid)
|
|
if f:
|
|
files.append(
|
|
{
|
|
'id': f.id,
|
|
'filename': f.filename,
|
|
'directory_id': None,
|
|
'size': f.meta.get('size') if f.meta else None,
|
|
'type': f.meta.get('content_type') if f.meta else None,
|
|
'updated_at': f.updated_at,
|
|
'knowledge_id': None,
|
|
'knowledge_name': None,
|
|
}
|
|
)
|
|
|
|
return files
|
|
|
|
|
|
async def _resolve_dir_path(path: str, knowledge_id: str) -> str | None:
|
|
"""Walk a directory path one level at a time. Returns dir_id or None."""
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
parts = path.strip('/').split('/')
|
|
current_parent = None
|
|
|
|
for part in parts:
|
|
dirs = await Knowledges.get_directories(knowledge_id, parent_id=current_parent)
|
|
match = next((d for d in dirs if d.name == part), None)
|
|
if not match:
|
|
return None
|
|
current_parent = match.id
|
|
|
|
return current_parent
|
|
|
|
|
|
async def _get_descendant_dir_ids(dir_id: str, knowledge_id: str) -> set[str]:
|
|
"""Collect all descendant directory IDs recursively."""
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
result = {dir_id}
|
|
queue = [dir_id]
|
|
while queue:
|
|
parent = queue.pop()
|
|
children = await Knowledges.get_directories(knowledge_id, parent_id=parent)
|
|
for child in children:
|
|
if child.id not in result:
|
|
result.add(child.id)
|
|
queue.append(child.id)
|
|
return result
|
|
|
|
|
|
async def _resolve_file(ref: str, user: dict, model_knowledge: list[dict] | None) -> dict | None:
|
|
"""Resolve a file reference (ID, path, or filename) to a file info dict with content."""
|
|
from open_webui.models.files import Files
|
|
|
|
# Get accessible file IDs (lightweight — no path computation)
|
|
accessible = await _get_accessible_files(user, model_knowledge)
|
|
accessible_ids = {fi['id'] for fi in accessible}
|
|
|
|
# Try direct ID lookup first — but verify access
|
|
f = await Files.get_file_by_id(ref)
|
|
if f and f.data:
|
|
if f.id not in accessible_ids:
|
|
return None
|
|
return {
|
|
'id': f.id,
|
|
'filename': f.filename,
|
|
'content': f.data.get('content', ''),
|
|
'meta': f.meta,
|
|
'updated_at': f.updated_at,
|
|
'created_at': f.created_at,
|
|
}
|
|
|
|
# Try path match (e.g. "docs/api/auth.md") — lazy dir walk
|
|
ref_clean = ref.strip('/')
|
|
if '/' in ref_clean:
|
|
dir_path, filename = ref_clean.rsplit('/', 1)
|
|
# Try resolving in each accessible KB
|
|
kb_ids = {fi['knowledge_id'] for fi in accessible if fi.get('knowledge_id')}
|
|
for kb_id in kb_ids:
|
|
dir_id = await _resolve_dir_path(dir_path, kb_id)
|
|
if dir_id is None:
|
|
continue
|
|
# Find file with that name in that directory
|
|
matches = [fi for fi in accessible if fi['filename'] == filename and fi['directory_id'] == dir_id]
|
|
if len(matches) == 1:
|
|
f = await Files.get_file_by_id(matches[0]['id'])
|
|
if f and f.data:
|
|
return {
|
|
'id': f.id,
|
|
'filename': f.filename,
|
|
'content': f.data.get('content', ''),
|
|
'meta': f.meta,
|
|
'updated_at': f.updated_at,
|
|
'created_at': f.created_at,
|
|
'knowledge_id': matches[0].get('knowledge_id'),
|
|
'knowledge_name': matches[0].get('knowledge_name'),
|
|
}
|
|
|
|
# Try filename match within accessible files
|
|
matches = [fi for fi in accessible if fi['filename'] == ref]
|
|
|
|
if len(matches) == 1:
|
|
f = await Files.get_file_by_id(matches[0]['id'])
|
|
if f and f.data:
|
|
return {
|
|
'id': f.id,
|
|
'filename': f.filename,
|
|
'content': f.data.get('content', ''),
|
|
'meta': f.meta,
|
|
'updated_at': f.updated_at,
|
|
'created_at': f.created_at,
|
|
'knowledge_id': matches[0].get('knowledge_id'),
|
|
'knowledge_name': matches[0].get('knowledge_name'),
|
|
}
|
|
elif len(matches) > 1:
|
|
return {
|
|
'error': f'Ambiguous filename "{ref}". Use full path to disambiguate:\n'
|
|
+ '\n'.join(f' {m["id"]} {m["filename"]} ({m.get("knowledge_name", "direct")})' for m in matches)
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
async def _get_file_content(file_id: str) -> str | None:
|
|
"""Get file content by ID."""
|
|
from open_webui.models.files import Files
|
|
|
|
f = await Files.get_file_by_id(file_id)
|
|
if f and f.data:
|
|
return f.data.get('content', '')
|
|
return None
|
|
|
|
|
|
# =============================================================================
|
|
# COMMAND HANDLERS
|
|
# =============================================================================
|
|
|
|
|
|
async def _kb_ls(args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None) -> str:
|
|
"""List files and directories. Supports: ls, ls <path>, ls -a (flat)."""
|
|
from open_webui.models.knowledge import Knowledges
|
|
|
|
flat_mode = 'a' in flags
|
|
path_arg = args[0] if args else None
|
|
|
|
kb_ids = await _get_accessible_kb_ids(user, model_knowledge, knowledge_id=None)
|
|
direct_files = (
|
|
[f for f in await _get_accessible_files(user, model_knowledge) if not f.get('knowledge_id')]
|
|
if model_knowledge
|
|
else []
|
|
)
|
|
|
|
# If path_arg looks like a KB ID, scope to that KB
|
|
target_kb_id = None
|
|
dir_path = None
|
|
if path_arg:
|
|
for kb_id, kb_name, _ in kb_ids:
|
|
if kb_id == path_arg:
|
|
target_kb_id = kb_id
|
|
break
|
|
if not target_kb_id:
|
|
dir_path = path_arg.strip('/')
|
|
|
|
if target_kb_id:
|
|
kb_ids = [(kid, kn, kd) for kid, kn, kd in kb_ids if kid == target_kb_id]
|
|
|
|
if not kb_ids and not direct_files:
|
|
return 'No knowledge bases found.'
|
|
|
|
lines = []
|
|
for kb_id, kb_name, kb_desc in kb_ids:
|
|
header = f'Knowledge Base: {kb_name} ({kb_id})'
|
|
if kb_desc:
|
|
header += f'\n {kb_desc}'
|
|
lines.append(header)
|
|
|
|
if flat_mode:
|
|
# Flat mode: build full tree (legitimate use)
|
|
tree = await _build_directory_tree(kb_id)
|
|
for f in tree['files']:
|
|
lines.append(f' {f["id"]} {f["path"]} {_fmt_size(f)} {_fmt_date(f)}')
|
|
lines.append('')
|
|
continue
|
|
|
|
# Resolve target directory (lazy walk)
|
|
target_dir_id = None
|
|
if dir_path:
|
|
target_dir_id = await _resolve_dir_path(dir_path, kb_id)
|
|
if target_dir_id is None:
|
|
lines.append(f' Directory not found: {dir_path}')
|
|
lines.append('')
|
|
continue
|
|
lines.append(f' Path: {dir_path}/')
|
|
|
|
# Show subdirectories (targeted query — only this level)
|
|
subdirs = await Knowledges.get_directories(kb_id, parent_id=target_dir_id)
|
|
for d in subdirs:
|
|
lines.append(f' 📁 {d.name}/')
|
|
|
|
# Show files at this level (filter from accessible files)
|
|
accessible = await _get_accessible_files(user, model_knowledge, knowledge_id=kb_id)
|
|
dir_files = [f for f in accessible if f['directory_id'] == target_dir_id]
|
|
for f in dir_files:
|
|
lines.append(f' {f["id"]} {f["filename"]} {_fmt_size(f)} {_fmt_date(f)}')
|
|
|
|
if not subdirs and not dir_files:
|
|
lines.append(' (empty)')
|
|
lines.append('')
|
|
|
|
if direct_files and not target_kb_id and not dir_path:
|
|
lines.append('Attached Files:')
|
|
for f in direct_files:
|
|
lines.append(f' {f["id"]} {f["filename"]} {_fmt_size(f)} {_fmt_date(f)}')
|
|
lines.append('')
|
|
|
|
return '\n'.join(lines).rstrip()
|
|
|
|
|
|
def _fmt_size(f: dict) -> str:
|
|
return f'{f["size"]:,} bytes' if f.get('size') else ''
|
|
|
|
|
|
def _fmt_date(f: dict) -> str:
|
|
if f.get('updated_at'):
|
|
from datetime import datetime, timezone
|
|
|
|
dt = datetime.fromtimestamp(f['updated_at'], tz=timezone.utc)
|
|
return dt.strftime('%Y-%m-%d')
|
|
return ''
|
|
|
|
|
|
async def _kb_cat(args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None) -> str:
|
|
"""Read file content. Use -n for line numbers."""
|
|
if not args:
|
|
return 'Usage: cat [-n] <file_id or filename>'
|
|
|
|
resolved = await _resolve_file(args[0], user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {args[0]}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
content = resolved['content']
|
|
if 'n' in flags:
|
|
lines = content.split('\n')
|
|
content = '\n'.join(f'{i}: {line}' for i, line in enumerate(lines, 1))
|
|
|
|
return content
|
|
|
|
|
|
async def _kb_head(
|
|
args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None, piped_input: str | None = None
|
|
) -> str:
|
|
"""First N lines of a file or piped input."""
|
|
n, args = _extract_numeric_flag(args)
|
|
if n is None:
|
|
n = DEFAULT_HEAD_LINES
|
|
|
|
if piped_input is not None:
|
|
lines = piped_input.split('\n')
|
|
return '\n'.join(lines[:n])
|
|
|
|
if not args:
|
|
return 'Usage: head [-N] <file>'
|
|
|
|
resolved = await _resolve_file(args[0], user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {args[0]}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
lines = resolved['content'].split('\n')
|
|
total = len(lines)
|
|
result = '\n'.join(lines[:n])
|
|
if total > n:
|
|
result += f'\n[showing {n} of {total} lines]'
|
|
return result
|
|
|
|
|
|
async def _kb_tail(
|
|
args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None, piped_input: str | None = None
|
|
) -> str:
|
|
"""Last N lines of a file or piped input."""
|
|
n, args = _extract_numeric_flag(args)
|
|
if n is None:
|
|
n = DEFAULT_TAIL_LINES
|
|
|
|
if piped_input is not None:
|
|
lines = piped_input.split('\n')
|
|
return '\n'.join(lines[-n:])
|
|
|
|
if not args:
|
|
return 'Usage: tail [-N] <file>'
|
|
|
|
resolved = await _resolve_file(args[0], user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {args[0]}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
lines = resolved['content'].split('\n')
|
|
total = len(lines)
|
|
result = '\n'.join(lines[-n:])
|
|
if total > n:
|
|
result += f'\n[showing last {n} of {total} lines]'
|
|
return result
|
|
|
|
|
|
async def _kb_grep(
|
|
args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None, piped_input: str | None = None
|
|
) -> str:
|
|
"""Text search across files or piped input. Supports -E for regex."""
|
|
if not args:
|
|
return 'Usage: grep [-E] [-i] [-l] [-c] "pattern" [file] [*.ext]'
|
|
|
|
pattern = args[0]
|
|
file_ref = None
|
|
ext_filter = None
|
|
dir_scope = None
|
|
|
|
for arg in args[1:]:
|
|
if '*' in arg or arg.startswith('.'):
|
|
ext_filter = arg.lstrip('*').lstrip('.')
|
|
elif arg.endswith('/'):
|
|
dir_scope = arg.strip('/')
|
|
else:
|
|
file_ref = arg
|
|
|
|
case_insensitive = 'i' in flags
|
|
filenames_only = 'l' in flags
|
|
count_only = 'c' in flags
|
|
use_regex = 'E' in flags
|
|
|
|
_matches, err = build_matcher(pattern, case_insensitive, use_regex)
|
|
if err:
|
|
return err
|
|
|
|
# Grep on piped input
|
|
if piped_input is not None:
|
|
lines = piped_input.split('\n')
|
|
matched = []
|
|
for i, line in enumerate(lines, 1):
|
|
if _matches(line):
|
|
matched.append(f'{i}: {line}')
|
|
if count_only:
|
|
return str(len(matched))
|
|
if filenames_only:
|
|
return '(standard input)' if matched else f'No matches for "{pattern}"'
|
|
return '\n'.join(matched) if matched else f'No matches for "{pattern}"'
|
|
|
|
# Single file grep
|
|
if file_ref and not dir_scope:
|
|
resolved = await _resolve_file(file_ref, user, model_knowledge)
|
|
if not resolved:
|
|
# Maybe it's a directory path without trailing /
|
|
dir_scope = file_ref
|
|
elif 'error' in resolved:
|
|
return resolved['error']
|
|
else:
|
|
lines = resolved['content'].split('\n')
|
|
matched = []
|
|
for i, line in enumerate(lines, 1):
|
|
if _matches(line):
|
|
matched.append(f'{i}: {line}')
|
|
|
|
if count_only:
|
|
return f'{resolved["id"]} {resolved["filename"]}: {len(matched)}'
|
|
if filenames_only:
|
|
return f'{resolved["id"]} {resolved["filename"]}' if matched else f'No matches for "{pattern}"'
|
|
|
|
if not matched:
|
|
return f'No matches for "{pattern}" in {resolved["filename"]}'
|
|
return '\n'.join(matched)
|
|
|
|
# Cross-file grep (optionally scoped to directory)
|
|
accessible = await _get_accessible_files(user, model_knowledge)
|
|
|
|
if dir_scope:
|
|
# Resolve directory and collect all descendant IDs
|
|
kb_ids = {fi['knowledge_id'] for fi in accessible if fi.get('knowledge_id')}
|
|
target_dir_ids = set()
|
|
for kb_id in kb_ids:
|
|
dir_id = await _resolve_dir_path(dir_scope, kb_id)
|
|
if dir_id:
|
|
desc = await _get_descendant_dir_ids(dir_id, kb_id)
|
|
target_dir_ids.update(desc)
|
|
if not target_dir_ids:
|
|
return f'No files found under "{dir_scope}/"'
|
|
accessible = [f for f in accessible if f.get('directory_id') in target_dir_ids]
|
|
if not accessible:
|
|
return f'No files found under "{dir_scope}/"'
|
|
|
|
if ext_filter:
|
|
accessible = [f for f in accessible if f['filename'].endswith(f'.{ext_filter}')]
|
|
|
|
if len(accessible) > KB_EXEC_MAX_GREP_FILES:
|
|
return f'Too many files ({len(accessible)}). Scope your search: grep "{pattern}" docs/ or grep "{pattern}" *.py'
|
|
|
|
from open_webui.models.files import Files
|
|
|
|
results = []
|
|
file_match_counts = []
|
|
files_with_matches = []
|
|
total_matches = 0
|
|
|
|
for file_info in accessible:
|
|
f = await Files.get_file_by_id(file_info['id'])
|
|
if not f or not f.data:
|
|
continue
|
|
|
|
content = f.data.get('content', '')
|
|
if not content:
|
|
continue
|
|
|
|
lines = content.split('\n')
|
|
file_matches = []
|
|
for i, line in enumerate(lines, 1):
|
|
if _matches(line):
|
|
file_matches.append((i, line))
|
|
|
|
if file_matches:
|
|
files_with_matches.append(file_info)
|
|
file_match_counts.append((file_info, len(file_matches)))
|
|
total_matches += len(file_matches)
|
|
|
|
if not count_only and not filenames_only:
|
|
for line_num, line_text in file_matches:
|
|
if len(results) < KNOWLEDGE_GREP_MAX_MATCHES:
|
|
results.append(f'{file_info["id"]} {file_info["filename"]}:{line_num}: {line_text.rstrip()}')
|
|
|
|
if count_only:
|
|
if not file_match_counts:
|
|
return f'No matches for "{pattern}"'
|
|
lines = [f'{fi["id"]} {fi["filename"]}: {cnt}' for fi, cnt in file_match_counts]
|
|
lines.append(f'Total: {total_matches} matches in {len(file_match_counts)} files')
|
|
return '\n'.join(lines)
|
|
|
|
if filenames_only:
|
|
if not files_with_matches:
|
|
return f'No matches for "{pattern}"'
|
|
return '\n'.join(f'{fi["id"]} {fi["filename"]}' for fi in files_with_matches)
|
|
|
|
if not results:
|
|
return f'No matches for "{pattern}" across {len(accessible)} files'
|
|
|
|
output = '\n'.join(results)
|
|
if total_matches > KNOWLEDGE_GREP_MAX_MATCHES:
|
|
output += f'\n[showing {KNOWLEDGE_GREP_MAX_MATCHES} of {total_matches} matches]'
|
|
return output
|
|
|
|
|
|
async def _kb_find(args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None) -> str:
|
|
"""Find files by name/glob pattern, optionally scoped to a directory."""
|
|
if not args:
|
|
return 'Usage: find "*.md" or find docs/ "*.md"'
|
|
|
|
import fnmatch
|
|
|
|
# If two args and first looks like a dir scope
|
|
dir_scope = None
|
|
if len(args) >= 2 and ('/' in args[0] or not ('*' in args[0] or '?' in args[0])):
|
|
dir_scope = args[0].strip('/')
|
|
pattern = args[1]
|
|
else:
|
|
pattern = args[0]
|
|
|
|
accessible = await _get_accessible_files(user, model_knowledge)
|
|
|
|
if dir_scope:
|
|
kb_ids = {fi['knowledge_id'] for fi in accessible if fi.get('knowledge_id')}
|
|
target_dir_ids = set()
|
|
for kb_id in kb_ids:
|
|
dir_id = await _resolve_dir_path(dir_scope, kb_id)
|
|
if dir_id:
|
|
desc = await _get_descendant_dir_ids(dir_id, kb_id)
|
|
target_dir_ids.update(desc)
|
|
accessible = [f for f in accessible if f.get('directory_id') in target_dir_ids]
|
|
|
|
matched = [f for f in accessible if fnmatch.fnmatch(f['filename'], pattern)]
|
|
|
|
if not matched:
|
|
scope_str = f' under "{dir_scope}/"' if dir_scope else ''
|
|
return f'No files matching "{pattern}"{scope_str}'
|
|
|
|
lines = []
|
|
for f in matched:
|
|
kb_info = f' ({f["knowledge_name"]})' if f.get('knowledge_name') else ''
|
|
lines.append(f'{f["id"]} {f["filename"]}{kb_info}')
|
|
return '\n'.join(lines)
|
|
|
|
|
|
async def _kb_wc(
|
|
args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None, piped_input: str | None = None
|
|
) -> str:
|
|
"""Word, line, character counts."""
|
|
if piped_input is not None:
|
|
lines = piped_input.count('\n') + (1 if piped_input and not piped_input.endswith('\n') else 0)
|
|
words = len(piped_input.split())
|
|
chars = len(piped_input)
|
|
if 'l' in flags:
|
|
return str(lines)
|
|
return f' {lines} {words} {chars}'
|
|
|
|
if not args:
|
|
return 'Usage: wc [-l] <file>'
|
|
|
|
resolved = await _resolve_file(args[0], user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {args[0]}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
content = resolved['content']
|
|
lines = content.count('\n') + (1 if content and not content.endswith('\n') else 0)
|
|
words = len(content.split())
|
|
chars = len(content)
|
|
|
|
if 'l' in flags:
|
|
return f' {lines} {resolved["filename"]}'
|
|
return f' {lines} {words} {chars} {resolved["filename"]}'
|
|
|
|
|
|
async def _kb_stat(args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None) -> str:
|
|
"""File metadata."""
|
|
if not args:
|
|
return 'Usage: stat <file>'
|
|
|
|
resolved = await _resolve_file(args[0], user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {args[0]}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
content = resolved['content']
|
|
lines = content.count('\n') + (1 if content and not content.endswith('\n') else 0)
|
|
words = len(content.split())
|
|
chars = len(content)
|
|
|
|
meta = resolved.get('meta') or {}
|
|
size = meta.get('size', chars)
|
|
content_type = meta.get('content_type', 'unknown')
|
|
|
|
out = [
|
|
f' File: {resolved["filename"]}',
|
|
f' ID: {resolved["id"]}',
|
|
f' Size: {size:,} bytes',
|
|
f' Type: {content_type}',
|
|
f' Lines: {lines:,}',
|
|
f' Words: {words:,}',
|
|
f' Chars: {chars:,}',
|
|
]
|
|
|
|
if resolved.get('created_at'):
|
|
from datetime import datetime, timezone
|
|
|
|
dt = datetime.fromtimestamp(resolved['created_at'], tz=timezone.utc)
|
|
out.append(f' Created: {dt.strftime("%Y-%m-%d %H:%M:%S UTC")}')
|
|
if resolved.get('updated_at'):
|
|
from datetime import datetime, timezone
|
|
|
|
dt = datetime.fromtimestamp(resolved['updated_at'], tz=timezone.utc)
|
|
out.append(f' Updated: {dt.strftime("%Y-%m-%d %H:%M:%S UTC")}')
|
|
if resolved.get('knowledge_name'):
|
|
out.append(f' KB: {resolved["knowledge_name"]} ({resolved.get("knowledge_id", "")})')
|
|
|
|
return '\n'.join(out)
|
|
|
|
|
|
async def _kb_sed(
|
|
args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None, piped_input: str | None = None
|
|
) -> str:
|
|
"""Extract line range from a file. Usage: sed -n 'M,Np' <file>"""
|
|
if piped_input is not None:
|
|
# sed on piped input: parse range from args
|
|
start, end = 1, None
|
|
if 'n' in flags and args:
|
|
m = re.match(r'^(\d+),(\d+)p?$', args[0])
|
|
if m:
|
|
start, end = int(m.group(1)), int(m.group(2))
|
|
args = args[1:]
|
|
lines = piped_input.split('\n')
|
|
selected = lines[max(0, start - 1) : (end or len(lines))]
|
|
return '\n'.join(selected)
|
|
|
|
# Parse: sed -n '40,60p' <file>
|
|
range_str = None
|
|
file_ref = None
|
|
|
|
for arg in args:
|
|
m = re.match(r"^'?(\d+),(\d+)p?'?$", arg)
|
|
if m:
|
|
range_str = arg
|
|
else:
|
|
file_ref = arg
|
|
|
|
if not range_str or not file_ref:
|
|
return "Usage: sed -n '40,60p' <file>"
|
|
|
|
m = re.match(r"^'?(\d+),(\d+)p?'?$", range_str)
|
|
start, end = int(m.group(1)), int(m.group(2))
|
|
|
|
if start > end:
|
|
return f'Invalid range: start ({start}) > end ({end})'
|
|
|
|
resolved = await _resolve_file(file_ref, user, model_knowledge)
|
|
if not resolved:
|
|
return f'File not found: {file_ref}'
|
|
if 'error' in resolved:
|
|
return resolved['error']
|
|
|
|
lines = resolved['content'].split('\n')
|
|
total = len(lines)
|
|
selected = lines[max(0, start - 1) : end]
|
|
result = '\n'.join(selected)
|
|
result += f'\n[lines {start}-{min(end, total)} of {total}]'
|
|
return result
|
|
|
|
|
|
# =============================================================================
|
|
# PIPE EXECUTOR
|
|
# =============================================================================
|
|
|
|
|
|
async def _kb_tree(args: list[str], flags: set[str], user: dict, model_knowledge: list[dict] | None) -> str:
|
|
"""Show directory tree structure."""
|
|
kb_ids = await _get_accessible_kb_ids(user, model_knowledge)
|
|
direct_files = (
|
|
[f for f in await _get_accessible_files(user, model_knowledge) if not f.get('knowledge_id')]
|
|
if model_knowledge
|
|
else []
|
|
)
|
|
if not kb_ids and not direct_files:
|
|
return 'No knowledge bases found.'
|
|
|
|
dir_scope = args[0].strip('/') if args else None
|
|
output = []
|
|
|
|
for kb_id, kb_name, kb_desc in kb_ids:
|
|
tree = await _build_directory_tree(kb_id)
|
|
header = f'Knowledge Base: {kb_name} ({kb_id})'
|
|
if kb_desc:
|
|
header += f'\n {kb_desc}'
|
|
output.append(header)
|
|
|
|
# Find root to start from
|
|
root_dir_id = None
|
|
if dir_scope:
|
|
root_dir_id = _resolve_path(dir_scope, tree)
|
|
if root_dir_id is None:
|
|
output.append(f' Directory not found: {dir_scope}')
|
|
output.append('')
|
|
continue
|
|
output.append(f' {dir_scope}/')
|
|
|
|
def _render_tree(parent_id, prefix=' '):
|
|
items = []
|
|
subdirs = _get_subdirs(tree, parent_id)
|
|
files = _get_files_in_dir(tree, parent_id)
|
|
entries = [('dir', d) for d in subdirs] + [('file', f) for f in files]
|
|
|
|
for idx, (etype, entry) in enumerate(entries):
|
|
is_last = idx == len(entries) - 1
|
|
connector = '└── ' if is_last else '├── '
|
|
child_prefix = prefix + (' ' if is_last else '│ ')
|
|
|
|
if etype == 'dir':
|
|
items.append(f'{prefix}{connector}📁 {entry["name"]}/')
|
|
items.extend(_render_tree(entry['id'], child_prefix))
|
|
else:
|
|
items.append(f'{prefix}{connector}{entry["filename"]}')
|
|
return items
|
|
|
|
output.extend(_render_tree(root_dir_id))
|
|
|
|
# Summary
|
|
total_dirs = len(tree['dirs'])
|
|
total_files = len(tree['files'])
|
|
output.append(f'\n {total_dirs} directories, {total_files} files')
|
|
output.append('')
|
|
|
|
if direct_files and not dir_scope:
|
|
output.append('Attached Files:')
|
|
for idx, f in enumerate(direct_files):
|
|
connector = '└── ' if idx == len(direct_files) - 1 else '├── '
|
|
output.append(f' {connector}{f["filename"]}')
|
|
output.append(f'\n 0 directories, {len(direct_files)} files')
|
|
output.append('')
|
|
|
|
return '\n'.join(output).rstrip()
|
|
|
|
|
|
COMMAND_MAP = {
|
|
'ls': _kb_ls,
|
|
'cat': _kb_cat,
|
|
'head': _kb_head,
|
|
'tail': _kb_tail,
|
|
'grep': _kb_grep,
|
|
'find': _kb_find,
|
|
'wc': _kb_wc,
|
|
'stat': _kb_stat,
|
|
'sed': _kb_sed,
|
|
'tree': _kb_tree,
|
|
}
|
|
|
|
|
|
async def _execute_pipeline(
|
|
segments: list[list[str]],
|
|
user: dict,
|
|
model_knowledge: list[dict] | None,
|
|
) -> str:
|
|
"""Execute a pipeline of commands, passing text between them."""
|
|
piped_input = None
|
|
|
|
for tokens in segments:
|
|
cmd_name = tokens[0].lower()
|
|
rest = tokens[1:]
|
|
|
|
handler = COMMAND_MAP.get(cmd_name)
|
|
if not handler:
|
|
return f'Unknown command: {cmd_name}. Available: {", ".join(sorted(COMMAND_MAP.keys()))}'
|
|
|
|
flags, args = _extract_flags(rest)
|
|
|
|
# Commands that accept piped input
|
|
if piped_input is not None and cmd_name in ('head', 'tail', 'grep', 'wc', 'sed'):
|
|
piped_input = await handler(args, flags, user, model_knowledge, piped_input=piped_input)
|
|
else:
|
|
piped_input = await handler(args, flags, user, model_knowledge)
|
|
|
|
return piped_input or ''
|
|
|
|
|
|
# =============================================================================
|
|
# ENTRY POINT
|
|
# =============================================================================
|
|
|
|
|
|
async def kb_exec(
|
|
command: str,
|
|
__request__: Request = None,
|
|
__user__: dict = None,
|
|
__model_knowledge__: Optional[list[dict]] = None,
|
|
) -> str:
|
|
"""
|
|
Run a filesystem command against the knowledge base.
|
|
|
|
Commands:
|
|
ls — list root files and directories
|
|
ls docs/ — list contents of a directory
|
|
ls -a — flat list of all files with full paths
|
|
tree — recursive directory tree view
|
|
tree docs/ — subtree from a directory
|
|
cat -n <file> — read file with line numbers
|
|
head -20 <file> — first 20 lines
|
|
tail -10 <file> — last 10 lines
|
|
sed -n '40,60p' <file> — view lines 40-60
|
|
grep "text" <file> — exact text search (auto-detects regex)
|
|
grep -i "text" — case-insensitive
|
|
grep -l "text" — filenames-only
|
|
grep -c "text" — match counts
|
|
grep "text" docs/ — search within a directory
|
|
grep "text" *.py — filter by extension
|
|
find "*.md" — find files by glob
|
|
find docs/ "*.md" — find within a directory
|
|
wc <file> — line/word/char counts
|
|
stat <file> — file metadata
|
|
|
|
Pipes: grep "auth" | head -5
|
|
Files: reference by path (docs/api/auth.md), filename, or file ID
|
|
|
|
:param command: A filesystem command string
|
|
:return: Command output as text
|
|
"""
|
|
if not __user__:
|
|
return 'Error: User context not available'
|
|
|
|
if not command or not command.strip():
|
|
return 'Usage: kb_exec("<command>"). Run kb_exec("ls") to start.'
|
|
|
|
try:
|
|
segments = _parse_pipeline(command.strip())
|
|
if not segments:
|
|
return 'Could not parse command. Run kb_exec("ls") to start.'
|
|
|
|
# One budget for the whole command: a per-search budget would multiply by segment count.
|
|
with match_budget():
|
|
output = await _execute_pipeline(segments, __user__, __model_knowledge__)
|
|
if len(output) > KB_EXEC_MAX_OUTPUT_CHARS:
|
|
output = output[:KB_EXEC_MAX_OUTPUT_CHARS] + (
|
|
f'\n[output truncated at {KB_EXEC_MAX_OUTPUT_CHARS:,} chars'
|
|
' — narrow the command with a path, glob, head/tail/sed or grep]'
|
|
)
|
|
return output
|
|
except Exception as e:
|
|
log.exception(f'kb_exec error: {e}')
|
|
return f'Error: {e}'
|