들어가기

성공적인 서비스를 개발하기 위해서는 서비스를 잘 기획/설계하고 기술적 기반을 안정적으로 만드는 것이 중요합니다. 그러나 대부분의 사업은 서비스가 동작해야 할 적절한 시점이 있고, 이해 관계자를 설득하기 위해서는 눈에 보이는 무언가가 존재하는 것이 앞으로 나아가기 위한 동력이 되기도 합니다. 그런 의미에서 구현의 복잡도를 낮추면서 빠른 개발과 실험을 반복 가능하게 하는 것은 안정적으로 잘 만드는 것만큼 중요합니다.

이 글은 약 1.1GB의 크기를 가진 텍스트를 얼마의 시간에 데이터를 입력할 수 있고 색인하여 검색 가능한 상태로 만들어 서비스를 개발하기까지 어느 정도의 시간이 걸리는지 설명합니다. 여기에서는 Full-Text Search에 대한 내용만 다루고 있으며, 자연어 검색 혹은 하이브리드 검색은 상품 검색에 자연어 검색을 적용하기에서 참고하실 수 있습니다.

데모에 사용된 텍스트 데이터는 법제처에서 제공하는 판례 데이터로 판례 검색 데모에서 확인하실 수 있습니다.

판례 검색 서비스 구현

구현 내용은 1. 데이터 수집 및 전처리, 2. 색인, 3. 데모 서비스 개발의 순서로 구성되어 있습니다.

1. 데이터 수집 및 전처리

판례 데이터는 법제처의 국가법령정보 공동활용 사이트에서 Open API를 신청하여 다운로드 받을 수 있습니다. 크게 목록과 본문을 질의하는 API로 분리되어 있고, 본문은 XML과 HTML 형식을 제공하고 있습니다. XML에 비해 HTML은 정보량이 조금 더 많지만 판례 내용을 구체적으로 분석하지 않고 전체 내용을 검색 대상으로 하기 때문에 HTML 태그 제거가 필요 없는 XML 데이터를 활용했습니다. 향후 법령정보 연결, 주문 분리 등을 하기 위해서 이 데이터를 고려할 필요도 있습니다.

판례 전체 목록을 다운로드 받고 목록에 표시된 XML 데이터를 모두 다운로드 받습니다. 다운로드된 XML 데이터 예시는 다음과 같습니다. 다운로드된 문서의 규모는 다음과 같습니다.

• 87,491건
• 1.1GB의 XML 텍스트 데이터

<PrecService>
<판례정보일련번호>64441</판례정보일련번호>
<사건명>
<![CDATA[ 임대차보증금등·지료등 ]]>
</사건명>
<사건번호>2006다62492,62508</사건번호>
<선고일자>20080925</선고일자>
<선고>선고</선고>
<법원명>대법원</법원명>
<법원종류코드>400201</법원종류코드>
<사건종류명>민사</사건종류명>
<사건종류코드>400101</사건종류코드>
<판결유형>판결</판결유형>
<판시사항>
<![CDATA[ [1] 계약의 법정 또는 약정 해지사유 발생시, 당사자가 경매신청 등 계약해지를 전제로 하는 행위 또는 기존 계약관계를 유지할 의사가 없음을 파악할 수 있는 행위를 하고 상대방도 그로 인하여 계약이 종료됨을 객관적으로 인식할 수 있었던 경우, 계약해지의 효과가 발생하는지 여부(적극)<br/>[2] 건물의 소유를 목적으로 하는 토지임대차에서 차임을 담보할 목적으로 그 건물에 대한 근저당권을 설정받은 임대인이 차임 연체를 이유로 근저당권을 실행하여 임의경매를 신청하였다면 이는 묵시적 임대차계약 해지의 의사표시라 볼 수 있으므로, 법원의 경매개시결정이 임차인에게 송달된 때에 위 임대차가 종료되었다고 본 사례<br/> ]]>
</판시사항>
<판결요지>
<![CDATA[ ]]>
</판결요지>
<참조조문>
<![CDATA[ [1] 민법 제105조, 제543조 / [2] 민법 제105조, 제543조, 제640조, 제641조<br/> ]]>
</참조조문>
<참조판례>
<![CDATA[ ]]>
</참조판례>
<판례내용>
<![CDATA[ 【원고(반소피고), 상고인 겸 피상고인】 <br/>【피고(반소원고), 피상고인 겸 상고인】 <br/>【원심판결】 청주지법 2006. 8. 14. 선고 2005나3466, 나3473(반소) 판결<br/>【주 문】<br/> 원심판결 중 임차보증금반환청구의 본소에 관한 부분과 철거 및 원상복구비용 청구의 반소에 관한 부분을 각 파기하고, 이 부분 사건을 청주지방법원 본원 합의부에 환송한다. 원고(반소피고) 및 피고(반소원고)의 나머지 상고를 모두 기각한다.<br/><br/>【이 유】 각 상고이유를 판단한다. ... 중략 ...<br/><br/>대법관 김영란(재판장) 이홍훈 안대희(주심) 양창수 ]]>
</판례내용>
</PrecService>

