Elasticsearch 벡터스토어와 하이브리드 검색

Elasticsearch 벡터스토어와 하이브리드 검색 (ElasticsearchStore)

벡터 검색만으로는 부족할 때, 하이브리드 검색이 정답이 될 때가 있어요. 의미 기반 검색과 키워드 기반 검색을 함께 써서 두 방식의 장점을 모두 얻는 전략이죠. Elasticsearch는 분산·RESTful 검색 및 분석 엔진으로, 벡터 검색과 어휘(lexical) 검색을 모두 수행할 수 있어서 하이브리드 검색을 구현하기에 아주 적합해요. 이 페이지에서는 LangChain의 ElasticsearchStore로 하이브리드 검색을 구성하는 방법을 함께 볼게요.

출처: 공식문서

Elasticsearch는 Apache Lucene 라이브러리 위에 구축된 분산 검색 엔진이에요. 벡터와 어휘 검색을 모두 처리할 수 있죠.

설정

Elasticsearch 벡터 검색을 쓰려면 langchain-elasticsearch 패키지를 설치해야 해요.

pip install -qU langchain-elasticsearch

자격 증명

Elasticsearch 인스턴스를 쓸 수 있는 두 가지 주요 방법이 있어요.

  1. Elastic Cloud: 관리형 Elasticsearch 서비스예요. 무료 체험판에 가입할 수 있어요. 로그인 자격 증명이 필요 없는 인스턴스라면 Elasticsearch URL과 인덱스 이름, 임베딩 객체를 생성자에 전달하면 돼요.
  2. 로컬 설치: 공식 Elasticsearch Docker 이미지로 로컬에서 실행할 수 있어요. 자세한 내용은 Elasticsearch Docker 문서를 참고하세요.

로컬에서 Elasticsearch 실행하기

개발·테스트용으로 가장 쉬운 방법은 start-local 스크립트예요. Docker로 Elasticsearch(Kibana 포함 가능)를 한 줄로 구성해 줘요.

curl -fsSL https://elastic.co/start-local | sh

그러면 설정 파일과 시작 스크립트가 들어 있는 elastic-start-local 폴더가 만들어져요. 시작하려면:

cd elastic-start-local
./start.sh

Elasticsearch는 http://localhost:9200에서 접근할 수 있어요. elastic 사용자의 비밀번호와 API 키는 자동 생성되어 elastic-start-local 폴더의 .env 파일에 저장돼요. Kibana 없이 Elasticsearch만 필요하면 --esonly 옵션을 사용하면 돼요.

curl -fsSL https://elastic.co/start-local | sh -s -- --esonly

start-local 설정은 로컬 테스트 전용이에요. 프로덕션에는 사용하면 안 되고, 공식 Elasticsearch 문서를 따르세요.

인증과 함께 실행하기

프로덕션에서는 보안을 활성화한 채 실행하는 걸 권장해요. 로그인 자격 증명으로 연결하려면 es_api_key 또는 es_user와 es_password 파라미터를 쓰면 돼요.

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

from langchain_elasticsearch import ElasticsearchStore

elastic_vector_search = ElasticsearchStore(
    es_url="http://localhost:9200",
    index_name="langchain_index",
    embedding=embeddings,
    es_user="elastic",
    es_password="changeme",
)

초기화

Elasticsearch가 localhost:9200에서 로컬로 실행 중이라면(앞서 docker로 실행한 경우) 아래처럼 초기화해요.

from langchain_elasticsearch import ElasticsearchStore

vector_store = ElasticsearchStore(
    "langchain-demo", embedding=embeddings, es_url="http://localhost:9201"
)

검색

벡터스토어를 만들고 관련 문서를 추가했다면, 체인이나 에이전트를 돌리는 동안 검색을 호출하게 될 거예요. 검색 시 메타데이터로 필터링을 걸 수도 있어요.

results = vector_store.similarity_search(
    query="LangChain provides abstractions to make working with LLMs easy",
    k=2,
    filter=[{"term": {"metadata.source.keyword": "tweet"}}],
)
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")

* Building an exciting new project with LangChain - come check it out! [{'source': 'tweet'}]
* LangGraph is the best framework for building stateful, agentic applications! [{'source': 'tweet'}]

벡터스토어는 리트리버로 바꿔 체인에서 더 쉽게 쓸 수도 있어요.

retriever = vector_store.as_retriever(
    search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.2}
)
retriever.invoke("Stealing from the bank is a crime")

[Document(metadata={'source': 'news'}, page_content='Robbers broke into the city bank and stole $1 million in cash.'),
 Document(metadata={'source': 'news'}, page_content='The stock market is down 500 points today due to fears of a recession.'),
 Document(metadata={'source': 'website'}, page_content='Is the new iPhone worth the price? Read this review to find out.'),
 Document(metadata={'source': 'tweet'}, page_content='Building an exciting new project with LangChain - come check it out!')]

검색 전략

