특허·법률·논문 검색을 위한 다국어 검색 엔진 설계 - BM25부터 RRF·Cross-encoder까지

{
  "settings": {
    "analysis": {
      "analyzer": {
        "ko_nori": {
          "type": "custom",
          "tokenizer": "nori_tokenizer",
          "filter": ["nori_part_of_speech", "lowercase", "ko_synonyms"]
        },
        "ko_bigram": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": ["lowercase", "cjk_bigram"]
        },
        "en_domain": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": ["lowercase", "english_minimal_stem", "ipc_synonyms"]
        }
      },
      "filter": {
        "ko_synonyms": { "type": "synonym_graph", "synonyms_path": "synonyms/ko-patent.txt" },
        "ipc_synonyms": { "type": "synonym_graph", "synonyms_path": "synonyms/ipc.txt" }
      }
    }
  },
  "mappings": {
    "properties": {
      "application_no": { "type": "keyword" },
      "ipc":             { "type": "keyword" },
      "filing_date":     { "type": "date" },
      "applicant":       { "type": "keyword", "fields": { "text": { "type": "text", "analyzer": "ko_nori" } } },
      "title_ko":        { "type": "text", "analyzer": "ko_nori",
                           "fields": { "bigram": { "type": "text", "analyzer": "ko_bigram" } } },
      "title_en":        { "type": "text", "analyzer": "en_domain" },
      "claims_ko":       { "type": "text", "analyzer": "ko_nori" },
      "claims_en":       { "type": "text", "analyzer": "en_domain" },
      "abstract_ko":     { "type": "text", "analyzer": "ko_nori" },
      "abstract_en":     { "type": "text", "analyzer": "en_domain" }
    }
  }
}

{
  "query": {
    "multi_match": {
      "query": "무선 통신 채널 추정",
      "type": "best_fields",
      "fields": [
        "title_ko^4",
        "title_ko.bigram^1.5",
        "claims_ko^3",
        "abstract_ko^2",
        "title_en^2",
        "claims_en^1.5"
      ],
      "tie_breaker": 0.3
    }
  }
}

GLOSSARY_KO_EN = {
    "채널 추정": ["channel estimation"],
    "무선 통신": ["wireless communication", "radio communication"],
    "직교 주파수 분할 다중화": ["OFDM", "orthogonal frequency-division multiplexing"],
    "제3자":     ["third party"],
    "선행기술":   ["prior art"],
}
 
def expand_query(q_ko: str) -> dict:
    en_terms = []
    for ko_term, en_list in GLOSSARY_KO_EN.items():
        if ko_term in q_ko:
            en_terms.extend(en_list)
    return {"ko": q_ko, "en": " ".join(en_terms) or None}

def hyde_expand(query: str, llm) -> str:
    prompt = f"""Write a one-paragraph patent abstract that would directly answer:
"{query}"
Use technical vocabulary. Do not add disclaimers."""
    return llm.complete(prompt)
 
q_vec = embed(hyde_expand(user_query, llm))

RRF(d) = Σ  1 / (k + rank_r(d))
         r ∈ R(d)

from collections import defaultdict
from typing import Iterable
 
def rrf_fuse(
    rankings: list[list[tuple[str, str]]],  # list of [(doc_id, chunk_id), ...]
    k: int = 60,
    weights: list[float] | None = None,
) -> list[tuple[str, float, list[str]]]:
    """문장 청크 순위 → 문헌 단위 RRF 점수.
    rankings 각 리스트는 rank 순서로 정렬되어 있다고 가정한다.
    """
    weights = weights or [1.0] * len(rankings)
    doc_score: dict[str, float] = defaultdict(float)
    doc_chunks: dict[str, list[str]] = defaultdict(list)
 
    for ranking, w in zip(rankings, weights):
        seen_in_run: set[str] = set()
        for rank, (doc_id, chunk_id) in enumerate(ranking, start=1):
            # 한 런(run) 안에서 같은 문헌의 두 번째 이후 청크는
            # 약하게만 더해주고 싶다면 여기서 decay를 줄 수도 있다.
            doc_score[doc_id] += w / (k + rank)
            if chunk_id not in doc_chunks[doc_id]:
                doc_chunks[doc_id].append(chunk_id)
 
    fused = sorted(
        ((d, s, doc_chunks[d]) for d, s in doc_score.items()),
        key=lambda x: x[1],
        reverse=True,
    )
    return fused

from sentence_transformers import CrossEncoder
 
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
 
def rerank(query: str, candidates: list[dict], top_n: int = 20) -> list[dict]:
    # candidates: [{"doc_id": ..., "best_chunk_text": ...}, ...]
    pairs = [(query, c["best_chunk_text"]) for c in candidates]
    scores = reranker.predict(pairs, batch_size=32)
    for c, s in zip(candidates, scores):
        c["rerank_score"] = float(s)
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:top_n]

import math
 
def ndcg_at_k(gains: list[int], k: int) -> float:
    dcg = sum((2**g - 1) / math.log2(i + 2) for i, g in enumerate(gains[:k]))
    ideal = sorted(gains, reverse=True)[:k]
    idcg = sum((2**g - 1) / math.log2(i + 2) for i, g in enumerate(ideal))
    return dcg / idcg if idcg > 0 else 0.0
 
