構成
- SQLite : FTS5(全文検索)、sqlite_vec(ベクトル検索)
- 融合 : weighted(正規化スコアの加重和)
- リランク : hotchpotch/japanese-reranker-cross-encoder-small-v1
- 形態素解析 : sudachi.rs、同義語辞書で代表語へ正規化
- 埋込みmodel : hotchpotch/static-embedding-japanese (1024 次元)
python script
"""
hybrid検索system by SQLite, 加重和fusion, cross-encoder rerank
- FTS(bm25) と vector(cosine) で検索後、正規化スコアの加重和で統合
- 上位候補を日本語 cross-encoder で再スコアし最終順位を決定
構成:
- SQLite : FTS5(全文検索)、sqlite_vec(ベクトル検索)
- 融合 : weighted(正規化スコアの加重和)
- リランク : hotchpotch/japanese-reranker-cross-encoder-small-v1
- 形態素解析 : sudachi.rs、同義語辞書で代表語へ正規化
- 埋込みmodel : hotchpotch/static-embedding-japanese (1024 次元)
入力データ:
- Shift-JIS / DOS 改行 の Q&A レコードを対象
- 相談No を業務上の主キー
- hybird検索の対象は、「表題 + 相談内容」を連結
テーブル:
documents : QA レコード (id, 相談No ほか)
documents_fts : FTS5 仮想テーブル (rowid = documents.id, 分かち書き済みtext)
documents_vec : sqlite-vec 仮想テーブル (rowid = documents.id, embedding)
"""
from datetime import datetime
from pathlib import Path
from sudachipy import Dictionary, SplitMode
from urllib.parse import quote
import csv
import html
import jaconv
import json
import logging
import logging.config
import numpy as np
import os
import re
import sqlite3
import sqlite_vec
import sys
import traceback
import unicodedata
os.environ.setdefault("HF_HUB_OFFLINE", "1")
from sentence_transformers import SentenceTransformer
MODEL_NAME = "hotchpotch/static-embedding-japanese"
DB_PATH = "hybrid_search.db"
EMBED_DIM = 1024
SOUDAN_URL = ("http://～/genDB2/sodan_hnweb/dtl4sntk?c_ccode={}")
CSV_ENCODING = "cp932"
CSV_COL = {
"SOUDAN_NO": 0,
"UKETSUKE" : 1,
"HONSHA" : 7,
"TITLE" : 11,
"SOUDAN" : 12,
"TEITYPE" : 14,
"BUSHO": (18,20,22,24,26,28,30,32),
"KAITO" : 35,
"HM_OR_TU" : 50,
"KONKYO" : 54,
}
SUDACHI_DICT_DIR = Path(__file__).resolve().parent.parent / "SUDACHI" / "dict"
SUDACHI_SYSTEM_DIC = SUDACHI_DICT_DIR / "system_core.dic"
SUDACHI_USER_DIC = SUDACHI_DICT_DIR / "user.dic"
SUDACHI_SYNONYMS_JSON = SUDACHI_DICT_DIR / "synonyms.json"
SUDACHI_SPLIT_MODE = SplitMode.A
FTS_INCLUDE_POS = (("名詞", "固有名詞"), ("名詞", "普通名詞"), ("接尾辞",))
FTS_DF_MAX_RATIO = 0.30
HYBRID_SEARCH_1ST_CANDIDATES = 50
HYBRID_SEARCH_VEC_WEIGHT = 0.5
RERANKER_MODEL = "hotchpotch/japanese-reranker-cross-encoder-small-v1"
RERANK_ENABLE = True
RERANK_CANDIDATES = 30
RERANK_MIN_SCORE = 0.5
LOG_DIR = Path(__file__).resolve().parent / "log"
LOG_DIR.mkdir(exist_ok=True)
LOG_FILE = LOG_DIR / f"{Path(__file__).stem}_{datetime.now():%m%d}.log"
LOG_CONF = {
"version": 1,
"disable_existing_loggers": False,
"loggers": {
"mainLogger": {
"level": "DEBUG",
"handlers": ["fileHandler", "consoleHandler"],
"propagate": False },
},
"handlers": {
"fileHandler": {
"level": "DEBUG",
"formatter": "mainFormatter",
"class": "logging.handlers.RotatingFileHandler",
"filename": str(LOG_FILE),
"encoding": "utf-8",
"maxBytes": 1024 * 1024 * 10,
"backupCount": 30,
},
"consoleHandler": {
"level": "INFO",
"formatter": "consoleFormatter",
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
},
},
"formatters": {
"mainFormatter": {
"format": "%(asctime)s\t%(levelname)s\t%(filename)s"
"\tL%(lineno)d\t%(funcName)s\t%(message)s",
"datefmt": "%Y/%m/%d %H:%M:%S" },
"consoleFormatter": {
"format": "%(message)s" },
},
}
logging.config.dictConfig(LOG_CONF)
logger = logging.getLogger("mainLogger")
def main(argv=None):
argv = sys.argv[1:] if argv is None else argv
cmd = argv[0] if argv else None
csv_arg = argv[1] if len(argv) > 1 else None
def usage():
print("USAGE: %s <CMD> <CSV>" % (os.path.basename(__file__)) )
print(" CMD:makedb <CSV>: 指定CSV から %s 作成" % (DB_PATH))
print(" CMD:searchtest <CSV>: 指定CSV を検索し htmlに出力")
if cmd == "makedb":
if not csv_arg:
usage()
return
db = HybridDb()
try:
db.make_db(Path(csv_arg))
finally:
db.close()
elif cmd == "searchtest":
if not csv_arg:
usage()
return
if not Path(DB_PATH).exists():
logger.warning( "%s がありません", DB_PATH )
return
db = HybridDb()
try:
db.search_test(Path(csv_arg))
finally:
db.close()
else:
usage()
class Embedder:
"""static-embedding-japanese のラッパー """
def __init__(self, model_name: str = MODEL_NAME):
self.model = SentenceTransformer(model_name, device="cpu")
get_dim = getattr(self.model, "get_embedding_dimension", None) \
or self.model.get_sentence_embedding_dimension
self.dim = get_dim()
def encode(self, texts) -> np.ndarray:
single = isinstance(texts, str)
if single:
texts = [texts]
vecs = self.model.encode(
texts,
normalize_embeddings=True,
convert_to_numpy=True,
).astype(np.float32)
return vecs[0] if single else vecs
class Reranker:
"""日本語 cross-encoder リランカーのラッパー """
def __init__(self, model_name: str = RERANKER_MODEL):
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name, device="cpu", max_length=512)
def rank(self, query, docs):
"""docs=[(doc_id, text), ...] を (query, text) で採点し降順で返す。
返り値: [(doc_id, score), ...] score は高いほど関連 (sigmoid 出力)。
"""
if not docs:
return []
scores = self.model.predict([(query, text) for _, text in docs])
ranked = [(doc_id, float(s)) for (doc_id, _), s in zip(docs, scores)]
ranked.sort(key=lambda x: x[1], reverse=True)
return ranked
class HybridDb:
"""hybrid.db への接続・索引作成・検索をまとめたクラス """
def __init__(self, db_path: str = DB_PATH):
self.db_path = db_path
self.conn = self._connect(db_path)
self._tokenizer = None
self._embedder = None
self._synonyms = None
self._reranker = None
self._vocab_ready = False
@staticmethod
def _connect(db_path: str) -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.enable_load_extension(True)
sqlite_vec.load(conn)
conn.enable_load_extension(False)
return conn
@staticmethod
def _build_tokenizer():
"""SudachiPy のトークナイザ生成。system_core.dic + user.dicを使用 """
try:
from sudachipy import Config
users = [str(SUDACHI_USER_DIC)] if SUDACHI_USER_DIC.exists() else []
config = Config(system=str(SUDACHI_SYSTEM_DIC), user=users)
return Dictionary(config=config).create()
except Exception as e:
logger.warning("Sudachi 辞書の読込に失敗 (%s). ", e)
@property
def tokenizer(self):
if self._tokenizer is None:
self._tokenizer = self._build_tokenizer()
return self._tokenizer
@property
def embedder(self) -> Embedder:
if self._embedder is None:
logger.info("埋込みmodelをload中: %s", MODEL_NAME)
self._embedder = Embedder()
logger.debug("埋込み次元: %d", self._embedder.dim)
return self._embedder
@property
def synonyms(self) -> dict:
if self._synonyms is None:
self._synonyms = self._load_synonyms()
logger.info("同義語entry数: %d", len(self._synonyms))
return self._synonyms
@property
def reranker(self) -> Reranker:
if self._reranker is None:
logger.info("re-ranker load中: %s", RERANKER_MODEL)
self._reranker = Reranker()
return self._reranker
@staticmethod
def _load_synonyms() -> dict:
""" 同義語 JSON から {語: 代表語} 辞書を作成 """
if not SUDACHI_SYNONYMS_JSON.exists():
logger.warning("同義語辞書がない為、同義語正規化を無効化: %s",
SUDACHI_SYNONYMS_JSON)
return {}
with open(SUDACHI_SYNONYMS_JSON, encoding="utf-8") as f:
groups = json.load(f)
logger.debug("同義語辞書を読込: %s", SUDACHI_SYNONYMS_JSON)
mapping: dict[str, str] = {}
for words in groups.values():
words = [HybridDb.normalize_word(w) for w in words if w]
if not words:
continue
rep = words[0]
for w in words:
mapping.setdefault(w, rep)
return mapping
def close(self):
self.conn.close()
def count(self) -> int:
"""documents の登録件数 """
try:
return self.conn.execute(
"SELECT COUNT(*) FROM documents").fetchone()[0]
except sqlite3.OperationalError:
return 0
@staticmethod
def normalize_word(word: str) -> str:
"""半角/全角・大文字/小文字などの表記を統一 (形態素解析の前処理)。
- 空白(全角空白・改行含む)を除去
- NFKC 正規化 (全角英数→半角、半角カナ→全角 など)
- 英数字は半角へ、カナは全角へ寄せる
- 英字は大文字に統一、カンマは除去
"""
try:
if word is None:
return ""
word = re.sub(r"[\s\n ]+", "", word)
word = unicodedata.normalize("NFKC", word)
word = jaconv.z2h(word, kana=False, ascii=True, digit=True)
word = jaconv.h2z(word, kana=True, ascii=False, digit=False)
word = word.upper().replace(",", "")
except Exception as e:
logger.error("%s %s", e, word)
logger.error(traceback.format_exc())
return word
def morphemes(self, text: str):
"""text を Sudachi で形態素に分割 """
return self.tokenizer.tokenize(text, SUDACHI_SPLIT_MODE)
_SYNONYM_MAX_SPAN = 8
@staticmethod
def _expandable(surface: str) -> bool:
"""同義語展開してよいかの判定。
「なし」(無し⇔梨)のような同音異義の誤った正規化を防ぐため
"""
return any(not ("ぁ" <= ch <= "ゟ") for ch in surface)
def analyzed(self, text: str):
"""text を形態素解析し、[(代表語, 品詞タプル), ...] を返す
品詞タプルは Sudachi の part_of_speech() 6 項目
(大分類, 中分類, 小分類, 細分類, 活用型, 活用形)。
分割単位 A では複数形態素になる同義語 (例 不明確=不+明確) も拾えるよう
連続する形態素の表層形を連結して最長一致で同義語辞書を引く
"""
text = self.normalize_word(text)
ms = list(self.morphemes(text))
surfaces = [m.surface() for m in ms]
syn = self.synonyms
out = []
i, n = 0, len(ms)
while i < n:
hit = None
for j in range(min(n, i + self._SYNONYM_MAX_SPAN), i, -1):
key = "".join(surfaces[i:j])
rep = syn.get(key)
if rep is not None and self._expandable(key):
hit = (rep, tuple(ms[i].part_of_speech()), j)
break
if hit:
out.append((hit[0], hit[1]))
i = hit[2]
else:
out.append((surfaces[i], tuple(ms[i].part_of_speech())))
i += 1
return out
def tokenize(self, text: str) -> str:
"""代表語へ正規化した形態素を空白区切りで返す (FTS5 索引用)。"""
return " ".join(w for w, _pos in self.analyzed(text))
def normalize_text(self, text: str) -> str:
"""同義語を代表語へ置換した文字列を返す (埋め込み入力用)。"""
return "".join(w for w, _pos in self.analyzed(text)) or " "
def _ensure_vocab(self):
"""fts5vocab (documents_fts_vocab) を用意 """
if not self._vocab_ready:
self.conn.execute(
"CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts_vocab "
"USING fts5vocab('documents_fts', 'row')")
self._vocab_ready = True
def doc_freq(self, term: str) -> int:
"""term を含む文書数 (df)。未知語は 0。"""
self._ensure_vocab()
row = self.conn.execute(
"SELECT doc FROM documents_fts_vocab WHERE term = ?", (term,)
).fetchone()
return row[0] if row else 0
def build_fts_query(self, text: str) -> str:
"""検索クエリ文字列を FTS5 の MATCH 式に変換 (特徴語のみ)。
内容語(品詞が FTS_INCLUDE_POS のもの)を代表語へ正規化し、
文書頻度(df)が FTS_DF_MAX_RATIO を超える汎用語(例: 対応/可能/設置)を除外、
識別力のある「特徴語」だけを OR 連結
"""
def _included(pos):
return any(tuple(pos[:len(pat)]) == pat for pat in FTS_INCLUDE_POS)
seen, terms, pos_of = set(), [], {}
for w, pos in self.analyzed(text):
w = w.strip()
if w and _included(pos) and w not in seen:
seen.add(w)
terms.append(w)
pos_of[w] = pos
total = self.count() or 1
max_df = FTS_DF_MAX_RATIO * total
with_df = [(w, self.doc_freq(w)) for w in terms]
feature = [w for w, df in with_df if df <= max_df]
if not feature:
feature = [w for w, _ in sorted(with_df, key=lambda x: x[1])[:3]]
dropped = [w for w, df in with_df if w not in feature]
df_of = dict(with_df)
def _pos_str(w):
return "-".join(p for p in pos_of.get(w, ()) if p and p != "*")
def _fmt(words):
return [f"{w}({_pos_str(w)},df={df_of[w]})" for w in words]
logger.debug("build_fts_query: 特徴語=%s / 除外(汎用語)=%s",
_fmt(feature), _fmt(dropped))
quoted = ['"' + t.replace('"', '""') + '"' for t in feature]
return " OR ".join(quoted)
def create_schema(self):
self.conn.executescript(
f"""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY, -- FTS5/sqlite-vec 用の rowid
soudan_no TEXT UNIQUE NOT NULL, -- 相談No (業務上の主キー)
uketsuke TEXT, -- 初回受付日
honsha TEXT, -- 販社本社名
title TEXT, -- 表題
soudan TEXT, -- 相談内容
tei_type TEXT, -- 邸タイプ
busho TEXT, -- 回答部所1~8 (空白区切り)
kaito TEXT, -- 回答内容
hm_or_tu TEXT, --
konkyo TEXT -- 根拠となる基準
);
-- FTS5: 分かち書き済み検索テキストを索引。rowid で documents と対応
CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts USING fts5(
text,
tokenize = 'unicode61 remove_diacritics 2'
);
-- sqlite-vec: 1024 次元ベクトルを cosine 距離で近傍検索。
CREATE VIRTUAL TABLE IF NOT EXISTS documents_vec USING vec0(
embedding FLOAT[{EMBED_DIM}] distance_metric=cosine
);
-- fts5vocab: 語ごとの文書頻度(df)を参照し、特徴語抽出に使う。
CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts_vocab
USING fts5vocab('documents_fts', 'row');
"""
)
self.conn.commit()
self._vocab_ready = True
def load_qa_csv(self, path: Path):
"""CSV を読み込み、dict リストを返す
読み込み時に次の整形を行う:
- 相談内容・回答内容の空白行(連続改行)を削除
- 表題・相談内容に含まれる氏名(人名)を「※氏名※」に置換
"""
def cell(row, idx):
return row[idx].strip() if idx < len(row) else ""
docs = []
with open(path, encoding=CSV_ENCODING, newline="") as f:
reader = csv.reader(f)
next(reader, None)
for row in reader:
soudan_no = cell(row, CSV_COL["SOUDAN_NO"])
if not soudan_no:
continue
busho = " ".join(
cell(row, i) for i in CSV_COL["BUSHO"] if cell(row, i)
)
soudan = self._collapse_blank_lines(cell(row, CSV_COL["SOUDAN"]))
kaito = self._collapse_blank_lines(cell(row, CSV_COL["KAITO"]))
title = self._mask_names(cell(row, CSV_COL["TITLE"]))
soudan = self._mask_names(soudan)
docs.append({
"soudan_no": soudan_no,
"uketsuke" : cell(row, CSV_COL["UKETSUKE"]),
"honsha" : cell(row, CSV_COL["HONSHA"]),
"title" : title,
"soudan" : soudan,
"tei_type" : cell(row, CSV_COL["TEITYPE"]),
"busho" : busho,
"kaito" : kaito,
"hm_or_tu" : cell(row, CSV_COL["HM_OR_TU"]),
"konkyo" : cell(row, CSV_COL["KONKYO"]),
})
return docs
_NAME_POS = ("名詞", "固有名詞", "人名")
def _mask_names(self, text: str) -> str:
"""text 中の氏名(人名)形態素を「※氏名※」に置換
Sudachi で形態素解析し、品詞が 名詞-固有名詞-人名-* の形態素の
文字位置を集める。連続する人名(姓+名 等)は 1 つの「※氏名※」に
"""
if not text:
return text
spans = []
for m in self.morphemes(text):
pos = tuple(m.part_of_speech())
if pos[:3] == self._NAME_POS:
spans.append((m.begin(), m.end()))
if not spans:
return text
merged = [list(spans[0])]
for b, e in spans[1:]:
if b <= merged[-1][1]:
merged[-1][1] = max(merged[-1][1], e)
else:
merged.append([b, e])
out = text
for b, e in reversed(merged):
out = out[:b] + "※氏名※" + out[e:]
return out
@staticmethod
def search_text(doc: dict) -> str:
"""hybird検索の対象テキスト = 表題 + 相談内容。"""
parts = [doc["title"], doc["soudan"]]
text = "\n".join(p for p in parts if p)
return text or " "
def index_documents(self, docs, batch_size: int = 256):
"""QA レコードの dict リストを 3 tableへ登録
埋め込みは batch_size 件ずまとめて計算
"""
logger.info("索引作成を開始: %d 件 (batch=%d)", len(docs), batch_size)
cur = self.conn.cursor()
texts = [self.search_text(d) for d in docs]
total = len(docs)
for start in range(0, total, batch_size):
chunk = docs[start:start + batch_size]
chunk_texts = texts[start:start + batch_size]
analyzed = [self.analyzed(t) for t in chunk_texts]
fts_texts = [" ".join(w for w, _ in a) for a in analyzed]
emb_texts = ["".join(w for w, _ in a) or " " for a in analyzed]
embeddings = self.embedder.encode(emb_texts)
for doc, fts_text, emb in zip(chunk, fts_texts, embeddings):
cur.execute(
"""INSERT INTO documents
(soudan_no, uketsuke, honsha, title, soudan, tei_type,
busho, kaito, hm_or_tu, konkyo)
VALUES (?,?,?,?,?,?,?,?,?,?)""",
(doc["soudan_no"], doc["uketsuke"], doc["honsha"],
doc["title"], doc["soudan"], doc["tei_type"], doc["busho"],
doc["kaito"], doc["hm_or_tu"], doc["konkyo"]),
)
doc_id = cur.lastrowid
cur.execute(
"INSERT INTO documents_fts(rowid, text) VALUES (?,?)",
(doc_id, fts_text),
)
cur.execute(
"INSERT INTO documents_vec(rowid, embedding) VALUES (?,?)",
(doc_id, sqlite_vec.serialize_float32(emb)),
)
self.conn.commit()
logger.info("登録 %d/%d", min(start + batch_size, total), total)
def fts_search(self, query: str, top_k: int = 10):
"""全文検索。返り値: [(doc_id, bm25_score), ...] スコア昇順(良い順)。"""
match = self.build_fts_query(query)
logger.debug("fts_search: query=%r match=%r", query, match)
if not match:
return []
rows = self.conn.execute(
"""
SELECT rowid, bm25(documents_fts) AS score
FROM documents_fts
WHERE documents_fts MATCH ?
ORDER BY score
LIMIT ?
""",
(match, top_k),
).fetchall()
logger.debug("fts_search: %d 件ヒット", len(rows))
return [(r[0], r[1]) for r in rows]
def vector_search(self, query: str, top_k: int = 10):
"""vector近傍検索。返り値: [(doc_id, distance), ...] """
normalized = self.normalize_text(query)
q = self.embedder.encode(normalized)
rows = self.conn.execute(
"""
SELECT rowid, distance
FROM documents_vec
WHERE embedding MATCH ? AND k = ?
ORDER BY distance
""",
(sqlite_vec.serialize_float32(q), top_k),
).fetchall()
return [(r[0], r[1]) for r in rows]
def _fetch_rows(self, doc_ids, detail):
"""doc_id 順に documents の各列を取得、detail をマージし返す """
cols = ("soudan_no", "uketsuke", "honsha", "title", "soudan",
"tei_type", "busho", "kaito", "hm_or_tu", "konkyo")
results = []
for doc_id in doc_ids:
row = self.conn.execute(
f"SELECT {', '.join(cols)} FROM documents WHERE id = ?",
(doc_id,),
).fetchone()
results.append({
"id": doc_id,
**dict(zip(cols, row)),
**detail.get(doc_id, {}),
})
return results
@staticmethod
def _weighted_fuse(fts, vec, vec_weight, top_k):
"""FTS(bm25) と ベクトル(距離) のスコアを正規化して加重和で統合
bm25 は -bm25(大きいほど良い)、距離は 1-距離(=類似度) に直し、
各々 min-max 正規化 [0,1] してから
score = vec_weight*類似度 + (1-vec_weight)*bm25
で合成する。片方にしか出ない文書はその成分を 0 とする (OR 的)。
返り値: [(doc_id, fused_score), ...] 降順。
"""
fts_rel = {d: -s for d, s in fts}
vec_rel = {d: 1.0 - dist for d, dist in vec}
def norm(m):
if not m:
return {}
lo, hi = min(m.values()), max(m.values())
if hi <= lo:
return {k: 1.0 for k in m}
return {k: (v - lo) / (hi - lo) for k, v in m.items()}
fn, vn = norm(fts_rel), norm(vec_rel)
scores = {}
for d in set(fn) | set(vn):
scores[d] = \
vec_weight * vn.get(d, 0.0) + (1 - vec_weight) * fn.get(d, 0.0)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
def _doc_text(self, doc_id, limit: int = 1500) -> str:
"""re-ranker入力用テキスト (表題+相談内容, 生テキスト) """
row = self.conn.execute(
"SELECT title, soudan FROM documents WHERE id = ?", (doc_id,)
).fetchone()
return "\n".join(p for p in row if p)[:limit]
def hybrid_search(self, query: str, top_k: int = 5,
candidates: int = None, rerank: bool = None):
"""hybrid検索 (加重和 fusion + cross-encoder re-rank)。
FTS(bm25) と ベクトル(cosine) を独立検索し、正規化スコアの加重和で
統合する。candidates / rerank を省略 (None) した場合は global 変数
HYBRID_SEARCH_1ST_CANDIDATES / RERANK_ENABLE を使う
rerank=True の場合、加重和で集めた上位候補を日本語 cross-encoder で
再スコアし、最終順位を決める。
"""
if candidates is None:
candidates = HYBRID_SEARCH_1ST_CANDIDATES
if rerank is None:
rerank = RERANK_ENABLE
fusion_k = max(top_k, RERANK_CANDIDATES) if rerank else top_k
fts = self.fts_search(query, candidates)
vec = self.vector_search(query, candidates)
fts_rank = {d: r for r, (d, _) in enumerate(fts, start=1)}
vec_rank = {d: r for r, (d, _) in enumerate(vec, start=1)}
bm25_of = dict(fts)
dist_of = dict(vec)
fused = self._weighted_fuse(fts, vec, HYBRID_SEARCH_VEC_WEIGHT, fusion_k)
fused_of = dict(fused)
ordered_ids = [d for d, _ in fused]
detail = {
d: {
"distance": dist_of.get(d),
"bm25": bm25_of.get(d),
"fts_rank": fts_rank.get(d),
"vec_rank": vec_rank.get(d),
"fused": fused_of.get(d),
}
for d in ordered_ids
}
if rerank and ordered_ids:
pairs = [(d, self._doc_text(d)) for d in ordered_ids]
ranked = self.reranker.rank(query, pairs)
for rank, (d, s) in enumerate(ranked, start=1):
detail.setdefault(d, {}).update(rerank_score=s, rerank_rank=rank)
kept = [d for d, s in ranked if s >= RERANK_MIN_SCORE]
if len(kept) < len(ranked):
logger.info("rerank 足切り: %d/%d 件が閾値 %.2f 未満で除外",
len(ranked) - len(kept), len(ranked), RERANK_MIN_SCORE)
ordered_ids = kept[:top_k]
else:
ordered_ids = ordered_ids[:top_k]
return self._fetch_rows(ordered_ids, detail)
def make_db(self, csv_path: Path):
"""指定 CSV から hybrid.db を作成する。"""
self.create_schema()
n = self.count()
if n > 0:
logger.info("既に %d 件登録。%s を削除し再実行", n, self.db_path)
return
docs = self.load_qa_csv(csv_path)
logger.info("%s から %d 件を読み込み", Path(csv_path).name, len(docs))
self.index_documents(docs)
logger.info("完了: %s に %d 件を登録", self.db_path, len(docs))
@staticmethod
def _fmt(v, spec=""):
"""None は '-'、数値は書式付きで返す (表示用)。"""
return "-" if v is None else format(v, spec)
@staticmethod
def _collapse_blank_lines(text: str) -> str:
"""空白行(連続改行)を改行 1 個に集約し、前後の空白を除く。"""
if not text:
return ""
return re.sub(r"[ \t]*\r?\n[ \t\r\n]*", "\n", text).strip()
def print_results(self, heading, results):
print(f"\n=== {heading} ===")
for i, r in enumerate(results, 1):
if isinstance(r, dict):
metrics = (f"rerank={self._fmt(r.get('rerank_score'), '.4f')}, "
f"fused={self._fmt(r.get('fused'), '.4f')}, "
f"dist={self._fmt(r.get('distance'), '.4f')}, "
f"bm25={self._fmt(r.get('bm25'), '.4f')}, "
f"fts_rank={self._fmt(r.get('fts_rank'))}, "
f"vec_rank={self._fmt(r.get('vec_rank'))}")
print(f"{i}. [{r['soudan_no']}] {metrics}")
print("表題:", r.get("title") or "")
print("相談内容:\n", self._collapse_blank_lines(r.get("soudan")))
print()
else:
doc_id, score = r
row = self.conn.execute(
"SELECT soudan_no, title FROM documents WHERE id = ?",
(doc_id,),
).fetchone()
print(f"{i}. [{row[0]}] {row[1]} (score={score:.4f})")
_HTML_STYLE = """
body{color:#222;padding:0 1rem}
h1{font-size:1.4rem}
h2{font-size:1.1rem;border-bottom:2px solid #ccc;}
.query{background:#f2f6fc; border-left:4px solid #4477cc;
white-space:pre-wrap}
.hit{border:1px solid #ddd;}
.head{font-weight:bold}
.no{color:#4477cc}
.metrics{font-size:.8rem; color:#666; font-weight:normal}
.title{font-weight:bold;}
.label{display:inline-block; color:#fff;background:#88a;}
.body{white-space:pre-wrap; background:#fafafa;}
"""
@staticmethod
def _soudan_link(soudan_no: str) -> str:
"""相談番号を詳細画面 URL へのリンク付き HTML ([番号]) にする。"""
no = html.escape(soudan_no or "")
url = html.escape(SOUDAN_URL.format(quote(soudan_no or "")))
return (f'[<a href="{url}" target="_blank" rel="noopener">{no}</a>]')
def _result_html(self, results) -> str:
"""1 件分の検索結果リストを HTML 断片にする。"""
def field(label, text):
body = html.escape(self._collapse_blank_lines(text)).replace("\n", "<br>")
return (f'<div class="field"><span class="label">{label}</span>'
f'<div class="body">{body}</div></div>')
blocks = []
for i, r in enumerate(results, 1):
metrics = (f"rerank={self._fmt(r.get('rerank_score'), '.4f')}, "
f"fused={self._fmt(r.get('fused'), '.4f')}, "
f"dist={self._fmt(r.get('distance'), '.4f')}, "
f"bm25={self._fmt(r.get('bm25'), '.4f')}, "
f"fts_rank={self._fmt(r.get('fts_rank'))}, "
f"vec_rank={self._fmt(r.get('vec_rank'))}")
blocks.append(
f'<div class="hit">'
f'<div class="head">{i}. '
f'<span class="no">{self._soudan_link(r["soudan_no"])}</span> '
f'<span class="title">{html.escape(r.get("title") or "")}</span></div>'
f'<div class="metrics">{html.escape(metrics)}</div>'
f'{field("相談内容", r.get("soudan") or "")}'
f'{field("回答内容", r.get("kaito") or "")}'
f'</div>'
)
return "\n".join(blocks)
def write_html(self, path, sections, source=None):
"""sections=[(heading, query, results), ...] を 1 つの HTMLに出力。
source を渡すと <h1> と <title> に検索元の 相談番号・表題 を出す。
"""
parts = []
for heading, query, results in sections:
parts.append(
f'<div class="query">{html.escape(query.strip())}</div>')
parts.append(f"<div>{html.escape(heading)}</div>")
parts.append(self._result_html(results))
no, ttl = source.get("soudan_no") or "", source.get("title") or ""
title_txt = html.escape(f"[{no}] {ttl}".strip())
h1_html = f"{self._soudan_link(no)} {html.escape(ttl)}".strip() if no \
else html.escape(ttl)
doc = (
'<!doctype html>\n<html lang="ja"><head><meta charset="utf-8">'
f"<title>{title_txt}</title>"
f"<style>{self._HTML_STYLE}</style></head><body>"
f"<h1>{h1_html}</h1> "
+ "\n".join(parts) +
"</body></html>"
)
Path(path).write_text(doc, encoding="utf-8")
logger.info("HTML 出力: %s", path)
def search_test(self, csv_path: Path):
"""クエリ CSV の各行(表題+相談内容)で検索し、相談番号.html に出力 """
n = self.count()
if n == 0:
logger.warning( "%s が空", self.db_path)
return
cases = self.load_qa_csv(csv_path)
logger.info("%s から %d 件のクエリを読み込みました。",
Path(csv_path).name, len(cases))
out_dir = Path(__file__).resolve().parent / "html"
out_dir.mkdir(exist_ok=True)
heading = f"AI類似検索結果 (keyword + vector)"
for i, case in enumerate(cases):
query = self.search_text(case)
logger.info("\n【%d/%d】相談No %s",
i + 1,
len(cases),
case["soudan_no"])
results = self.hybrid_search(query, 20)
self.print_results(heading, results)
query_html = case.get("soudan") or ""
safe = re.sub(r'[\\/:*?"<>|]', "_", case["soudan_no"])
out = out_dir / f"{safe}.html"
self.write_html(out, [(heading, query_html, results)], source=case)
if __name__ == "__main__":
main()
