[Elastic Search] 검색 구현하기(with Fast API)

ES로 검색엔진을 구현하였다. 구현된 검색 엔진을 실제 서비스처럼 사용하기 위해 REST API를 구현해보자.

REST API의 로직은 단순하게 본다면 2단계이다.

1. 사용자가 검색 키워드를 입력한다. 2. 검색 키워드에 해당하는 문서를 찾는다.

사용자 입력 구현

사용자 입력 구현에서 고려할 점은 단순하게 하나의 키워드만 입력받아서 문서를 상세하게 검색할 수는 없다는 것이다.
유사어, 제외어, 여러 키워드, And 조건, Or 조건 등 다양한 조건으로 검색이 가능하면 사용자에게 더 좋은 검색 시스템이 될 수 있다.
그래서 검색 키워드에서 사용할 수 있도록 특수 커맨드를 정리해보았다.

단일 검색: search=<키워드>                      예) 맨투맨
유사어 검색: search=<Keyword-A + keyword-B>    예) 맨투맨 +아디다스
    => 맨투맨이 포함된 문서에서 아디다스가 포함된 문서
제외어 검색: search=<Keyword-A - keyword-B>    예) 맨투맨 -아디다스
    => 맨투맨이 포함된 문서에서 아디다스라는 단어가 있다면 제외
여러 키워드 검색: search=<keyword-A, keyword-B> 예) 맨투맨, 아디다스
    => 맨투맨이 포함된 문서에서 아디다스러는 단어가 포함된 문서(하지만 강제는 아님)
And 조건 검색: search=<Keyword-A & keyword-B> 예) 맨투맨 & 아디다스
    => 맨투맨과 아디다스가 무조건 들어간 문서
OR 조건 검색: search=<Keyword-A | Keyword-B>  예) 나이키 | 아디다스
    => 맨투맨 또는 아디다스가 포함된 문서
성별 검색결과 부스팅: search=<keyword-A> (여성: ^^, 남성:^~) 
    => 키워드 뒤 해당 특수문자를 붙여 여성 또는 남성이 작성한 것 같은 결과를 부스팅
명사로만 검색: search=<keyword-A>^&            예) 예쁨^&
    => 입력 키워드를 명사로 분류, 해당 명사가 포함된 문서만 가져옴

해당 검색 키워드를 받을 수 있도록 API를 구현해보면 아래와 같은 코드가 된다. 간단한 API 구현을 위해 Python의 Fast API를 사용하였다.

from typing import Optional
from fastapi import FastAPI
from create_es_query import CreateEsIndex
import uvicorn

import json
import elasticsearch

app = FastAPI()
# Elastic Search config
f = open('config.json')
config = json.loads(f.read())
es = elasticsearch.Elasticsearch([f"http://{config['host_url']}:9200"], http_auth=(config['username'], config['password']))

# 페이지 기본 접속을 위한 API
@app.get("/")
def read_root():
    return {"Hello": "World"}

# 검색 키워드를 입력받기 위한 API
@app.get("/search/keyword")
def search_result(search: str = None):
    qry = CreateEsIndex()
    search_key = search
    query = qry.create_complex_query(search_key)

    t = es.search(index="review_pos", body=query)
    doc_list = t.body['hits']['hits']
    hit_docs = [d['_source'] for d in doc_list]
    return {"result": hit_docs}

if __name__ == '__main__':
    uvicorn.run(app, host="localhost", port=8000)

위 코드에서 search_result 함수로 검색어를 받는 API를 구현하였다. 실제로 해당 API로 요청하려면
http://localhost:8000/search/keyword?search=<keyword>로 요청하면 된다.

입력 키워드 전처리 로직 구현

사용자로 부터 입력받은 키워드에 유사어가 포함되었는지, 제외어가 포함되었는지 컴퓨터는 알지 못한다. 이를 위해 별도의 모듈을 하나 만들어 입력받은 키워드를 구분할 수 있도록 하였다.

