BigtableByteStore 통합

BigtableByteStore 통합

키-값 저장소를 안정적이고 빠른 와이드 컬럼 DB에 두고 싶을 때가 있어요. 구조화·반구조화·비구조화 데이터를 빠르게 읽고 쓰는 데 강한 백엔드가 필요하다면, Google Cloud Bigtable이 좋은 선택이 돼요. LangChain의 BigtableByteStore는 Bigtable을 키-값 저장소의 백엔드로 연결해 줘요.

출처: 공식문서 - BigTableByteStore integration

Bigtable은 키-값이면서 와이드 컬럼 구조의 저장소예요. 구조화된 데이터든 반구조화든, 심지어 구조 없는 데이터든 빠른 접근에 적합하죠. BigtableByteStore는 동기·비동기 연산을 모두 지원하며, 키-값 쌍을 저장·조회·삭제할 수 있어요.

Open In Colab 버튼으로 노트북에서 바로 실행해 볼 수도 있어요.

준비하기

사전 요건

시작하려면 활성화된 Bigtable 인스턴스와 테이블이 있는 Google Cloud 프로젝트가 필요해요.

• Google Cloud 프로젝트 만들기
• Bigtable API 활성화
• Bigtable 인스턴스와 테이블 만들기

설치

통합은 langchain-google-bigtable 패키지에 들어 있어요. 아래 명령은 임베딩 캐시 예시를 위해 langchain-google-vertexai도 함께 설치해요.

pip install -qU langchain-google-bigtable langchain-google-vertexai

프로젝트 설정

노트북 안에서 사용할 Google Cloud 프로젝트를 지정해요. 프로젝트 ID를 모른다면 gcloud config list를 실행하거나 프로젝트 ID 찾기 문서를 참고하면 돼요.

# @markdown 프로젝트, 인스턴스, 테이블 정보를 채워 주세요.
PROJECT_ID = "your-gcp-project-id"  # @param {type:"string"}
INSTANCE_ID = "your-instance-id"  # @param {type:"string"}
TABLE_ID = "your-table-id"  # @param {type:"string"}

!gcloud config set project {PROJECT_ID}

인증

프로젝트 리소스에 접근하려면 Google Cloud에 인증해야 해요. Colab에서는 아래 셀을 쓰고, Gemini Enterprise Agent Platform Workbench에서는 설정 문서를 참고해요.

from google.colab import auth

auth.authenticate_user()

만들기

BigtableByteStore를 쓰려면 먼저 테이블이 존재하는지 확인하고, 연결을 관리할 BigtableEngine을 초기화해요.

from langchain_google_bigtable import (
    BigtableByteStore,
    BigtableEngine,
    init_key_value_store_table,
)

# 테이블과 컬럼 패밀리가 있는지 확인해요.
init_key_value_store_table(
    project_id=PROJECT_ID,
    instance_id=INSTANCE_ID,
    table_id=TABLE_ID,
)

BigtableEngine

BigtableEngine 객체는 저장소의 실행 컨텍스트를 다뤄요. 특히 비동기 연산에서 중요하죠. 성능을 위해 엔진 하나를 초기화해서 여러 저장소에서 재사용하는 걸 권장해요.

# 비동기 연산을 관리할 엔진을 초기화해요.
engine = await BigtableEngine.async_initialize(
    project_id=PROJECT_ID, instance_id=INSTANCE_ID
)

BigtableByteStore

이게 키-값 저장소와 상호작용하는 핵심 클래스예요. 데이터를 저장·조회·삭제하는 메서드를 제공하죠.

# 저장소를 초기화해요.
store = await BigtableByteStore.create(engine=engine, table_id=TABLE_ID)

사용하기

저장소는 동기(mset, mget)와 비동기(amset, amget) 메서드를 모두 지원해요. 이 가이드에서는 비동기 버전을 쓸게요.

저장

amset으로 키-값 쌍을 저장해요.

kv_pairs = [
    ("key1", b"value1"),
    ("key2", b"value2"),
    ("key3", b"value3"),
]

await store.amset(kv_pairs)

조회

amget으로 값을 가져와요. 키가 없으면 그 키에 대해 None이 돌아와요.

retrieved_vals = await store.amget(["key1", "key2", "nonexistent_key"])
print(retrieved_vals)

삭제

amdelete로 저장소에서 키를 제거해요.

await store.amdelete(["key3"])

# 키가 삭제됐는지 확인해요.
await store.amget(["key1", "key3"])

키 순회

ayield_keys로 모든 키 또는 특정 접두사로 시작하는 키를 순회할 수 있어요.

all_keys = [key async for key in store.ayield_keys()]
print(f"All keys: {all_keys}")