추가적인 정보를 추출하지 않는다면 데이터는 깨끗한 편으로 아래와 같은 작업의 전처리를 진행합니다.

• 필드명을 영문 변수명으로 변경
• 판례정보일련번호를 숫자 타입으로 변경
• 일부 필드의 <br> 태그를 개행으로 변경

import logging
import re
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import click
import pandas as pd
from lxml import etree

logger = logging.getLogger(__file__)

_COLUMN_NAME_MAPPING = {
    "판례정보일련번호": "doc_id",
    "사건명": "name",
    "사건번호": "number",
    "선고일자": "judgment_date",
    "선고": "judgment",
    "법원명": "court_name",
    "법원종류코드": "court_type_code",
    "사건종류명": "type_name",
    "사건종류코드": "type_code",
    "판결유형": "judgment_type",
    "판시사항": "holding_statement",
    "판결요지": "judgment_summary",
    "참조조문": "reference_provisions",
    "참조판례": "reference_cases",
    "판례내용": "content",
}
_TAG_PATTERN = re.compile(r"<br ?/>")
_MAX_WORKERS = 1
_CHUNK_SIZE = 10000


def _rename_fields(row: dict) -> dict:
    return {_COLUMN_NAME_MAPPING[k]: v for k, v in row.items()}


def _convert_tag(text: str) -> str:
    return _TAG_PATTERN.sub("\n", text)


def _to_int(text: str) -> int | None:
    if isinstance(text, str) and text.isnumeric():
        return int(text)
    return None


_PROCESSING_MAP = {
    "doc_id": _to_int,
    "holding_statement": _convert_tag,
    "judgment_summary": _convert_tag,
    "reference_provisions": _convert_tag,
    "reference_cases": _convert_tag,
    "content": _convert_tag,
}


def _convert_fields(row: dict[str, any]) -> dict[str, any]:
    return {
        k: _PROCESSING_MAP[k](v) if k in _PROCESSING_MAP else v
        for k, v in row.items()
    }


def _convert(file: Path) -> dict:
    logger.debug("file: %s", file.name)
    with open(file, encoding="utf-8") as f:
        text = f.read()
        root = etree.fromstring(text.encode())
        row = {x.tag: x.text.strip() if x.text else x.text for x in root}
        row = _rename_fields(row)
        row = _convert_fields(row)
    return row


def _save(data: list[dict], index: int, output_path: Path) -> None:
    file = output_path / f"docs_{index}.pk"
    df = pd.DataFrame(data)
    df.to_pickle(file, compression="gzip")
    logger.info("saved: %s", file.name)


@click.command()
@click.option("--data_path", type=Path, help="Data path", default="data/raw/docs")
def main(data_path):
    output_path = Path("data/docs")
    output_path.mkdir(exist_ok=True, parents=True)

    files = sorted(data_path.glob("*.xml"))
    with ThreadPoolExecutor(_MAX_WORKERS) as pool:
        futures = []
        index = 0
        for file in files:
            futures.append(pool.submit(_convert, file))

        chunk = []
        for future in as_completed(futures):
            row = future.result()
            chunk.append(row)

            if len(chunk) >= _CHUNK_SIZE:
                _save(chunk, index, output_path)
                index += 1
                chunk = []

        if chunk:
            _save(chunk, index, output_path)


if __name__ == "__main__":
    main()

2. 데이터 삽입 및 색인

변환된 데이터의 분포를 확인하여 primary key로 사용할 필드를 찾습니다. 만 개의 샘플에서 판례정보일련번호(doc_id)는 유일하지만 사건번호(number)는 유일하지 않음을 알 수 있습니다.

doc_id를 primary key로 지정하고 Aeca에 데이터를 입력합니다.

from pathlib import Path