class CreateEsIndex:

    def __init__(self):
        self.main_keyword = []
        self.synonym, self.exclude, self.multi_key, self.and_cond, \
        self.or_cond, self.woman, self.man, self.noun = [], [], [], [], [], [], [], []
        self.condition_map = {
            '-': self.exclude, '+': self.synonym, '|': self.or_cond, '&': self.and_cond,
            ',': self.multi_key, '^^': self.woman, '^~': self.man, '^&': self.noun
        }
        return

    @staticmethod
    def text_clear(keyword: str):
        rep_list = ['-', '+', '|', '&', ',', '^^', '^~', '^&']
        for r in rep_list:
            keyword = keyword.replace(r, '')
        return keyword

    def process_word(self, search_word: str):
        word_list = search_word.split(' ')
        for i in range(0, len(word_list)):
            w = word_list[i]
            if i == 0:
                self.main_keyword.append({"match": {"review": self.text_clear(w)}})
            [self.condition_map[s].append(self.text_clear(w)) if s in w else None for s in self.condition_map.keys()]

    def create_complex_query(self, search_word: str):
        self.process_word(search_word)
        q = {
            "query": {
                "bool": {
                    "must": self.main_keyword,
                    "must_not": [],
                    "should": []
                }
            }
        }
        bol = q['query']['bool']
        for k, v in self.condition_map.items():
            if not v:
                continue
            words = [{"match": {"review": w}} for w in v]
            if k == "+":
                bol['should'].extend(words)
            if k == "-":
                bol['must_not'].extend(words)
            if k == ",":
                bol['should'].extend(words)
            if k == "&":
                bol['must'].extend(words)
            if k == "|":
                bol['should'].extend(words)
            if k == "^^":
                boost = {"rank_feature": {"field": "genders.female", "boost": 4}}
                bol['should'].append(boost)
            if k == "^~":
                boost = {"rank_feature": {"field": "genders.male", "boost": 4}}
                bol['should'].append(boost)
            if k == "^&":
                words = [{"match": {"review.nori_noun": w}} for w in v]
                bol['should'].extend(words)
        return q

위 코드에서 CreateEsIndex 클래스의 초기 변수로 검색 조건에 해당하는 리스트를 생성하였다.
여기에 적절한 단어를 추가하기 위해 process_word 함수로 사용자로부터 입력받은 키워드를 분류한다. 이를 통해 ES에 조회할 수 있는 쿼리를 생성하였다.

예시

키워드: 맨투맨

입력 Query

{'query': {'bool': {'must': [{'match': {'review': '맨투맨'}}], 'must_not': [], 'should': []}}}

결과

{
  "result": [
    {
      "gender": {
        "female": 0.385272741317749,
        "male": 0.614727258682251
      },
      "prd_id": 957878,
      "review": "아디다스 맨투맨 후기진짜 오버핏 제대로인 맨투맨 인생 맨투맨!!!!!",
      "review_id": 7325803
    },
    {
      "gender": {
        "female": 0.5234223008155823,
        "male": 0.4765776991844177
      },
      "prd_id": 897632,
      "review": "맨투맨 뭐살지 고민하신다면 이 맨투맨 추천드려요",
      "review_id": 4934238
    },
    ...
  ]
}

키워드: 맨투맨 -아디다스

입력 Query

{'query': {'bool': {'must': [{'match': {'review': '맨투맨'}}], 'must_not': [{'match': {'review': '아디다스'}}], 'should': []}}}

결과

{
  "result": [
    {
      "gender": {
        "female": 0.5234223008155823,
        "male": 0.4765776991844177
      },
      "prd_id": 897632,
      "review": "맨투맨 뭐살지 고민하신다면 이 맨투맨 추천드려요",
      "review_id": 4934238
    },
    {
      "gender": {
        "female": 0.6838095188140869,
        "male": 0.3161904811859131
      },
      "prd_id": 1163605,
      "review": "브러시트 클럽 맨투맨 블랙입니다브러시트 클럽 맨투맨 블랙입니다",
      "review_id": 7883291
    },...
   ]
}