MLTransform으로 임베딩 생성

이 페이지에서는 MLTransform 특성을 사용하여 머신러닝 (ML) 모델을 학습시킬 수 있는 데이터를 준비하는 이유와 방법을 설명합니다. 특히 이 페이지에서는 MLTransform를 사용하여 임베딩을 생성하여 데이터를 처리하는 방법을 보여줍니다.

한 클래스에 여러 데이터 처리 변환을 결합하면 MLTransform에서 Apache Beam ML 데이터 처리 작업을 워크플로에 적용하는 프로세스를 간소화합니다.

데이터 처리 단계가 강조표시된 Dataflow ML 워크플로 다이어그램

그림 1. 전체 Dataflow ML 워크플로. 워크플로의 전처리 단계에서 MLTransform을 사용합니다.

임베딩 개요

임베딩은 최신 시맨틱 검색 및 검색 증강 생성 (RAG) 애플리케이션에 필수적입니다. 임베딩을 사용하면 시스템이 더 깊고 개념적인 수준에서 정보를 이해하고 상호작용할 수 있습니다. 시맨틱 검색에서 임베딩은 쿼리와 문서를 벡터 표현으로 변환합니다. 이러한 표현은 기본 의미와 관계를 포착합니다. 따라서 키워드가 직접 일치하지 않더라도 관련 결과를 찾을 수 있습니다. 이는 표준 키워드 기반 검색을 크게 뛰어넘는 기능입니다. 제품 추천에 임베딩을 사용할 수도 있습니다. 여기에는 이미지와 텍스트를 사용하는 멀티모달 검색, 로그 분석, 중복 제거와 같은 작업이 포함됩니다.

RAG 내에서 임베딩은 기술 자료에서 가장 관련성이 높은 컨텍스트를 검색하여 대규모 언어 모델(LLM)의 대답을 그라운딩하는 데 중요한 역할을 합니다. 사용자의 질문과 기술 자료의 정보 청크를 모두 삽입함으로써 RAG 시스템은 의미적으로 가장 유사한 부분을 효율적으로 식별하고 검색할 수 있습니다. 이러한 시맨틱 매칭을 통해 LLM은 정확하고 유익한 대답을 생성하는 데 필요한 정보에 액세스할 수 있습니다.

임베딩을 위한 데이터 수집 및 처리

청킹 및 임베딩 생성의 데이터 처리 단계가 강조 표시된 임베딩 지식 수집 다이어그램

그림 2. 지식 수집 다이어그램 입력 멀티모달 문서 데이터와 청킹 및 임베딩 생성이라는 두 가지 처리 단계를 보여줍니다. 청킹은 임베딩 생성 전에 복잡한 데이터에 사용되는 전처리 단계입니다. 데이터가 처리되면 임베딩이 벡터 데이터베이스에 저장됩니다.

핵심 임베딩 사용 사례의 경우 지식을 수집하고 처리하는 방법이 주요 고려사항입니다. 이 수집은 일괄 또는 스트리밍 방식으로 이루어질 수 있습니다. 이 지식의 출처는 매우 다양할 수 있습니다. 예를 들어 이 정보는 Cloud Storage에 저장된 파일에서 가져올 수도 있고 Pub/Sub 또는 Apache Kafka용 Google Cloud 관리형 서비스와 같은 스트리밍 소스에서 가져올 수도 있습니다.

스트리밍 소스의 경우 데이터 자체가 원시 콘텐츠 (예: 일반 텍스트)이거나 문서를 가리키는 URI일 수 있습니다. 소스와 관계없이 첫 번째 단계에서는 일반적으로 정보를 사전 처리합니다. 원시 텍스트의 경우 기본 데이터 정리와 같이 최소한일 수 있습니다. 하지만 더 큰 문서나 더 복잡한 콘텐츠의 경우 중요한 단계는 청킹입니다. 청킹은 소스 자료를 관리 가능한 더 작은 단위로 나누는 것입니다. 최적의 청크 전략은 표준화되어 있지 않으며 특정 데이터와 애플리케이션에 따라 달라집니다. Dataflow와 같은 플랫폼은 다양한 청크 요구사항을 처리하는 기본 제공 기능을 제공하여 이 필수 전처리 단계를 간소화합니다.

이점

MLTransform 클래스는 다음과 같은 이점을 제공합니다.

  • 데이터를 벡터 데이터베이스로 푸시하거나 추론을 실행하는 데 사용할 수 있는 임베딩을 생성합니다.
  • 복잡한 코드를 작성하거나 기본 라이브러리를 관리하지 않고도 데이터를 변환할 수 있습니다.
  • 인터페이스 하나에서 여러 유형의 처리 작업을 효율적으로 연결할 수 있습니다.