import pandas as pd
from aeca import Channel, DocumentDB
from tqdm import tqdm

_COLLECTION_NAME = "demo.law"
_DATA_PATH = "data/docs"
_AECA_HOST = "localhost"
_AECA_PORT = 10080

def _insert(docdb: DocumentDB, collection_name: str, data_path: Path) -> None:
    if collection_name in docdb.list_collections():
        docdb.drop_collection(collection_name)

    indexes = [
        {
            "fields": ["doc_id"],
            "unique": True,
            "index_type": "kPrimaryKey",
        },
    ]
    docdb.create_collection(collection=collection_name, indexes=indexes)

    files = sorted(data_path.glob("*.pk"))
    for file in tqdm(files):
        df = pd.read_pickle(file, compression="gzip")
        data = df.to_dict(orient="records")
        docdb.insert(collection_name, data)


def main():
    collection_name = _COLLECTION_NAME
    channel = Channel(_AECA_HOST, _AECA_PORT)
    docdb = DocumentDB(channel)
    data_path = Path(_DATA_PATH)

    _insert(docdb, collection_name, data_path)
    _create_index(docdb, collection_name)


if __name__ == "__main__":
    main()

이후 색인을 생성합니다. FTS 검색을 사용할 필드는 standard_cjk analyzer를 사용하고, 문서의 특성상 문어체이며 문법 오류가 작을 확률이 높기 때문에 토크나이저로 mecab을 지정합니다. 필터로 활용할 수 있는 사건종류명(type_name), 선고(judgment) 등은 keyword analyzer를 사용합니다. 마지막으로 선고일자(judgment_date)를 datetime analyzer로 지정합니다.

def _create_index(docdb: DocumentDB, collection_name: str) -> None:
    default_analyzer = {
        "analyzer": {"type": "standard_cjk", "options": {"tokenizer": "mecab"}},
        "index_options": "offsets",
    }
    int_analyzer = {"analyzer": {"type": "int64"}, "index_options": "doc_freqs"}
    keyword_analyzer = {
        "analyzer": {"type": "keyword"},
        "index_options": "doc_freqs",
    }
    datetime_analyzer = {
        "analyzer": {"type": "datetime"},
        "index_options": "doc_freqs",
    }
    index = {
        "index_name": "sk_fts",
        "fields": [
            "doc_id",
            "name",
            "number",
            "judgment_date",
            "judgment",
            "court_name",
            "court_type_code",
            "type_name",
            "type_code",
            "judgment_type",
            "holding_statement",
            "judgment_summary",
            "reference_provisions",
            "reference_cases",
            "content",
        ],
        "index_type": "kFullTextSearchIndex",
        "unique": False,
        "options": {
            "doc_id": int_analyzer,
            "name": default_analyzer,
            "number": keyword_analyzer,
            "judgment_date": datetime_analyzer,
            "judgment": keyword_analyzer,
            "court_name": keyword_analyzer,
            "court_type_code": keyword_analyzer,
            "type_name": keyword_analyzer,
            "type_code": keyword_analyzer,
            "judgment_type": keyword_analyzer,
            "holding_statement": default_analyzer,
            "judgment_summary": default_analyzer,
            "reference_provisions": default_analyzer,
            "reference_cases": default_analyzer,
            "content": default_analyzer,
        },
    }
    docdb.create_index(collection_name, **index)

8.7만 건, 1.1GB의 문서를 입력하고 색인을 구성하는 시간과 색인 용량은 다음과 같습니다. 시간 측정은 MacBook Pro M3 Max, 36GB 메모리에서 측정되었습니다. 색인 생성으로 인해 전체 데이터의 크기가 증가 되었지만 1.1GB의 텍스트 데이터는 244MB로 약 20% 수준으로 압축된 것을 확인할 수 있습니다.

색인 구성에 따라 시간이 달라질 수 있으나, 약 1GB 수준의 텍스트 데이터는 4분 정도에 데이터 입력과 검색이 가능한 상태로 만들 수 있음을 의미합니다. 예를 들어, 작업 과정에서 법원종류코드(court_type_code)가 null을 포함하는 것을 인식하지 못한 상태에서 int64 analyzer를 지정했을 때 색인 과정 오류를 발견하고 keyword analyzer로 변경하여 재색인을 시도하는 과정을 빠르게 진행할 수 있었습니다.

3. 데모 개발

