임베딩

임베딩 (Embeddings)

임베딩(embedding)은 텍스트의 의미를 포착하는 벡터 표현이에요. 다음과 같은 것을 만들 때 필수적입니다:

• 시맨틱 검색 (Semantic search) — 키워드 매칭이 아니라 의미로 문서 찾기
• RAG (Retrieval-Augmented Generation) — AI 에이전트에 관련 컨텍스트 검색
• 유사도 탐지 (Similarity detection) — 유사 문서 찾기, 복제 탐지, 콘텐츠 클러스터링
• 분류 (Classification) — 다운스트림 ML 모델의 피처로 임베딩 사용

Pydantic AI는 여러 프로바이더에 걸친 임베딩 생성을 위한 통합 인터페이스를 제공해요.

출처: 문서

본문

빠른 시작 (Quick Start)

Embedder 클래스가 임베딩 생성을 위한 고수준 인터페이스예요:

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')


async def main():
    # Embed a search query
    result = await embedder.embed_query('What is machine learning?')
    print(f'Embedding dimensions: {len(result.embeddings[0])}')
    #> Embedding dimensions: 1536

    # Embed multiple documents at once
    docs = [
        'Machine learning is a subset of AI.',
        'Deep learning uses neural networks.',
        'Python is a programming language.',
    ]
    result = await embedder.embed_documents(docs)
    print(f'Embedded {len(result.embeddings)} documents')
    #> Embedded 3 documents

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

쿼리 vs 문서

일부 임베딩 모델은 쿼리와 문서에 대해 다르게 최적화돼요. 검색 쿼리에는 embed_query()를, 인덱싱하는 콘텐츠에는 embed_documents()를 사용하세요.

임베딩 결과 (Embedding Result)

모든 임베드 메서드는 임베딩과 함께 유용한 메타데이터를 담은 EmbeddingResult를 반환해요.

편의를 위해 임베딩을 인덱스(result[0]) 또는 원래 입력 텍스트(result['Hello world'])로 접근할 수 있어요.

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')


async def main():
    result = await embedder.embed_query('Hello world')

    # Access embeddings - each is a sequence of floats
    embedding = result.embeddings[0]  # By index via .embeddings
    embedding = result[0]  # Or directly via __getitem__
    embedding = result['Hello world']  # Or by original input text
    print(f'Dimensions: {len(embedding)}')
    #> Dimensions: 1536

    # Check usage
    print(f'Tokens used: {result.usage.input_tokens}')
    #> Tokens used: 2

    # Calculate cost (requires `genai-prices` to have pricing data for the model)
    cost = result.cost()
    print(f'Cost: ${cost.total_price:.6f}')
    #> Cost: $0.000000

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

모델 고르기 (Choosing a model)

가장 좋은 임베딩 모델은 언어, 도메인, 지연 시간, 배포, 평가 제약에 따라 달라져요. 이 표를 출발점으로 삼고, 대형 인덱스에 모델을 확정하기 전에 MTEB 리더보드, Hugging Face Hub의 모델 카드, 그리고 자신의 리트리벌 평가를 확인하세요.

원하는 것	예시
관리형 API	openai:text-embedding-3-small(저렴한 기본값), openai:text-embedding-3-large, voyageai:voyage-3.5, cohere:embed-v4.0
API 키 없음·비공개·무료	sentence-transformers:google/embeddinggemma-300m, sentence-transformers:lightonai/DenseOn, sentence-transformers:Qwen/Qwen3-Embedding-0.6B, 또는 다른 Hugging Face 모델
다국어	cohere:embed-multilingual-v3.0, sentence-transformers:jinaai/jina-embeddings-v5-text-small-retrieval, sentence-transformers:Snowflake/snowflake-arctic-embed-l-v2.0
특수 도메인	voyageai:voyage-code-3, voyageai:voyage-law-2, voyageai:voyage-finance-2, sentence-transformers:nomic-ai/CodeRankEmbed, sentence-transformers:TechWolf/JobBERT-v3
이미 보유한 AWS 인프라에서 실행	bedrock:amazon.titan-embed-text-v2:0 또는 bedrock:cohere.embed-v4:0
인덱스 크기 축소	차원 제어가 있는 모델 (Settings 참고)

나중에 모델을 바꾸면 임베딩 공간이 바뀌고 출력 차원도 바뀔 수 있어서, 문서를 다시 임베딩하고 인덱스를 갱신해야 해요. 계속 쓸 모델을 고르거나, 차원 제어를 지원해 모델을 바꾸지 않고 인덱스 크기를 조정할 수 있는 모델을 고르세요.

RAG에 임베딩 사용하기 (Using embeddings for RAG)