지원 및 제한 사항

MLTransform 클래스에는 다음과 같은 제한사항이 있습니다.

  • Apache Beam Python SDK 버전 2.53.0 이상을 사용하는 파이프라인에 사용할 수 있습니다.
  • 파이프라인에서 기본 기간을 사용해야 합니다.

텍스트 임베딩 변환:

사용 사례

예시 노트북에서는 특정 사용 사례에 MLTransform를 사용하는 방법을 보여줍니다.

Vertex AI를 사용하여 LLM의 텍스트 임베딩 생성
Vertex AI 텍스트 임베딩 API와 함께 Apache Beam MLTransform 클래스를 사용하여 텍스트 임베딩을 생성합니다. 텍스트 임베딩은 텍스트를 숫자 벡터로 표현하기 위한 방법으로, 많은 자연어 처리(NLP) 작업에 필요합니다.
Hugging Face를 사용하여 LLM의 텍스트 임베딩 생성
Hugging Face Hub 모델과 함께 Apache Beam MLTransform 클래스를 사용하여 텍스트 임베딩을 생성합니다. Hugging Face SentenceTransformers 프레임워크는 Python을 사용하여 문장, 텍스트, 이미지 임베딩을 생성합니다.
텍스트 임베딩을 생성하여 PostgreSQL용 AlloyDB에 수집하고 싶습니다
Apache Beam, 특히 Hugging Face Hub 모델과 함께 MLTransform 클래스를 사용하여 텍스트 임베딩을 생성합니다. 그런 다음 VectorDatabaseWriteTransform를 사용하여 이러한 임베딩과 연결된 메타데이터를 PostgreSQL용 AlloyDB에 로드합니다. 이 노트북에서는 PostgreSQL용 AlloyDB 벡터 데이터베이스를 채우기 위한 확장 가능한 배치 및 스트리밍 Beam 데이터 파이프라인을 빌드하는 방법을 보여줍니다. 여기에는 Pub/Sub 또는 기존 데이터베이스 테이블과 같은 다양한 소스의 데이터 처리, 맞춤 스키마 만들기, 데이터 업데이트가 포함됩니다.
텍스트 임베딩을 생성하고 BigQuery로 수집하고 싶습니다
Hugging Face Hub 모델과 함께 Apache Beam MLTransform 클래스를 사용하여 제품 카탈로그와 같은 애플리케이션 데이터에서 텍스트 임베딩을 생성합니다. 이를 위해 Apache Beam HuggingfaceTextEmbeddings 변환이 사용됩니다. 이 변환은 문장 및 텍스트 임베딩을 생성하는 모델을 제공하는 Hugging Face SentenceTransformers 프레임워크를 사용합니다. 생성된 이러한 임베딩과 메타데이터는 Apache Beam VectorDatabaseWriteTransform를 사용하여 BigQuery로 수집됩니다. 이 노트북에서는 Enrichment 변환을 사용하여 BigQuery에서 벡터 유사성 검색을 추가로 보여줍니다.

사용 가능한 변환의 전체 목록은 Apache Beam 문서의 변환을 참고하세요.

임베딩 생성을 위해 MLTransform 사용

MLTransform 클래스를 사용하여 정보를 청크하고 임베딩을 생성하려면 파이프라인에 다음 코드를 포함합니다.

  
  def create_chunk(product: Dict[str, Any]) -> Chunk:
    return Chunk(
        content=Content(
            text=f"{product['name']}: {product['description']}"
        ),
        id=product['id'],  # Use product ID as chunk ID
        metadata=product,  # Store all product info in metadata
    )

  [...]
  with beam.Pipeline() as p:
    _ = (
            p
            | 'Create Products' >> beam.Create(products)
            | 'Convert to Chunks' >> beam.Map(create_chunk)
            | 'Generate Embeddings' >> MLTransform(
  write_artifact_location=tempfile.mkdtemp())
  .with_transform(huggingface_embedder)
            | 'Write to AlloyDB' >> VectorDatabaseWriteTransform(alloydb_config)
        )

이전 예에서는 요소당 하나의 청크를 만들지만 LangChain을 사용하여 청크를 만들 수도 있습니다.

splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=20)
provider = beam.ml.rag.chunking.langchain.LangChainChunker(
document_field='content', metadata_fields=[], text_splitter=splitter)

with beam.Pipeline() as p:
_ = (
p
| 'Create Products' >> beam.io.textio.ReadFromText(products)
| 'Convert to Chunks' >> provider.get_ptransform_for_processing()

다음 단계