데모는 상품 검색에 자연어 검색을 적용하기에서 구현된 내용을 약간 수정하여 구현했습니다. 동일하게 Next.js를 사용하고 Aeca Javascript SDK를 사용하여 별도의 API 서버를 구성하지 않고 구현했습니다.

API를 구성하는 코드는 다음과 같습니다.

import { StatusCode, statusSuccess } from "@_api/_lib/status"
import { SearchResult } from "@_app/api/law/_lib/document"
import config from "@_app/config"
import { parseJSONStringsInObject } from "@_lib/transform"
import { Channel, DocumentDB } from "@aeca/client"
import { NextResponse } from "next/server"
import { NextRequest } from "next/server"

const _COLLECTION = "demo.law"

export async function GET(
  request: NextRequest,
): Promise<NextResponse<SearchResult>> {
  const channel = new Channel(config.host, config.port)
  const docdb = new DocumentDB(channel)
  const searchParams = request.nextUrl.searchParams
  let query = searchParams.get("query")

  if (!query) {
    return NextResponse.json(
      { code: StatusCode.INVALID_ARGUMENT, message: "query is empty" },
      { status: 500 },
    )
  }

  const aql = {
    $search: {
      query: query,
      highlight: true,
      limit: 30,
    },
  }

  const findStartTime = performance.now()
  const df = await docdb.find(_COLLECTION, aql)
  const findElapsedTime = performance.now() - findStartTime

  return NextResponse.json({
    ...statusSuccess,
    docs: df?.data,
    info: {
      findElapsedTime: findElapsedTime,
    },
  })
}

위의 코드와 같이 단순히 query로 전달되는 변수를 Aeca 쿼리 언어로 변환하는 동작이 대부분입니다. 쿼리를 통해 결과는 대부분 200ms 미만에 받을 수 있습니다. 참고로 현재 배포되어 있는 데모는 Aeca 서버와 웹서버 간 물리적인 거리로 인해 약간의 지연이 포함되어 있습니다.

여기에서는 생략되었지만, 필요하다면 $project를 통해 미사용 필드를 제외하거나 KeyValueDB를 사용하여 검색 결과를 캐싱하는 방법도 있습니다. KeyValueDB를 활용하면 Aeca는 대부분의 상황에서 Redis와 같은 in-memory 캐시 서버가 필요하지 않습니다.

{
  "$search": {
    "query": "공직자 부정청탁",
    "highlight": true,
    "limit": 12
  }
}

추가적으로 판례 검색 데모에는 다음과 같은 필터가 포함되어 있습니다.

이는 다음과 같은 쿼리로 변환됩니다. $search.query의 문법은 Lucene과 유사합니다. 이 기능을 활용하여 API에 변수를 추가하거나 API 엔드포인트 추가 없이 API를 단순화할 수 있습니다.

{
  "$search": {
    "query": "(공직자 부정청탁) AND (type_name:(민사 OR 형사) AND judgment:(선고))",
    "highlight": true,
    "limit": 30
  }
}

마무리

데모 구현은 상품 검색에 자연어 검색을 적용하기의 일부 내용을 사용했습니다. 하지만 완전히 다른 도메인의 서비스를 데이터 수집부터 서비스 개발까지 약 하루의 시간이면 충분했습니다. 이 내용이 실제 사례에 완벽하게 적용되기는 어렵겠지만, 적어도 인프라 구성, 데이터 가공과 저장을 위한 실험 비용이 크지 않고 서비스 구현에 필요한 기능을 온전히 지원한다는 의미이기도 합니다.

Aeca를 사용하면 데이터 저장을 위한 DBMS, Full-text search를 위한 색인 서비스, 빠른 응답을 위한 캐시 서비스, 벡터 임베딩을 위한 모델 서빙을 모두 포함하면서 in-memory 저장소와 달리 메모리 사용량이 억제되어 있습니다. 이 때문에 서비스를 위한 비용과 복잡도를 줄이면서 로컬의 개발 환경과 프로덕션에 배포된 환경을 일치시킬 수 있습니다.

이 데모는 시맨틱 검색을 위한 벡터 검색과 기능을 보강되었으며, 개선된 내용은 자연어로 판례 데이터 검색하기에서 확인하실 수 있습니다.

Aeca 만나보기

Aeca를 더 사용해 보고 싶다면 Docker로 간단히 설치하여 바로 사용할 수 있습니다. Aeca 도입에 대한 더 자세한 설명이나 제품 소개서 요청은 고객 지원에서 문의해 주세요.