Retrieval-Augmented Generation(RAG)에서 임베딩은 더 큰 리트리벌 파이프라인의 한 부분이에요. Pydantic AI는 쿼리·문서 임베딩용 Embedder를, 그리고 검색된 컨텍스트를 에이전트에 주는 도구를 제공해요. RAG 예제는 미리 분할된 데이터를 사용해 벡터 저장, 리트리벌, 에이전트에 검색된 컨텍스트 전달을 시연해요.

프로바이더 관리 파이프라인을 원한다면, 먼저 파일을 프로바이더 관리 스토어에 업로드·임포트한 뒤 그 ID를 FileSearchTool에 넘기세요. 프로바이더가 청킹·임베딩·저장·리트리벌을 처리해요. 지원 프로바이더는 File Search Tool 문서를 참고하세요. 이 섹션의 나머지는 자신의 파이프라인을 구축하는 방법을 다루며, 이 선택들은 애플리케이션별로 남아 있어요.

일반적인 커스텀 RAG 파이프라인은 이렇게 생겼어요:

1. 필요할 때 원본 문서를 청크로 분할.
2. embed_documents()로 청크 임베딩.
3. 각 벡터를 소스 URL·제목·헤딩·페이지 번호·권한 같은 메타데이터와 함께 저장.
4. 사용자의 검색 텍스트를 embed_query()로 임베딩.
5. 벡터 인덱스에서 유사한 청크 검색.
6. 선택적으로 숏리스트를 리랭크.
7. 검색된 텍스트를 도구로 에이전트에 전달.

청킹 (Chunking)

청크는 리트리벌 인덱스가 저장하고 반환하는 원본 텍스트의 단위예요. 모든 문서를 분할할 필요는 없어요. 청킹은 다음 경우에 유용합니다:

• 전체 문서 대신 관련 하위 섹션을 검색하고 싶을 때
• 문서가 임베딩 모델의 최대 입력 길이보다 길 때
• 문서가 너무 많은 독립적인 사실이나 주제를 담아 하나의 임베딩으로 정확히 표현하지 못할 때

이 중 어느 것도 해당하지 않으면 전체 문서를 임베딩하는 게 합리적일 수 있어요. 보편적으로 가장 좋은 청킹 전략이나 청크 크기는 없어요. 올바른 선택은 임베딩 모델, 소스 형식, 도메인, 사용자가 묻는 질문에 따라 달라져요.

전략	유용한 출발점	트레이드오프
헤딩·문단·문장·페이지·레코드 같은 문서 구조	깔끔하고 일관되게 구조화된 소스	저렴하고 자연스러운 경계 보존, 그러나 청크 크기가 고르지 않음
고정 크기 토큰 윈도우	비구조화 텍스트 또는 엄격한 모델 입력 제한	단순하고 예측 가능, 그러나 관련 텍스트를 쪼갤 수 있음. 오버랩은 경계 컨텍스트를 보존하되 인덱스가 커지고 중복 결과 발생
시맨틱 분할	레이아웃보다 주제 경계가 중요한 소스	관련 아이디어를 함께 유지 가능, 그러나 수집 작업이 늘고 모델·임계값에 의존
LLM 지원 분할	해석이 필요한 불규칙하거나 도메인 특화 소스	유연하지만, 지연·비용이 가장 많이 늘고 선택한 경계와 소스 커버리지 검증 필요

첫 구현에서는 자연스러운 문서 경계에 토큰 제한을 걸고, 모든 청크에 전체 문서 소스 메타데이터와 링크·식별자를 유지한 뒤, 더 비싼 전략을 추가하기 전에 평가하세요. 청크는 집중된 질문에 답하기에 충분히 크면서도 무관한 컨텍스트를 피할 만큼 작아야 해요.

흔한 전략의 비교와 재현 가능한 평가 코드는 Chroma의 청킹 전략 평가, end-to-end RAG 평가 워크플로우는 Hugging Face의 RAG 평가 쿡북을 참고하세요.

벡터 저장 (Vector storage)

Pydantic AI는 벡터 데이터베이스를 규정하지 않아요. 인덱스 크기, 쿼리·수집 볼륨, 메타데이터 필터링·권한 요구, 하이브리드 키워드 검색 니즈, 가용성 요구, 운영 중인 인프라에 따라 고르세요:

• FAISS, LanceDB, 또는 벡터 확장이 있는 SQLite 같은 로컬 인덱스·내장 DB는 프로토타입과 작은 인덱스에 적합할 수 있음
• pgvector가 있는 PostgreSQL은 이미 PostgreSQL을 쓰는 애플리케이션에 적합할 수 있음
• 관리형 벡터·검색 서비스는 호스팅 확장과 운영 지원이 필요한 애플리케이션에 적합할 수 있음

