Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions openviking/storage/viking_fs.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import os
import re
import time
import unicodedata
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
Expand Down Expand Up @@ -87,6 +88,25 @@ def _ensure_non_empty_search_query(query: str) -> None:
raise InvalidArgumentError("Search query must not be empty.")


def _normalize_search_query(query: str) -> str:
"""Normalize a semantic search query for consistent embedding/tokenization.

NFKC + casefold + whitespace collapse so that colloquial or CJK
full/half-width variants of the same query embed and tokenize
consistently with indexed content (e.g. "OpenViking" / "openvaking"
case-fold to a form matching "OpenViking"). Idempotent and only widens
recall — does not alter semantics for already-normalized ASCII.

Applied only to the semantic find/search path, NOT to grep (whose
pattern is a regular expression; casefold/NFKC would corrupt explicit
character classes and the ``case_insensitive`` flag already covers
case folding there).
"""
if not query:
return query
return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", query).casefold()).strip()


def _is_directory_not_empty_error(message: str) -> bool:
"""Check if an error message indicates a directory not empty error.

Expand Down Expand Up @@ -1925,6 +1945,7 @@ async def find(
FindResult
"""
_ensure_non_empty_search_query(query)
query = _normalize_search_query(query)
telemetry = get_current_telemetry()
from openviking.retrieve.hierarchical_retriever import HierarchicalRetriever
from openviking_cli.retrieve import (
Expand Down Expand Up @@ -2017,6 +2038,7 @@ async def search(
FindResult
"""
_ensure_non_empty_search_query(query)
query = _normalize_search_query(query)
telemetry = get_current_telemetry()
from openviking.retrieve.hierarchical_retriever import HierarchicalRetriever
from openviking.retrieve.intent_analyzer import IntentAnalyzer
Expand Down
41 changes: 41 additions & 0 deletions tests/unit/test_search_query_normalization.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0

"""Tests for semantic search query normalization."""

from openviking.storage.viking_fs import _normalize_search_query


def test_normalize_cjk_fullwidth_to_halfwidth():
# NFKC folds full-width Latin letters/digits to ASCII
assert _normalize_search_query("OpenViking") == "openviking"


def test_normalize_casefold_mixed_case():
# casefold is stronger than lower() — handles ß, etc.
assert _normalize_search_query("OpenViking") == "openviking"
assert _normalize_search_query("OpenVAKING") == "openvaking"


def test_normalize_collapses_whitespace():
assert _normalize_search_query("Harms agent") == "harms agent"
assert _normalize_search_query(" hello world ") == "hello world"


def test_normalize_strips_leading_trailing_whitespace():
assert _normalize_search_query("\tquery\n") == "query"


def test_normalize_empty_query_returns_empty():
assert _normalize_search_query("") == ""


def test_normalize_none_query_passthrough():
# None is falsy; helper returns it unchanged (guard: `if not query`)
assert _normalize_search_query(None) is None


def test_normalize_idempotent():
once = _normalize_search_query("OpenViking Agent")
twice = _normalize_search_query(once)
assert once == twice == "openviking agent"