実際にやりたかったこと
RAG や社内文書検索を作っていると、ベクトル検索だけでは拾いにくいものがあります。
たとえば、規程名、品番、エラーコード、部署名、固有名詞、申請フロー名のようなものです。
こういう「文字として一致してほしい検索」は、keyword 検索が強いです。
Oracle Database を使うなら、Oracle Text の CONTAINS を使うと素直に実装できます。
今回は、Oracle Database 側で keyword 検索をどう実装したかを、実コードと実行結果ベースでまとめます。
今回出すコードの流れ
先に、コードのつながりを書いておきます。
ORACLE_TEXT_STOP_WORDS
↓
Oracle Text stoplist を作る DDL
↓
CREATE INDEX ... INDEXTYPE IS CTXSYS.CONTEXT
ユーザー入力
↓
oracle_text_terms()
↓
_oracle_text_query()
↓
CONTAINS(c.chunk_text, :query, 1)
filters
↓
_oracle_retrieval_where()
↓
WHERE {where_sql}
_keyword_search_with_oracle()
↓
keyword_rank / keyword_score を付けて返す
この記事では keyword 検索に関係する部分だけを取り出します。
1. 検索対象は chunk 単位にする
文書全体を 1 行で検索するより、chunk 単位にしたほうが扱いやすいです。
keyword 検索に必要な列だけ抜き出すと、こういう形です。
CREATE TABLE rag_chunks (
chunk_id VARCHAR2(128) PRIMARY KEY,
document_id VARCHAR2(64) NOT NULL,
tenant_id_hash CHAR(64),
chunk_index NUMBER NOT NULL,
chunk_text CLOB NOT NULL,
metadata_json JSON,
created_at TIMESTAMP DEFAULT SYSTIMESTAMP
);
keyword 検索の対象は chunk_text です。
document_id は文書単位の絞り込みに使います。
tenant_id_hash は tenant scope に使います。
chunk_index は文書内の順序を安定させるために使います。
metadata_json は章タイトルや chunk 種別で絞り込むために使います。
次に、この chunk_text に Oracle Text index を作ります。
2. Oracle Text index を作る
Oracle Text を使うには、検索対象の列に CTXSYS.CONTEXT index を作ります。
実装では、lexer と stoplist も明示しています。
ORACLE_TEXT_LEXER_PREFERENCE = "RAG_TEXT_WORLD_LEXER"
ORACLE_TEXT_STOPLIST = "RAG_TEXT_STOPLIST"
ORACLE_TEXT_LEXER = "WORLD_LEXER"
lexer は、文章を検索用 token に分解するための設定です。
今回は、日本語と英語が混ざる文書を扱うために WORLD_LEXER を使っています。
Oracle Database の keyword search 設定
クラウド利用料 approval flow
stopword は 1 か所にまとめる
stopword は Python 側で 1 か所にまとめています。
ORACLE_TEXT_STOP_WORDS = (
"の", "は", "が", "を", "に", "へ", "で", "と", "も", "か",
"です", "ます", "なん", "んで",
)
この定義は 2 か所で使います。
1つ目は、Oracle Text stoplist を作る DDL です。
stopword_calls = "\n".join(
f" add_stopword('{word}');" for word in ORACLE_TEXT_STOP_WORDS
)
これで、CTX_DDL.ADD_STOPWORD(...) を含む SQL を生成します。
記事では展開後の add_stopword('の')... は省略します。
2つ目は、Python 側の query 正規化です。
JAPANESE_QUERY_STOP_TERMS = set(ORACLE_TEXT_STOP_WORDS)
つまり、DB 側の stoplist と Python 側の query filter は同じ定義を使います。
stoplist と lexer を作る
DDL 生成の考え方はこうです。
DECLARE
v_count NUMBER;
BEGIN
SELECT COUNT(*) INTO v_count
FROM ctx_user_preferences
WHERE pre_name = 'RAG_TEXT_WORLD_LEXER'
AND pre_class = 'LEXER';
IF v_count = 0 THEN
CTX_DDL.CREATE_PREFERENCE('RAG_TEXT_WORLD_LEXER', 'WORLD_LEXER');
END IF;
END;
/
SELECT COUNT(*) してから作っているのは、DDL を再実行しても失敗しないようにするためです。
stopword の追加も、同じ考え方です。
PROCEDURE add_stopword(p_word VARCHAR2) IS
BEGIN
CTX_DDL.ADD_STOPWORD('RAG_TEXT_STOPLIST', p_word);
EXCEPTION
WHEN OTHERS THEN
NULL;
END;
index を作る
最後に chunk_text に Oracle Text index を作ります。
CREATE INDEX rag_chunks_text_idx
ON rag_chunks (chunk_text)
INDEXTYPE IS CTXSYS.CONTEXT
PARAMETERS ('LEXER RAG_TEXT_WORLD_LEXER STOPLIST RAG_TEXT_STOPLIST SYNC (ON COMMIT)');
ここで、さきほど作った lexer と stoplist を使います。
この index を作ると、次の検索が使えるようになります。
CONTAINS(c.chunk_text, :query, 1) > 0
SYNC (ON COMMIT) は、commit 時に Oracle Text index を同期する指定です。
文書を登録して commit したあと、すぐ検索できるようにしています。
3. 自然文 query を Oracle Text query に変換する
ユーザー入力をそのまま CONTAINS に渡すのは避けます。
Oracle Text には ABOUT、ACCUM、NEAR、OR などの query syntax があるためです。
流れはこうです。
ユーザー入力
↓
oracle_text_terms()
↓
_oracle_text_query()
↓
CONTAINS(c.chunk_text, :query, 1)
term 抽出で使う定義です。
import re
TOKEN_PATTERN = re.compile(r"[a-z0-9_]+|[ぁ-んァ-ン一-龯々ー]+", re.IGNORECASE)
ASCII_TOKEN_PATTERN = re.compile(r"^[a-z0-9_]+$", re.IGNORECASE)
KANJI_RUN_PATTERN = re.compile(r"[一-龯々]+")
KATAKANA_RUN_PATTERN = re.compile(r"[ァ-ンー]+")
ORACLE_TEXT_MAX_TERMS = 12
ENGLISH_QUERY_STOP_TERMS = {
"a", "an", "and", "are", "at", "be", "by", "can", "could",
"did", "do", "does", "for", "from", "how", "in", "is",
"it", "near", "of", "on", "or", "please", "should", "the",
"to", "was", "were", "what", "when", "where", "which",
"who", "why", "with", "would",
}
ORACLE_TEXT_OPERATOR_TERMS = {
"about", "accum", "and", "equiv", "fuzzy", "haspath",
"inpath", "minus", "near", "not", "or", "soundex",
"stem", "within",
}
JAPANESE_QUERY_STOP_TERMS = set(ORACLE_TEXT_STOP_WORDS)
DB 側の stoplist と query 側の stop terms は同じ定義から作っています。
JAPANESE_QUERY_STOP_TERMS = set(ORACLE_TEXT_STOP_WORDS)
こうしておくと、DB index 側と query 正規化側の考え方がズレにくくなります。
term 抽出
def oracle_text_terms(query: str) -> list[str]:
terms: list[str] = []
for match in TOKEN_PATTERN.finditer(query):
raw_token = match.group(0).strip()
token = raw_token.casefold()
if len(token) < 2:
continue
if ASCII_TOKEN_PATTERN.fullmatch(raw_token):
term = _english_query_term(raw_token, token)
if term is not None:
terms.append(term)
continue
terms.extend(_japanese_query_terms(token))
return _unique_optional_sequence(terms)[:ORACLE_TEXT_MAX_TERMS]
この関数は、自然文から検索に使う term だけを取り出します。
英数字なら _english_query_term() に渡します。
日本語なら _japanese_query_terms() に渡します。
最後に重複を消して、最大 12 term に制限します。
英語 term
def _english_query_term(raw_token: str, token: str) -> str | None:
if token in ORACLE_TEXT_OPERATOR_TERMS:
return None
if token in ENGLISH_QUERY_STOP_TERMS and not raw_token.isupper():
return None
return token
what、is、the のような検索意図が弱い語は落とします。
また、Oracle Text operator として解釈される可能性がある near や or も落とします。
一方で、IT や HR のような大文字略語は残します。
業務文書では部署名や略語が重要な keyword になるためです。
日本語 term
def _japanese_query_terms(token: str) -> list[str]:
terms: list[str] = []
for run in KANJI_RUN_PATTERN.findall(token):
if len(run) < 2 or run in JAPANESE_QUERY_STOP_TERMS:
continue
if len(run) <= 3:
terms.append(run)
terms.extend(_kanji_compound_terms(run))
for run in KATAKANA_RUN_PATTERN.findall(token):
if len(run) >= 2 and run not in JAPANESE_QUERY_STOP_TERMS:
terms.append(run)
return terms
日本語側では、漢字 run とカタカナ run を拾います。
たとえば、次の query です。
社内規程の申請フローは?
term はこうなります。
["社内", "規程", "申請", "フロー"]
の、は、? は検索語にしません。
長い漢字語は軽く分解します。
def _kanji_compound_terms(value: str) -> list[str]:
if len(value) == 3:
return [value[:2]]
if len(value) >= 4:
return [value[:2], value[-2:]]
return []
これは形態素解析ではありません。
まずは軽いルールで、Oracle Text に渡す検索語を安全に短くしています。
Oracle Text query
最後に、term の list を Oracle Text query にします。
def _oracle_text_query(query: str) -> str | None:
unique_terms = oracle_text_terms(query)
if not unique_terms:
return None
return " ACCUM ".join(f"{{{term}}}" for term in unique_terms)
たとえば、次の入力です。
社内規程の申請フローは?
Oracle Text query はこうなります。
{社内} ACCUM {規程} ACCUM {申請} ACCUM {フロー}
この文字列を、後で CONTAINS(c.chunk_text, :query, 1) に bind します。
ACCUM を使うことで、複数 keyword の一致を score に反映できます。
検索語が 1 つも残らない場合は None を返します。
この場合、DB には検索を投げません。
4. CONTAINS は大文字・小文字を区別するのか
今回の WORLD_LEXER 設定では、英字の大文字・小文字は区別されませんでした。
検証データです。
Oracle Keyword Search
小文字で検索しても、大文字で検索しても hit しました。
{"case": "english_lower_query", "query": "{oracle}", "rows": [{"id": 1, "score": 0.05, "body": "Oracle Keyword Search"}]}
{"case": "english_upper_query", "query": "{ORACLE}", "rows": [{"id": 1, "score": 0.05, "body": "Oracle Keyword Search"}]}
ただし、アプリ側では query term を casefold() して、小文字に寄せて Oracle Text に渡しています。
5. CONTAINS は半角・全角カタカナを区別するのか
今回の WORLD_LEXER 設定では、半角カタカナと全角カタカナは同一視されました。
検証データです。
カタカナ キーワード
カタカナ キーワード
全角カタカナで検索しても、半角カタカナで検索しても、両方に hit しました。
{"case": "katakana_full_query", "query": "{カタカナ}", "rows": [{"id": 2, "score": 0.04, "body": "カタカナ キーワード"}, {"id": 3, "score": 0.04, "body": "カタカナ キーワード"}]}
{"case": "katakana_half_query", "query": "{カタカナ}", "rows": [{"id": 2, "score": 0.04, "body": "カタカナ キーワード"}, {"id": 3, "score": 0.04, "body": "カタカナ キーワード"}]}
Oracle Text index 側では、この揺れは吸収されました。
ただし、アプリ側の term 抽出は全角カタカナを対象にしています。
KATAKANA_RUN_PATTERN = re.compile(r"[ァ-ンー]+")
半角カタカナの query を確実に拾いたいなら、アプリ側で先に正規化したほうがよいです。
import unicodedata
query = unicodedata.normalize("NFKC", query)
6. CONTAINS はカタカナ・ひらがなを区別するのか
今回の WORLD_LEXER 設定では、カタカナとひらがなは同一視されませんでした。
検証データです。
カタカナ キーワード
かたかな きーわーど
{カタカナ} はカタカナの行に hit しました。
{"case": "katakana_full_query", "query": "{カタカナ}", "rows": [{"id": 2, "score": 0.04, "body": "カタカナ キーワード"}, {"id": 3, "score": 0.04, "body": "カタカナ キーワード"}]}
{かたかな} はひらがなの行に hit しました。
{"case": "hiragana_query", "query": "{かたかな}", "rows": [{"id": 4, "score": 0.05, "body": "かたかな きーわーど"}]}
つまり、次は別物として扱われました。
カタカナ
かたかな
もし「かたかな」でも「カタカナ」に hit させたいなら、アプリ側でひらがなをカタカナへ寄せる正規化を追加するのがよいです。
7. 検索 SQL
ここまでで作った _oracle_text_query() の結果を、ここで :query に bind します。
keyword 検索の SQL は次です。
SELECT *
FROM (
SELECT
c.document_id,
c.chunk_id,
c.chunk_text,
c.metadata_json,
c.chunk_index,
d.file_name,
d.category_name,
SCORE(1) / 100 AS score
FROM rag_chunks c
JOIN rag_documents d ON d.document_id = c.document_id
WHERE d.status = 'INDEXED'
AND CONTAINS(c.chunk_text, :query, 1) > 0
ORDER BY
SCORE(1) DESC,
c.document_id ASC,
c.chunk_index ASC,
c.chunk_id ASC
)
WHERE ROWNUM <= :top_k
CONTAINS(c.chunk_text, :query, 1) が Oracle Text の検索です。
1 は label です。
この label は SCORE(1) と対応します。
SCORE(1) を使うことで、keyword relevance の高い順に並べられます。
同点時は document_id、chunk_index、chunk_id で安定ソートします。
検索評価や regression test では、結果順が揺れないことが大事です。
8. 絞り込み条件は SQL 側に入れる
実運用では、keyword が一致しただけで返してはいけません。
たとえば、文書ステータス、tenant scope、document filter、chunk 種別などで絞る必要があります。
そのため、検索 SQL の WHERE は filter から組み立てます。
where_sql, binds = _oracle_retrieval_where(filters)
この where_sql が、検索 SQL の {where_sql} に入ります。
WHERE {where_sql}
AND CONTAINS(c.chunk_text, :query, 1) > 0
例です。
d.document_id = :filter_document_id
LOWER(JSON_VALUE(c.metadata_json, '$.content_kind')) = :filter_content_kind
LOWER(JSON_VALUE(c.metadata_json, '$.section_title')) LIKE :filter_section_title ESCAPE '\'
keyword 検索と業務 filter は、同じ SQL の中で一緒に効かせます。
アプリ側で後から絞るより、DB 側で絞ったほうが安全で、返却件数も安定します。
9. なぜ LIKE ではなく CONTAINS なのか
LIKE でも検索はできます。
WHERE chunk_text LIKE '%クラウド利用料%'
ただし、業務文書検索ではすぐに限界が出ます。
- 長文
CLOBに対して重くなりやすい - relevance score がない
- 複数 keyword の ranking が弱い
- 日本語・英語混在の全文検索として扱いにくい
- index 設計や検索品質の調整がしにくい
Oracle Text の CONTAINS なら、全文検索 index を使いつつ、SCORE で順位付けできます。
WHERE CONTAINS(chunk_text, :query, 1) > 0
ORDER BY SCORE(1) DESC
keyword 検索を業務システムに入れるなら、LIKE より CONTAINS を使うほうが素直でした。
10. Python から呼ぶコード
ここまでの処理をまとめて呼ぶ入口が keyword_search() です。
async def keyword_search(
self,
query: str,
top_k: int,
filters: dict[str, str] | None = None,
) -> list[RetrievedChunk]:
"""Oracle Text 相当のキーワード検索を行う。"""
return await self._keyword_search_with_oracle(query, top_k, filters or {})
実際の検索処理です。
async def _keyword_search_with_oracle(
self, query: str, top_k: int, filters: dict[str, str]
) -> list[RetrievedChunk]:
"""Oracle Text で keyword chunk を取得する。"""
text_query = _oracle_text_query(query)
if text_query is None:
return []
where_sql, binds = _oracle_retrieval_where(filters)
binds.update({"query": text_query, "top_k": top_k})
rows = await self._fetch_all(
_render_sql(
"""
SELECT *
FROM (
SELECT
c.document_id,
c.chunk_id,
c.chunk_text,
c.metadata_json,
c.chunk_index,
d.file_name,
d.category_name,
SCORE(1) / 100 AS score
FROM rag_chunks c
JOIN rag_documents d ON d.document_id = c.document_id
WHERE {where_sql}
AND CONTAINS(c.chunk_text, :query, 1) > 0
ORDER BY
SCORE(1) DESC,
c.document_id ASC,
c.chunk_index ASC,
c.chunk_id ASC
)
WHERE ROWNUM <= :top_k
""",
where_sql=where_sql,
),
binds,
)
return [
_with_retrieval_metadata(
_retrieved_chunk_from_row(row),
retrieval_mode="keyword",
keyword_rank=rank,
keyword_score=round(_float_value(row.get("score", 0.0)), 6),
)
for rank, row in enumerate(rows, start=1)
]
この関数内で、前に出した部品がつながります。
query
↓
_oracle_text_query(query)
↓
binds["query"]
↓
CONTAINS(c.chunk_text, :query, 1)
filter もここで SQL に入ります。
filters
↓
_oracle_retrieval_where(filters)
↓
WHERE {where_sql}
検索結果には、keyword_rank と keyword_score を metadata として残します。
retrieval_mode="keyword",
keyword_rank=rank,
keyword_score=round(_float_value(row.get("score", 0.0)), 6),
これは後で検索品質を確認するときに効きます。
11. 実行テスト
実DBに一時文書と一時 chunk を登録し、実装の OracleClient.keyword_search() を呼びました。
最後に作成した文書を削除し、再検索で 0 件になることも確認しています。
検証コードの検索部分です。
for label, query in [
("ja_keyword", "社内規程の申請フローは?"),
("operator_escape", 'policy - NEAR(oracle?) OR "x"'),
("en_keyword", "oracle keyword score"),
("empty_query", "の んで です what is the"),
]:
hits = await client.keyword_search(
query,
top_k=5,
filters={"document_id": document_id},
)
print(
{
"case": label,
"input": query,
"terms": oracle_text_terms(query),
"oracle_text_query": _oracle_text_query(query),
"hits": [
{
"chunk_id": hit.chunk_id,
"chunk_index": hit.metadata.get("chunk_index"),
"keyword_rank": hit.metadata.get("keyword_rank"),
"keyword_score": hit.metadata.get("keyword_score"),
"text": hit.text,
}
for hit in hits
],
}
)
実行結果です。
{"setup": "ok", "document_id": "fa7f6182b7904c28a2e525ea8fa577a6", "marker": "KW-DEMO-09eeb948"}
{"case": "ja_keyword", "input": "社内規程の申請フローは?", "terms": ["社内", "規程", "申請", "フロー"], "oracle_text_query": "{社内} ACCUM {規程} ACCUM {申請} ACCUM {フロー}", "hits": [{"chunk_id": "fa7f6182b7904c28a2e525ea8fa577a6:0", "chunk_index": 0, "keyword_rank": 1, "keyword_score": 0.77, "text": "KW-DEMO-09eeb948 社内規程では、クラウド利用料の申請フローは部門長承認を必要とします。"}]}
{"case": "operator_escape", "input": "policy - NEAR(oracle?) OR \"x\"", "terms": ["policy", "oracle"], "oracle_text_query": "{policy} ACCUM {oracle}", "hits": [{"chunk_id": "fa7f6182b7904c28a2e525ea8fa577a6:2", "chunk_index": 2, "keyword_rank": 1, "keyword_score": 0.04, "text": "KW-DEMO-09eeb948 Oracle Database keyword search uses CONTAINS and SCORE."}]}
{"case": "en_keyword", "input": "oracle keyword score", "terms": ["oracle", "keyword", "score"], "oracle_text_query": "{oracle} ACCUM {keyword} ACCUM {score}", "hits": [{"chunk_id": "fa7f6182b7904c28a2e525ea8fa577a6:2", "chunk_index": 2, "keyword_rank": 1, "keyword_score": 0.7, "text": "KW-DEMO-09eeb948 Oracle Database keyword search uses CONTAINS and SCORE."}]}
{"case": "empty_query", "input": "の んで です what is the", "terms": [], "oracle_text_query": null, "hits": []}
{"cleanup": "ok", "document_id": "fa7f6182b7904c28a2e525ea8fa577a6", "remaining_hits": 0}
確認できたことです。
- 日本語自然文が
{社内} ACCUM {規程} ACCUM {申請} ACCUM {フロー}に変換される -
NEAR/ORは Oracle Text operator として使わず除外される -
SCOREがkeyword_scoreとして返る - 空 query は DB 検索せず
hits: []になる - 一時データは削除済みで
remaining_hits: 0
12. 実装してみて分かったこと
keyword 検索は、SQL だけ見ると簡単です。
でも、実運用では周辺の設計が効きます。
入れておいたほうがよいものは次です。
- query 正規化を 1 か所に集約する
- DB stoplist と query stop terms は同じ定義から作る
- ユーザー入力を Oracle Text query として直接渡さない
- 空 query は DB に投げない
- tenant / document / status filter を SQL 側に入れる
-
SCOREだけでなく同点時の安定ソートを入れる - 検索結果に
keyword_rankとkeyword_scoreを残す - query 本文を監査ログにそのまま保存しない
- lexer / stoplist / index の設定を migration で確認できるようにする
- 半角/全角カナやひらがな/カタカナの要件は実DBで確認する
特に、query 正規化と安定ソートは地味ですが重要です。
検索品質評価や regression test を回すときに、結果順が揺れないことはかなり助かります。
まとめ
Oracle Database で keyword 検索を作るなら、基本はこれです。
CONTAINS(chunk_text, :query, 1) > 0
ORDER BY SCORE(1) DESC
ただし、これだけでは運用しにくいです。
実際には、次のセットで考えるのがよいです。
-
CTXSYS.CONTEXTindex WORLD_LEXER- stoplist
- 自然文 query の正規化
CONTAINSSCORE- 業務 filter
- 安定ソート
- rank / score metadata
- 文字種感度の実DB確認
keyword 検索は、ベクトル検索より地味です。
でも、規程名、コード、部署名、固有名詞のような「文字として一致してほしい検索」ではかなり強いです。
Oracle Database を使っているなら、まずは Oracle Text の CONTAINS を素直に使うのがよいと思います。