인덱스의 벡터 차원은 임베딩 출력 차원과 일치해야 해요. 인덱싱과 쿼리에서 호환 가능한 모델 구성을 사용하세요. 차원만 일치한다고 벡터가 호환되진 않아요. 임베딩 구성을 바꾸면 문서를 다시 임베딩하고 저장 계층이 요구하는 대로 인덱스를 갱신·재구축하세요.

Hugging Face의 고급 RAG 쿡북은 벡터 인덱스, 유사도 선택, 리랭킹, 그 밖의 리트리벌 트레이드오프를 소개해요.

평가 (Evaluation)

청킹, 임베딩 모델, 유사도 메트릭, 검색된 청크 수, 리랭킹은 모두 상호작용해요. 공개 벤치마크는 후보를 좁힐 수 있지만, 어떤 완전한 파이프라인이 당신의 애플리케이션에 가장 잘 맞는지 증명할 수는 없어요. 프로덕션을 대표하는 쿼리와 문서로 변경을 비교하고, 관련 텍스트가 검색되는지와 무관한 텍스트가 에이전트 컨텍스트에서 제외되는지를 모두 확인하세요.

대표 쿼리 데이터셋 전반의 리트리벌 품질을 추적하려면 Pydantic Evals를 사용하세요. Hugging Face의 RAG 평가 쿡북은 합성 평가 세트를 만들고 생성된 답변을 평가하는 방법을 보여줘요.

프로바이더 (Providers)

OpenAI

OpenAIEmbeddingModel은 OpenAI의 embeddings API 및 OpenAI 호환 프로바이더와 함께 동작해요.

설치

pydantic-ai를 설치하거나, openai 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[openai]"

uv add "pydantic-ai-slim[openai]"

구성

platform.openai.com에 가서 API 키를 생성한 뒤 환경 변수로 설정하세요:

export OPENAI_API_KEY='your-api-key'

그러면 모델을 사용할 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 1536

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

사용 가능한 모델은 OpenAI의 임베딩 모델을 참고하세요.

차원 제어 (Dimension Control)

OpenAI의 text-embedding-3-* 모델은 dimensions 설정으로 차원 축소를 지원해요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings import EmbeddingSettings

embedder = Embedder(
    'openai:text-embedding-3-small',
    settings=EmbeddingSettings(dimensions=256),
)


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 256

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

OpenAI 호환 프로바이더

OpenAIEmbeddingModel은 OpenAIChatModel과 같은 프로바이더 시스템을 사용하므로, 어떤 OpenAI 호환 프로바이더와든 쓸 수 있어요:

# Using Azure OpenAI
from openai import AsyncAzureOpenAI

from pydantic_ai import Embedder
from pydantic_ai.embeddings.openai import OpenAIEmbeddingModel
from pydantic_ai.providers.openai import OpenAIProvider

azure_client = AsyncAzureOpenAI(
    azure_endpoint='https://your-resource.openai.azure.com',
    api_version='2024-02-01',
    api_key='your-azure-key',
)
model = OpenAIEmbeddingModel(
    'text-embedding-3-small',
    provider=OpenAIProvider(openai_client=azure_client),
)
embedder = Embedder(model)


# Using any OpenAI-compatible API
model = OpenAIEmbeddingModel(
    'your-model-name',
    provider=OpenAIProvider(
        base_url='https://your-provider.com/v1',
        api_key='your-api-key',
    ),
)
embedder = Embedder(model)

전용 프로바이더 클래스가 있는 프로바이더(OllamaProvider나 AzureProvider 같은)에서는 축약 문법을 쓸 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('azure:text-embedding-3-small')
embedder = Embedder('ollama:nomic-embed-text')
embedder = Embedder('vllm:intfloat/e5-mistral-7b-instruct')

vllm: 축약은 VLLM_BASE_URL을, 인증 서버에서는 VLLM_API_KEY를 사용해요. 서버는 vLLM이 지원하는 임베딩 모델을 실행 중이어야 합니다.

지원 프로바이더 전체 목록은 OpenAI 호환 모델을 참고하세요.

Google

GoogleEmbeddingModel은 Gemini API(Google AI Studio) 또는 Google Cloud(구 Vertex AI)를 통해 Google 임베딩 모델과 함께 동작해요.

설치

pydantic-ai를 설치하거나, google 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[google]"

uv add "pydantic-ai-slim[google]"

구성

Gemini API에 GoogleEmbeddingModel을 쓰려면 aistudio.google.com에 가서 API 키를 생성하세요. 키가 있으면 환경 변수로 설정합니다:

export GOOGLE_API_KEY='your-api-key'