def recall_at_k(retrieved_ids: list[str], gold_ids: set[str], k: int) -> float:
    if not gold_ids:
        return 0.0
    return len(set(retrieved_ids[:k]) & gold_ids) / len(gold_ids)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
 
qc = QdrantClient(url="http://qdrant:6333")
 
qc.recreate_collection(
    collection_name="patent_chunks",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)
 
points = [
    PointStruct(
        id=f"{doc['application_no']}#{chunk['idx']}",
        vector=embed(chunk["text"]),
        payload={
            "application_no": doc["application_no"],
            "lang":           chunk["lang"],
            "field":          chunk["field"],     # "claim" | "abstract" | "spec"
            "ipc":            doc["ipc"],
            "filing_date":    doc["filing_date"],
            "text":           chunk["text"],
        },
    )
    for doc, chunk in iter_chunks()
]
qc.upsert(collection_name="patent_chunks", points=points)

from opensearchpy import OpenSearch
 
os_client = OpenSearch(hosts=["https://opensearch:9200"])
 
def bm25_search(q_ko: str, q_en: str | None, size: int = 100):
    fields = ["title_ko^4", "claims_ko^3", "abstract_ko^2"]
    if q_en:
        fields += ["title_en^3", "claims_en^2", "abstract_en^1.5"]
    body = {
        "size": size,
        "_source": ["application_no", "field_id"],
        "query": {
            "multi_match": {
                "query": q_ko if not q_en else f"{q_ko} {q_en}",
                "type": "best_fields",
                "fields": fields,
                "tie_breaker": 0.3,
            }
        },
    }
    res = os_client.search(index="patent_chunks", body=body)
    return [(h["_source"]["application_no"], h["_source"]["field_id"])
            for h in res["hits"]["hits"]]

def dense_search(q_vec, size: int = 100, ipc_filter: list[str] | None = None):
    flt = {"must": [{"key": "ipc", "match": {"any": ipc_filter}}]} if ipc_filter else None
    res = qc.search(
        collection_name="patent_chunks",
        query_vector=q_vec,
        limit=size,
        query_filter=flt,
    )
    return [(p.payload["application_no"], p.id) for p in res]
 
def hybrid_search(query: str, llm, embedder, ipc_filter=None, top_n=20):
    expanded = expand_query(query)                 # 4.1
    bm25_a = bm25_search(expanded["ko"], None)
    bm25_b = bm25_search(expanded["ko"], expanded["en"])
    dense  = dense_search(embedder.encode(hyde_expand(query, llm)), ipc_filter=ipc_filter)
    fused  = rrf_fuse([bm25_a, bm25_b, dense], k=60, weights=[1.0, 0.6, 1.2])
    candidates = build_candidates(fused[:top_n * 3])  # best chunk per doc
    return rerank(query, candidates, top_n=top_n)

from pymilvus import MilvusClient, DataType
 
mc = MilvusClient(uri="http://milvus:19530")
 
schema = mc.create_schema(auto_id=False, enable_dynamic_field=False)
schema.add_field("id",             DataType.VARCHAR, is_primary=True, max_length=128)
schema.add_field("application_no", DataType.VARCHAR, max_length=32)
schema.add_field("lang",           DataType.VARCHAR, max_length=4)
schema.add_field("field",          DataType.VARCHAR, max_length=16)
schema.add_field("ipc",            DataType.VARCHAR, max_length=16)
schema.add_field("text",           DataType.VARCHAR, max_length=2048)
schema.add_field("vector",         DataType.FLOAT_VECTOR, dim=1024)
 
mc.create_collection(collection_name="patent_chunks", schema=schema)
mc.create_index(
    collection_name="patent_chunks",
    index_params=[{"field_name": "vector", "index_type": "HNSW",
                   "metric_type": "COSINE", "params": {"M": 16, "efConstruction": 200}}],
)

from elasticsearch import Elasticsearch
 
es = Elasticsearch("https://es:9200")
 
def bm25_search_es(q_ko, q_en, size=100):
    body = {
        "size": size,
        "_source": ["application_no", "chunk_id"],
        "query": {
            "multi_match": {
                "query": f"{q_ko} {q_en or ''}".strip(),
                "fields": ["title_ko^4", "claims_ko^3", "abstract_ko^2",
                           "title_en^3", "claims_en^2"],
                "tie_breaker": 0.3,
            }
        },
    }
    res = es.search(index="patent_chunks", body=body)
    return [(h["_source"]["application_no"], h["_source"]["chunk_id"])
            for h in res["hits"]["hits"]]
 
def dense_search_milvus(q_vec, size=100):
    res = mc.search(
        collection_name="patent_chunks",
        data=[q_vec],
        limit=size,
        output_fields=["application_no"],
        search_params={"metric_type": "COSINE", "params": {"ef": 128}},
    )[0]
    return [(hit["entity"]["application_no"], hit["id"]) for hit in res]