prefixed_keys = [key async for key in store.ayield_keys(prefix="key1")]
print(f"Prefixed keys: {prefixed_keys}")

고급 활용: 임베딩 캐시

키-값 저장소의 대표적인 용도는 텍스트 임베딩 계산처럼 비용이 큰 연산을 캐시해서 시간과 비용을 아끼는 거예요.

from langchain_classic.embeddings import CacheBackedEmbeddings
from langchain_google_vertexai.embeddings import VertexAIEmbeddings

underlying_embeddings = VertexAIEmbeddings(
    project=PROJECT_ID, model_name="textembedding-gecko@003"
)

# 다른 데이터와 키 충돌을 피하려고 네임스페이스를 써요.
cached_embedder = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings, store, namespace="text-embeddings"
)

print("First call (computes and caches embedding):")
%time embedding_result_1 = await cached_embedder.aembed_query("Hello, world!")

print("\nSecond call (retrieves from cache):")
%time embedding_result_2 = await cached_embedder.aembed_query("Hello, world!")

첫 호출은 임베딩을 계산해 캐시하고, 두 번째 호출은 캐시에서 바로 가져오는 걸 볼 수 있어요.

간단한 문서 검색기로 쓰기

이번엔 Bigtable 저장소를 이용해 간단한 검색기(retriever)를 만드는 예시예요. 저장소를 문서 영속 계층으로 삼아, 쿼리 접두사와 일치하는 문서를 가져와요.

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List, Optional, Any, Union
import json


class SimpleKVStoreRetriever(BaseRetriever):
    """키-값 저장소에서 접두사 일치로 문서를 가져오는 간단한 검색기예요."""

    store: BigtableByteStore
    documents: List[Union[Document, str]]
    k: int

    def set_up_store(self):
        kv_pairs_to_set = []
        for i, doc in enumerate(self.documents):
            if isinstance(doc, str):
                doc = Document(page_content=doc)
            if not doc.id:
                doc.id = str(i)
            value = (
                "Page Content\n"
                + doc.page_content
                + "\nMetadata"
                + json.dumps(doc.metadata)
            )
            kv_pairs_to_set.append((doc.id, value.encode("utf-8")))
        self.store.mset(kv_pairs_to_set)

    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: Optional[CallbackManagerForRetrieverRun] = None,
    ) -> List[Document]:
        keys = [key async for key in self.store.ayield_keys(prefix=query)][: self.k]
        documents_retrieved = []
        async for document in await self.store.amget(keys):
            if document:
                document_str = document.decode("utf-8")
                page_content = document_str.split("Content\n")[1].split("\nMetadata")[0]
                metadata = json.loads(document_str.split("\nMetadata")[1])
                documents_retrieved.append(
                    Document(page_content=page_content, metadata=metadata)
                )
        return documents_retrieved

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: Optional[CallbackManagerForRetrieverRun] = None,
    ) -> list[Document]:
        keys = [key for key in self.store.yield_keys(prefix=query)][: self.k]
        documents_retrieved = []
        for document in self.store.mget(keys):
            if document:
                document_str = document.decode("utf-8")
                page_content = document_str.split("Content\n")[1].split("\nMetadata")[0]
                metadata = json.loads(document_str.split("\nMetadata")[1])
                documents_retrieved.append(
                    Document(page_content=page_content, metadata=metadata)
                )
        return documents_retrieved

documents = [
    Document(
        page_content="Goldfish are popular pets for beginners, requiring relatively simple care.",
        metadata={"type": "fish", "trait": "low maintenance"},
        id="fish#Goldfish",
    ),
    Document(
        page_content="Cats are independent pets that often enjoy their own space.",
        metadata={"type": "cat", "trait": "independence"},
        id="mammals#Cats",
    ),
    Document(
        page_content="Rabbits are social animals that need plenty of space to hop around.",
        metadata={"type": "rabbit", "trait": "social"},
        id="mammals#Rabbits",
    ),
]

retriever_store = BigtableByteStore.create_sync(
    engine=engine, instance_id=INSTANCE_ID, table_id=TABLE_ID
)

KVDocumentRetriever = SimpleKVStoreRetriever(
    store=retriever_store, documents=documents, k=2
)

KVDocumentRetriever.set_up_store()

KVDocumentRetriever.invoke("fish")

KVDocumentRetriever.invoke("mammals")

물고기(fish)를 조회하면 해당 접두사와 일치하는 문서가, 동물(mammals)을 조회하면 그에 맞는 문서가 돌아오는 것을 확인할 수 있어요.

API 레퍼런스

BigtableByteStore 클래스에 대한 자세한 내용은 GitHub의 소스 코드에서 확인할 수 있어요.

더 알아보기 (Learn more)

• 키-값 저장소(스토어) 통합