그러면 모델을 사용할 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('google:gemini-embedding-001')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 3072

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

사용 가능한 모델은 Google Embeddings 문서를 참고하세요.

Google Cloud

Gemini API 대신 Google Cloud(구 Vertex AI)로 Google 임베딩 모델을 쓰려면 google-cloud: 프로바이더 프리픽스를 사용하세요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.google import GoogleEmbeddingModel
from pydantic_ai.providers.google_cloud import GoogleCloudProvider

# Using provider prefix
embedder = Embedder('google-cloud:gemini-embedding-001')

# Or with explicit provider configuration
model = GoogleEmbeddingModel(
    'gemini-embedding-001',
    provider=GoogleCloudProvider(project='my-project', location='us-central1'),
)
embedder = Embedder(model)

애플리케이션 기본 자격 증명, 서비스 계정, API 키를 포함한 Google Cloud 인증 옵션에 대한 자세한 내용은 Google 프로바이더 문서를 참고하세요.

차원 제어

Google 임베딩 모델은 dimensions 설정으로 차원 축소를 지원해요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings import EmbeddingSettings

embedder = Embedder(
    'google:gemini-embedding-001',
    settings=EmbeddingSettings(dimensions=768),
)


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 768

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

작업 조건화 (Task Conditioning)

gemini-embedding-2는 다른 Google 모델이 쓰는 google_task_type 필드 대신, 입력 텍스트 앞에 짧은 작업 지시문을 붙여 임베딩할 작업에 조건화돼요. Pydantic AI는 google_task 설정을 통해 이 프리픽스를 만들어 줍니다:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.google import GoogleEmbeddingSettings

embedder = Embedder(
    'google:gemini-embedding-2',
    settings=GoogleEmbeddingSettings(google_task='question answering'),
)

google_task는 Google API의 작업 이름('search result', 'question answering', 'fact checking', 'code retrieval', 'classification', 'clustering', 'sentence similarity')을 받아요. 리트리벌형(비대칭) 작업에서는 쿼리와 문서가 다르게 프리픽스되므로 같은 작업이 쌍의 양쪽에 적용되고, 나머지(대칭) 작업은 두 입력에 같은 방식으로 프리픽스돼요.

google_task를 설정하지 않으면 gemini-embedding-2는 'search result'로 조건화돼요. Google이 이 모델에 권장하고 원본 텍스트를 임베딩하는 것보다 리트리벌 성능이 좋아서 조건화가 기본 켜져 있어요. 텍스트를 그대로 임베딩하고 싶으면 google_task='raw'를 넘기세요.

google_task는 gemini-embedding-2에만 적용되고, 다른 모델에서는 경고와 함께 무시돼요(그 모델들은 google_task_type으로 조건화함). 반대로 google_task_type은 gemini-embedding-2에서 텍스트 프리픽스로 조건화되므로 무시됩니다.

Google 특정 설정

Google 모델은 GoogleEmbeddingSettings로 추가 설정을 지원해요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.google import GoogleEmbeddingSettings

embedder = Embedder(
    'google:gemini-embedding-001',
    settings=GoogleEmbeddingSettings(
        dimensions=768,
        google_task_type='SEMANTIC_SIMILARITY',  # Optimize for similarity comparison
    ),
)

사용 가능한 작업 타입은 Google의 작업 타입 문서를 참고하세요. 기본적으로 embed_query()는 RETRIEVAL_QUERY, embed_documents()는 RETRIEVAL_DOCUMENT를 사용해요.

Cohere

CohereEmbeddingModel은 Cohere의 임베딩 모델에 접근을 제공하며, 다국어 지원과 다양한 모델 크기를 갖추고 있어요.

설치

pydantic-ai를 설치하거나, cohere 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[cohere]"

uv add "pydantic-ai-slim[cohere]"

구성

CohereEmbeddingModel을 쓰려면 dashboard.cohere.com/api-keys에 가서 API 키를 생성하세요. 키가 있으면 환경 변수로 설정합니다:

export CO_API_KEY='your-api-key'

그러면 모델을 사용할 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('cohere:embed-v4.0')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 1024

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

사용 가능한 모델은 Cohere Embed 문서를 참고하세요.

Cohere 특정 설정

Cohere 모델은 CohereEmbeddingSettings로 추가 설정을 지원해요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.cohere import CohereEmbeddingSettings

embedder = Embedder(
    'cohere:embed-v4.0',
    settings=CohereEmbeddingSettings(
        dimensions=512,
        cohere_truncate='END',  # Truncate long inputs instead of erroring
        cohere_max_tokens=256,  # Limit tokens per input
    ),
)

VoyageAI

