OpenSearchSQLRetriever

OpenSearch Document Store에 원시 OpenSearch SQL 쿼리를 실행하고 원시 JSON 응답을 돌려주는 컴포넌트예요.

출처: 문서

본문

OpenSearchSQLRetriever는 OpenSearchDocumentStore에 OpenSearch SQL 쿼리를 직접 실행할 수 있게 해 줘요. OpenSearchBM25Retriever나 OpenSearchEmbeddingRetriever처럼 검색어를 문서와 매칭하는 대신, SQL 문장을 실행하고 OpenSearch SQL API의 원시 JSON 응답을 돌려줘요.

실행 시점에 인덱스에 구조적으로 접근해야 할 때 유용한데, 예를 들어 특정 필드를 가져오거나, 메타데이터로 필터링하거나, 개수·평균 같은 집계(aggregation)를 계산할 때가 그래요.

다른 OpenSearch retriever와 달리 이 컴포넌트는 Document 객체 목록을 돌려주지 않아요. 출력은 단일 result 사전인데, result["result"]에 OpenSearch SQL 플러그인의 기본 JDBC 형식으로 된 원시 응답이 담겨요.

schema는 선택한 열별로 각각 name, 선택적 alias, type을 가진 열 설명자(column descriptor) 목록이에요.
datarows는 행 목록이고, 각 행은 schema와 순서가 맞는 값 목록이에요.
total, size, status는 각각 일치하는 행 수, 반환된 행 수, SQL 호출의 HTTP 상태를 나타내요.

일반 쿼리와 집계 쿼리 모두 같은 형태가 반환돼요. COUNT(*) 같은 집계도 datarows의 단일 행으로 돌아와요.

이 컴포넌트는 초기화 시 두 개의 선택 파라미터를 받아요.

raise_on_failure: True(기본값)면 SQL API 호출이 실패할 때 예외가 발생해요. False면 오류를 경고로 기록하고 결과는 비어요.
fetch_size: 페이지당 가져올 결과 수예요. 설정하지 않으면 OpenSearch에 설정된 기본 fetch size를 사용해요.

설치

OpenSearch를 설치한 뒤 인스턴스를 시작해요.

Docker가 설정되어 있다면 Docker 이미지를 받아 실행하는 것을 권장해요.

docker pull opensearchproject/opensearch:3

docker run -p 9200:9200 -p 9600:9600 -e "discovery.type=single-node" -e "OPENSEARCH_INITIAL_ADMIN_PASSWORD=<custom-admin-password>" opensearchproject/opensearch:3

대안으로 OpenSearch integration GitHub에 가서 제공된 docker-compose.yml로 Docker 컨테이너를 시작할 수 있어요.

docker compose up

실행 중인 OpenSearch 인스턴스가 준비되면 opensearch-haystack 통합을 설치해요.

pip install opensearch-haystack

더 알아보기 (Learn more)

단독으로 쓰기

인덱스에 문서 몇 개를 쓴 뒤 SQL 쿼리를 실행해요. 아래 예제는 인덱스에서 content 필드를 선택하고 반환된 히트를 읽어요.

from haystack import Document
from haystack_integrations.components.retrievers.opensearch import (
    OpenSearchSQLRetriever,
)
from haystack_integrations.document_stores.opensearch import (
    OpenSearchDocumentStore,
)
from haystack.document_stores.types import DuplicatePolicy


document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="my_index",
    use_ssl=True,
    verify_certs=False,
    http_auth=("admin", "<custom-admin-password>"),
)

documents = [
    Document(content="There are over 7,000 languages spoken around the world today."),
    Document(
        content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.",
    ),
    Document(
        content="In certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.",
    ),
]

# DuplicatePolicy.SKIP is optional, but useful to run the script multiple times without throwing errors
document_store.write_documents(documents=documents, policy=DuplicatePolicy.SKIP)

retriever = OpenSearchSQLRetriever(document_store=document_store)

output = retriever.run(query="SELECT content FROM my_index LIMIT 10")

result = output["result"]

for row in result["datarows"]:
    print(row)

schema 항목은 행의 각 위치가 어떤 열에 해당하는지 알려줘요.

print(result["schema"])

# [{'name': 'content', 'type': 'text'}]

집계 쿼리 실행

이 컴포넌트는 원시 SQL 응답을 돌려주므로, 문서 기반 retriever가 지원하지 않는 문서 개수 세기 같은 집계에도 쓸 수 있어요.

retriever = OpenSearchSQLRetriever(document_store=document_store)

output = retriever.run(query="SELECT COUNT(*) AS doc_count FROM my_index")

result = output["result"]

print(result)

# {'schema': [{'name': 'COUNT(*)', 'alias': 'doc_count', 'type': 'long'}],
#  'datarows': [[3]], 'total': 1, 'size': 1, 'status': 200}

잘못되었거나 실패하는 쿼리에서 예외가 발생하지 않게 하려면 raise_on_failure=False로 컴포넌트를 초기화하세요. 그 경우 실패한 쿼리는 경고를 기록하고 빈 결과를 반환해요.