Elasticsearch가 다른 벡터 전용 데이터베이스보다 크게 앞서는 점은, 다양한 검색 전략을 지원한다는 거예요. 기본적으로 ElasticsearchStore는 DenseVectorStrategy(버전 0.2.0 이전에는 ApproxRetrievalStrategy로 불림)를 사용해요.

DenseVectorStrategy

쿼리 벡터와 가장 유사한 상위 k개 벡터를 반환해요. k는 초기화 시 설정하며 기본값은 10이에요.

from langchain_elasticsearch import DenseVectorStrategy

db = ElasticsearchStore.from_documents(
    docs,
    embeddings,
    es_url="http://localhost:9200",
    index_name="test",
    strategy=DenseVectorStrategy(),
)

docs = db.similarity_search(
    query="What did the president say about Ketanji Brown Jackson?", k=10
)

하이브리드 검색: 밀집 벡터 + 키워드 검색

하이브리드 검색(approximate semantic search + keyword search)을 구성하는 예시예요. 두 검색 방식의 스코어를 균형 있게 합치기 위해 **RRF(Reciprocal Rank Fusion)**를 사용해요. 하이브리드를 활성화하려면 DenseVectorStrategy 생성자에 hybrid=True를 설정하면 돼요.

db = ElasticsearchStore.from_documents(
    docs,
    embeddings,
    es_url="http://localhost:9200",
    index_name="test",
    strategy=DenseVectorStrategy(hybrid=True),
)

하이브리드를 켜면 쿼리는 approximate semantic search와 keyword search가 결합된 형태로 수행돼요. 두 검색 방법의 스코어를 균형 있게 합치려면 RRF(Reciprocal Rank Fusion)를 사용해요. RRF는 Elasticsearch 8.9.0 이상이 필요해요.

{
    "retriever": {
        "rrf": {
            "retrievers": [
                {
                    "standard": {
                        "query": {
                            "bool": {
                                "filter": [],
                                "must": [{"match": {"text": {"query": "foo"}}}],
                            }
                        },
                    },
                },
                {
                    "knn": {
                        "field": "vector",
                        "filter": [],
                        "k": 1,
                        "num_candidates": 50,
                        "query_vector": [1.0, ..., 0.0],
                    },
                },
            ]
        }
    }
}

SparseVectorStrategy (ELSER)

Elasticsearch의 희소 벡터 검색으로 top-k 결과를 가져오는 전략이에요. 지금은 자체 ELSER 임베딩 모델만 지원하며, ELSER 모델이 Elasticsearch ML 노드에 배포·실행 중이어야 해요.

from langchain_elasticsearch import SparseVectorStrategy

# 토큰은 Elasticsearch 내부에서 인덱스 시점과 쿼리 시점에 추론되므로 임베딩 함수가 필요 없어요.
db = ElasticsearchStore.from_documents(
    docs,
    es_cloud_id="<cloud id>",
    es_user="elastic",
    es_password="<cloud password>",
    index_name="test-elser",
    strategy=SparseVectorStrategy(model_id=".elser_model_2"),
)

db.client.indices.refresh(index="test-elser")

results = db.similarity_search(
    "What did the president say about Ketanji Brown Jackson", k=4
)
print(results[0])

BM25Strategy

마지막으로 전체 텍스트 키워드 검색을 쓸 수도 있어요. 벡터 검색 없이 순수 BM25로 검색하고 싶으면 BM25Strategy를 ElasticsearchStore 생성자에 지정하면 돼요.

from langchain_elasticsearch import BM25Strategy

db = ElasticsearchStore.from_documents(
    docs,
    es_url="http://localhost:9200",
    index_name="test",
    strategy=BM25Strategy(),
)

from langchain_elasticsearch import ElasticsearchStore

db = ElasticsearchStore(
    es_url="http://localhost:9200",
    index_name="test_index",
    strategy=ElasticsearchStore.BM25RetrievalStrategy(),
)

db.add_texts(
    ["foo", "foo bar", "foo bar baz", "bar", "bar baz", "baz"],
)

results = db.similarity_search(query="foo", k=10)
print(results)

검색 증강 생성(RAG)에서 사용하기

이 벡터스토어를 검색 증강 생성(RAG)에 활용하는 가이드는 다음 섹션을 참고하세요.

FAQ

Elasticsearch에 문서를 색인할 때 타임아웃 오류가 나요. 어떻게 고치죠?

문서 색인이 예상보다 오래 걸릴 수 있어요. ElasticsearchStore는 Elasticsearch bulk API를 사용하는데, 타임아웃 가능성을 줄이도록 조정할 수 있는 기본값이 몇 가지 있어요. 기본값은 chunk_size: 500, max_chunk_bytes: 100MB예요. 이를 조정하려면 add_texts 메서드에 chunk_size와 max_chunk_bytes 파라미터를 전달하면 돼요.

    vector_store.add_texts(
        texts,
        bulk_kwargs={
            "chunk_size": 50,
            "max_chunk_bytes": 200000000
        }
    )

더 알아보기 (Learn more)