VoyageAIEmbeddingModel은 코드·금융·법률 도메인 특화 모델로 리트리벌에 최적화된 VoyageAI 임베딩 모델에 접근을 제공해요.

설치

voyageai 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[voyageai]"

uv add "pydantic-ai-slim[voyageai]"

구성

VoyageAIEmbeddingModel을 쓰려면 dash.voyageai.com에서 API 키를 생성하세요. 키가 있으면 환경 변수로 설정합니다:

export VOYAGE_API_KEY='your-api-key'

그러면 모델을 사용할 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('voyageai:voyage-3.5')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 1024

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

사용 가능한 모델은 VoyageAI Embeddings 문서를 참고하세요.

VoyageAI 특정 설정

VoyageAI 모델은 VoyageAIEmbeddingSettings로 추가 설정을 지원해요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.voyageai import VoyageAIEmbeddingSettings

embedder = Embedder(
    'voyageai:voyage-3.5',
    settings=VoyageAIEmbeddingSettings(
        dimensions=512,  # Reduce output dimensions
        voyageai_input_type='document',  # Override input type for all requests
    ),
)

Bedrock

BedrockEmbeddingModel은 AWS Bedrock을 통해 Amazon Titan, Cohere, Amazon Nova 모델을 포함한 임베딩 모델에 접근을 제공해요.

설치

pydantic-ai를 설치하거나, bedrock 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[bedrock]"

uv add "pydantic-ai-slim[bedrock]"

구성

AWS Bedrock 인증은 표준 AWS 자격 증명을 사용해요. 환경 변수, AWS 자격 증명 파일, IAM 역할로 자격 증명을 구성하는 방법은 Bedrock 프로바이더 문서를 참고하세요.

AWS 계정이 사용하려는 Bedrock 임베딩 모델에 접근할 수 있는지 확인하세요. 자세한 내용은 AWS Bedrock 모델 접근을 참고하세요.

기본 사용법

from pydantic_ai import Embedder

# Using Amazon Titan
embedder = Embedder('bedrock:amazon.titan-embed-text-v2:0')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 1024

(이 예제는 AWS 자격 증명이 구성되어 있어야 합니다)

지원 모델

Bedrock은 세 가지 임베딩 모델 패밀리를 지원해요. 사용 가능한 모델 전체 목록은 AWS Bedrock 문서를 참고하세요.

Amazon Titan:

• amazon.titan-embed-text-v1 — 1536 차원(고정), 8K 토큰
• amazon.titan-embed-text-v2:0 — 256/384/1024 차원(구성 가능, 기본: 1024), 8K 토큰

Cohere Embed:

• cohere.embed-english-v3 — 영어 전용, 1024 차원(고정), 512 토큰
• cohere.embed-multilingual-v3 — 다국어, 1024 차원(고정), 512 토큰
• cohere.embed-v4:0 — 256/512/1024/1536 차원(구성 가능, 기본: 1536), 128K 토큰

Amazon Nova:

• amazon.nova-2-multimodal-embeddings-v1:0 — 256/384/1024/3072 차원(구성 가능, 기본: 3072), 8K 토큰

Titan 특정 설정

Titan v2는 bedrock_titan_normalize(기본: True)로 직접 유사도 계산을 위한 벡터 정규화를 지원해요. Titan v1은 이 설정을 지원하지 않아요.

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingSettings

embedder = Embedder(
    'bedrock:amazon.titan-embed-text-v2:0',
    settings=BedrockEmbeddingSettings(
        dimensions=512,
        bedrock_titan_normalize=True,
    ),
)

참고

Titan 모델은 truncate 설정을 지원하지 않아요. dimensions 설정은 Titan v2만 지원합니다.

Cohere 특정 설정

Bedrock의 Cohere 모델은 BedrockEmbeddingSettings로 추가 설정을 지원해요:

• bedrock_cohere_input_type — 기본적으로 embed_query()는 'search_query', embed_documents()는 'search_document'를 사용해요. 'classification' 또는 'clustering'도 받아요.
• bedrock_cohere_truncate — 세밀한 절단 제어: 'NONE'(기본, 오버플로 시 오류), 'START', 'END'. 기본 truncate 설정을 오버라이드해요.
• bedrock_cohere_max_tokens — 입력당 토큰 제한(기본: 128000). Cohere v4만 지원.

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingSettings

embedder = Embedder(
    'bedrock:cohere.embed-v4:0',
    settings=BedrockEmbeddingSettings(
        dimensions=512,
        bedrock_cohere_max_tokens=1000,
        bedrock_cohere_truncate='END',
    ),
)

참고

dimensions와 bedrock_cohere_max_tokens 설정은 Cohere v4만 지원해요. Cohere v3 모델은 고정 1024 차원입니다.

Nova 특정 설정

Bedrock의 Nova 모델은 BedrockEmbeddingSettings로 추가 설정을 지원해요:

• bedrock_nova_truncate — 세밀한 절단 제어: 'NONE'(기본, 오버플로 시 오류), 'START', 'END'. 기본 truncate 설정을 오버라이드해요.
• bedrock_nova_embedding_purpose — 기본적으로 embed_query()는 'GENERIC_RETRIEVAL', embed_documents()는 'GENERIC_INDEX'를 사용해요. 'TEXT_RETRIEVAL', 'CLASSIFICATION', 'CLUSTERING'도 받아요.

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingSettings

embedder = Embedder(
    'bedrock:amazon.nova-2-multimodal-embeddings-v1:0',
    settings=BedrockEmbeddingSettings(
        dimensions=1024,
        bedrock_nova_embedding_purpose='TEXT_RETRIEVAL',
        truncate=True,
    ),
)

동시성 설정 (Concurrency Settings)

배치 임베딩을 지원하지 않는 모델(Titan, Nova)은 각 입력 텍스트에 대해 개별 API 요청을 만들어요. 기본적으로 이 요청들은 최대 5개의 병렬 요청으로 동시에 실행됩니다.

bedrock_max_concurrency 설정으로 조정할 수 있어요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingSettings

# Increase concurrency for faster throughput
embedder = Embedder(
    'bedrock:amazon.titan-embed-text-v2:0',
    settings=BedrockEmbeddingSettings(bedrock_max_concurrency=10),
)

# Or reduce concurrency to avoid rate limits
embedder = Embedder(
    'bedrock:amazon.nova-2-multimodal-embeddings-v1:0',
    settings=BedrockEmbeddingSettings(bedrock_max_concurrency=2),
)

지역 프리픽스 (Regional Prefixes)

Bedrock은 us., eu., apac. 같은 지리 프리픽스로 크로스-리전 추론을 지원해요:

from pydantic_ai import Embedder

embedder = Embedder('bedrock:us.amazon.titan-embed-text-v2:0')

AWS 애플리케이션 추론 프로필 사용

모델 능력 감지를 위해 베이스 모델 이름은 유지하면서 추론 프로필로 요청을 라우팅하려면 bedrock_inference_profile을 설정하세요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingModel
from pydantic_ai.providers.bedrock import BedrockProvider

provider = BedrockProvider(region_name='us-east-1')

model = BedrockEmbeddingModel(
    'amazon.titan-embed-text-v2:0',
    provider=provider,
    settings={
        'bedrock_inference_profile': 'arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/my-embed-profile',
    },
)
embedder = Embedder(model)

커스텀 프로바이더 사용

명시적 자격 증명이나 커스텀 boto3 클라이언트 같은 고급 구성에는 BedrockProvider를 직접 만들 수 있어요. 자세한 내용은 Bedrock 프로바이더 문서를 참고하세요.

from pydantic_ai import Embedder
from pydantic_ai.embeddings.bedrock import BedrockEmbeddingModel
from pydantic_ai.providers.bedrock import BedrockProvider

provider = BedrockProvider(
    region_name='us-west-2',
    aws_access_key_id='your-access-key',
    aws_secret_access_key='your-secret-key',
)

model = BedrockEmbeddingModel('amazon.titan-embed-text-v2:0', provider=provider)
embedder = Embedder(model)

토큰 계수

Bedrock 임베딩 모델은 count_tokens() 메서드를 지원하지 않아요. AWS Bedrock의 토큰 계수 API는 텍스트 생성 모델(Claude, Llama 등)에서만 동작하고 임베딩 모델에서는 동작하지 않기 때문이에요. count_tokens()를 호출하면 NotImplementedError가 발생합니다.

Sentence Transformers (로컬)

SentenceTransformerEmbeddingModel은 Sentence Transformers 라이브러리로 로컬에서 임베딩을 실행해요. API 호출 없이 Hugging Face의 수천 개 임베딩 모델에 접근할 수 있게 해 줍니다. 이상적인 경우:

• 프라이버시 — 데이터가 인프라 밖으로 나가지 않음
• 비용 — 대용량 워크로드에 API 비용 없음
• 오프라인 사용 — 모델 다운로드 후 인터넷 연결 불필요
• 특수 도메인·언어 — MTEB 리더보드에서 코드·다국어·의생명·법률용으로 훈련된 모델 선택

설치

sentence-transformers 선택 그룹으로 pydantic-ai-slim을 설치해야 합니다:

pip install "pydantic-ai-slim[sentence-transformers]"

uv add "pydantic-ai-slim[sentence-transformers]"

사용법

from pydantic_ai import Embedder

# Model is downloaded from Hugging Face on first use
embedder = Embedder('sentence-transformers:lightonai/DenseOn')


async def main():
    result = await embedder.embed_query('Hello world')
    print(len(result.embeddings[0]))
    #> 768

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

lightonai/DenseOn은 쿼리와 문서를 비대칭으로 인코딩하는 강력한 최신 149M 파라미터 범용 모델이에요. embed_query()와 embed_documents()가 모델의 query:/document: 프롬프트를 자동 적용해요. 더 많은 옵션은 Sentence Transformers 사전훈련 모델 문서와 MTEB 리더보드, 그리고 위의 모델 고르기를 참고하세요.

디바이스 선택

추론에 사용할 디바이스를 제어할 수 있어요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings.sentence_transformers import (
    SentenceTransformersEmbeddingSettings,
)

embedder = Embedder(
    'sentence-transformers:sentence-transformers/all-MiniLM-L6-v2',
    settings=SentenceTransformersEmbeddingSettings(
        sentence_transformers_device='cuda',  # Use GPU
        sentence_transformers_normalize_embeddings=True,  # L2 normalize
    ),
)

기존 모델 인스턴스 사용

모델 초기화를 더 제어해야 한다면:

from sentence_transformers import SentenceTransformer

from pydantic_ai import Embedder
from pydantic_ai.embeddings.sentence_transformers import (
    SentenceTransformerEmbeddingModel,
)

# Create and configure the model yourself
st_model = SentenceTransformer('microsoft/harrier-oss-v1-270m', device='cpu')

# Wrap it for use with Pydantic AI
model = SentenceTransformerEmbeddingModel(st_model)
embedder = Embedder(model)

설정 (Settings)

EmbeddingSettings는 프로바이더 전반에서 동작하는 공통 구성 옵션을 제공해요:

• dimensions: 출력 임베딩 차원 축소(OpenAI, Google, Cohere, Bedrock, VoyageAI 지원)
• truncate: True일 때 모델 컨텍스트 길이를 초과하는 입력 텍스트를 오류 대신 절단(Cohere, Bedrock, VoyageAI 지원)

설정은 모델, 임베더 수준(모든 호출에 적용), 호출별로 지정할 수 있어요. 이 순서로 병합되는데, 나중 설정이 같은 키의 이전 값을 오버라이드하고, 이전 레이어에만 설정된 값은 유지돼요. 아래 예제는 한 호출에 대해 임베더 기본값을 오버라이드합니다:

from pydantic_ai import Embedder
from pydantic_ai.embeddings import EmbeddingSettings

# Default settings for all calls
embedder = Embedder(
    'openai:text-embedding-3-small',
    settings=EmbeddingSettings(dimensions=512),
)


async def main():
    # Override for a specific call
    result = await embedder.embed_query(
        'Hello world',
        settings=EmbeddingSettings(dimensions=256),
    )
    print(len(result.embeddings[0]))
    #> 256

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

토큰 계수 (Token Counting)

임베딩 전에 토큰 수를 확인해 모델 한도를 초과하지 않게 할 수 있어요:

from pydantic_ai import Embedder

embedder = Embedder('openai:text-embedding-3-small')


async def main():
    text = 'Hello world, this is a test.'

    # Count tokens in text
    token_count = await embedder.count_tokens(text)
    print(f'Tokens: {token_count}')
    #> Tokens: 7

    # Check model's maximum input tokens (returns None if unknown)
    max_tokens = await embedder.max_input_tokens()
    print(f'Max tokens: {max_tokens}')
    #> Max tokens: 1024

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

테스팅 (Testing)

API 호출 없이 테스트하려면 TestEmbeddingModel을 사용하세요:

from pydantic_ai import Embedder
from pydantic_ai.embeddings import TestEmbeddingModel


async def test_my_rag_system():
    embedder = Embedder('openai:text-embedding-3-small')
    test_model = TestEmbeddingModel()

    with embedder.override(model=test_model):
        result = await embedder.embed_query('test query')

        # TestEmbeddingModel returns deterministic embeddings
        assert result.embeddings[0] == [1.0] * 8

        # Check what settings were used
        assert test_model.last_settings is not None

ALLOW_MODEL_REQUESTS를 False로 설정하면 임베딩 요청도 차단되므로, 오버라이드하지 않은 임베더는 조용히 프로바이더를 호출하는 대신 오류를 던져요. TestEmbeddingModel과 SentenceTransformerEmbeddingModel은 둘 다 프로바이더에 닿지 않으므로 영향받지 않아요.

이것은 count_tokens()도 포함하지만, 토큰화가 서버 측에서 일어날 때만 그래요. Google과 Cohere는 API 호출로 토큰을 세므로 차단되지만, OpenAI는 tiktoken으로 로컬 토큰화하므로 토큰 계수가 차단되지 않아요.

계측 (Instrumentation)

디버깅과 모니터링을 위해 OpenTelemetry 계측을 활성화할 수 있어요:

import logfire

from pydantic_ai import Embedder

logfire.configure()

# Instrument a specific embedder
embedder = Embedder('openai:text-embedding-3-small', instrument=True)

# Or instrument all embedders globally
Embedder.instrument_all()

Pydantic AI에서 Logfire를 쓰는 자세한 내용은 디버깅·모니터링 가이드를 참고하세요.

리랭커를 이용한 2단계 리트리벌 (Two-stage retrieval with rerankers)

고품질 리트리벌을 위한 흔한 패턴은 2단계예요. 먼저 임베딩 모델로 광범위한 후보 숏리스트를 저렴하게 뽑고, 그다음 크로스-인코더 리랭커로 각 후보를 쿼리 대비 더 정밀하게 점수화해요. 크로스-인코더는 쿼리와 문서를 함께 읽으므로 임베딩 조회보다 느리지만 훨씬 정확해서, top-100 리콜 목록을 LLM에 실제로 넘기는 top-5 결과로 좁히는 데 이상적이에요.

Pydantic AI는 리랭커 프로바이더 클래스를 제공하지 않으므로 직접 가져와야 해요. 가장 흔한 로컬 옵션은 sentence-transformers의 CrossEncoder예요:

import asyncio
from functools import cache

from sentence_transformers import CrossEncoder


@cache
def get_reranker() -> CrossEncoder:
    # Loaded lazily on first call, then reused.
    return CrossEncoder('cross-encoder/ms-marco-MiniLM-L6-v2')


async def rerank(query: str, candidates: list[str], top_k: int = 3) -> list[str]:
    """Rerank retrieval candidates by relevance to `query`."""
    reranker = get_reranker()
    # CrossEncoder.rank is blocking, so run it off the event loop.
    ranked = await asyncio.to_thread(
        reranker.rank, query, candidates, top_k=top_k, return_documents=True
    )
    return [item['text'] for item in ranked]

결과를 LLM에 넘기기 전에 벡터 검색(RAG 예제의 retrieve 도구 같은)이 반환한 후보에 rerank()를 호출하세요.

관리형 리랭커 대안

로컬에서 리랭커를 실행하고 싶지 않다면, Cohere Rerank, VoyageAI Rerank, Jina Rerank 같은 호스팅 리랭커를 제공하는 프로바이더가 여러 곳 있어요. 위의 rerank()와 같은 시그니처의 헬퍼 함수에서 그 HTTP 클라이언트나 SDK를 호출하세요.

리트리브-앤-리랭크 파이프라인에 대한 더 많은 배경은 Hugging Face의 고급 RAG 쿡북을 참고하세요. 오픈소스 임베딩·리랭커 모델을 직접 서빙하려면 Hugging Face Text Embeddings Inference와 그 지원 리랭커를 참고하세요.

커스텀 임베딩 모델 구축 (Building Custom Embedding Models)

커스텀 임베딩 프로바이더를 통합하려면 EmbeddingModel을 서브클래스하세요:

from collections.abc import Sequence

from pydantic_ai.embeddings import EmbeddingModel, EmbeddingResult, EmbeddingSettings
from pydantic_ai.embeddings.result import EmbedInputType


class MyCustomEmbeddingModel(EmbeddingModel):
    @property
    def model_name(self) -> str:
        return 'my-custom-model'

    @property
    def system(self) -> str:
        return 'my-provider'

    async def embed(
        self,
        inputs: str | Sequence[str],
        *,
        input_type: EmbedInputType,
        settings: EmbeddingSettings | None = None,
    ) -> EmbeddingResult:
        inputs, settings = self.prepare_embed(inputs, settings)

        # Call your embedding API here
        embeddings = [[0.1, 0.2, 0.3] for _ in inputs]  # Placeholder

        return EmbeddingResult(
            embeddings=embeddings,
            inputs=inputs,
            input_type=input_type,
            model_name=self.model_name,
            provider_name=self.system,
        )

캐싱이나 로깅 같은 커스텀 동작을 추가하려고 기존 모델을 래핑하고 싶다면 WrapperEmbeddingModel을 사용하세요.

더 알아보기 (Learn more)

• RAG 예제 — 벡터 저장·리트리벌·에이전트 전달.
• File Search Tool 문서 — 프로바이더 관리 파이프라인.
• 고급 RAG 쿡북 — 리트리벌 트레이드오프.