증분 임베딩 업데이트

증분 임베딩 업데이트 (Incremental Embedding Updates)

소요 시간: 25분	난이도: 입문(Beginner)	결과물: GitHub

Qdrant 문서는 GitHub에 공개되어 있는데, 주로 코드 조각과 그림이 들어간 마크다운 페이지로 구성돼 있어요. 발전하는 제품의 문서가 다 그렇듯, 문서도 정적인 게 아니에요. 마크다운의 원본 데이터는 시간이 지나면서 바뀌고, 우리 문서를 검색하는 사용자들은 항상 최신 상태를 기대하죠. 문서 검색이 우리처럼 벡터를 쓴다면, 그 기대를 충족하려면 추가적인 설정과 유지보수가 필요해요.

출처: Qdrant 공식문서 - Incremental Embedding Updates

벡터 <-> 원본 데이터

벡터는 원본 데이터를 변환한 결과예요. 그런데 원본 데이터가 바뀐다고 해서 이 변환이 저절로 일어나지는 않아요. 벡터를 적극적으로 갱신하지 않으면, 문서 검색은 더 이상 존재하지 않는 텍스트의 임베딩을 상대로 실행될 거예요. 원본 데이터의 변화와 벡터를 동기화하는 재-임베딩 과정이 필요한 이유죠.

이 튜토리얼은, 처음부터 세팅해 두면 텍스트 데이터의 변화를 감지해서 증분 임베딩 업데이트를 실행하는 간단한 파이프라인을 제공해요. Qdrant 문서 청크의 완전하고 최신 목록을 Qdrant 컬렉션과 대조해서 맞춰 주죠. 실행마다 다음을 수행해요:

바뀌지 않은 청크는 건드리지 않고,
바뀐 텍스트를 다시 임베딩하고,
텍스트가 위치만 바뀐 경우에는 기존 벡터를 재사용하고,
새 텍스트를 추가하고,
원본 목록에 없는 텍스트는 삭제해요.

이 패턴은 청킹이 결정적이고 현재 원본을 나열하는 비용이 저렴할 때 적용할 수 있어요.

이 튜토리얼과 함께 제공되는 노트북이 있어요.

사전 준비 (Prerequisites)

원하는 Qdrant 클라이언트를 설치하세요.

우리는 Qdrant Cloud와 그 무료 임베딩 추론(Free Embedding Inference)을 사용할 거예요. 무료 티어 Qdrant Cloud 클러스터를 만들고, 환경 변수에 QDRANT_URL과 QDRANT_API_KEY를 설정해 주세요.

from qdrant_client import QdrantClient, models

# Replace url and api_key with your own from https://cloud.qdrant.io
client = QdrantClient(
    url="https://xyz-example.qdrant.io:6333",
    api_key="<your-api-key>",
    cloud_inference=True,
)

import { QdrantClient, Schemas } from "@qdrant/js-client-rest";
// Replace url and apiKey with your own from https://cloud.qdrant.io
const client = new QdrantClient({
  url: "https://xyz-example.qdrant.io:6333",
  apiKey: "<your-api-key>",
});

// Replace the URL and API key with your own from https://cloud.qdrant.io
let client = Qdrant::from_url("https://xyz-example.qdrant.io:6334")
    .api_key("<your-api-key>")
    .build()?;

// Replace the host and API key with your own from https://cloud.qdrant.io
static final QdrantClient client = new QdrantClient(
    QdrantGrpcClient.newBuilder("xyz-example.qdrant.io", 6334, true)
        .withApiKey("<your-api-key>")
        .build());

// Replace the host and API key with your own from https://cloud.qdrant.io
var client = new QdrantClient(
    host: "xyz-example.qdrant.io",
    https: true,
    apiKey: "<your-api-key>",
);

// Replace the host and API key with your own from https://cloud.qdrant.io
client, err := qdrant.NewClient(&qdrant.Config{
    Host:   "xyz-example.qdrant.io",
    APIKey: "<your-api-key>",
    UseTLS: true,
})

데이터: Qdrant 문서

운영 튜토리얼 탭을 잠깐 볼게요. 실제로 일어날 수 있는 변화의 예시 하나를 들어볼게요. 이 튜토리얼이 그 탭의 일부가 되었다면, 문서 검색에 쓰는 벡터 컬렉션도 그에 맞춰 갱신해야 하겠죠.

간단한 문서 계층 구조를 생각해 봐요:

하나의 url 뒤에 페이지 하나가 있어요: https://qdrant.tech/documentation/tutorials-operations/secure-qdrant
페이지는 섹션들로 구성돼요. 예를 들어 "Step 2: Enable TLS" 섹션처럼요. 섹션은 제목 텍스트에서 생성된 anchor로 표시돼요. "Step 2: Enable TLS" -> section_url의 "#step-2-enable-tls" 부분이 되는 거죠.

이 계층 구조로 문서를 나눠 볼게요. 어떤 섹션은 임베딩 모델의 컨텍스트 윈도우 한계(모델이 표현할 수 있는 텍스트의 크기)에 맞지 않을 수도 있어요. 그러면 0, 1, 2...로 번호를 매긴 청크로 나눠요. 최소한의 계층 인식(hierarchy awareness)을 위해, 각 청크는 섹션 제목을 앞에 붙여서 보관해요.

page https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/ (url)
├── section #prerequisites (anchor)
│   └── chunk_num 0 "Prerequisites - Docker and Docker Compose installed..."
├── section #secure-a-self-hosted-qdrant-instance (anchor)
│   ├── chunk_num 0 "Secure a Self-Hosted Qdrant Instance | Time: 45 min..."
│   └── chunk_num 1 "Secure a Self-Hosted Qdrant Instance > Qdrant Cloud..."
├── section #step-1-start-an-unsecured-instance (anchor)
│   └── chunk_num 0 "Step 1: Start an Unsecured Instance Start Qdrant..."
├── section #step-2-enable-tls (anchor)
│   └── chunk_num 0 "Step 2: Enable TLS Unencrypted connections allow..."
└── ...

즉, 페이지 하나는 {url, anchor, chunk_num, text} 형태의 청크 집합이 돼요. 벡터 하나 = 섹션 청크 하나예요.

이 튜토리얼에서 쓰는 CHUNKS 목록: operations 탭에서 가져온 실제 튜토리얼 세 개를 청킹한 것

CHUNKS = [
    # three tutorials: secure-qdrant, migration, time-based-sharding
    {
        "url": "https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/",
        "anchor": "prerequisites",
        "chunk_num": 0,
        "text": "Prerequisites - Docker and Docker Compose installed - `curl` available in your terminal - mkcert for generating a local self-signed certificate (installation instructions) - TLS requires Qdrant 1.2 or later, API key authentication requires Qdrant 1.2 or later, and granular access API keys (JWT) require Qdrant 1.9 or later. This tutorial uses the latest Qdrant image, which includes all these features. ---",
    },
    {
        "url": "https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/",
        "anchor": "secure-a-self-hosted-qdrant-instance",
        "chunk_num": 0,
        "text": "Secure a Self-Hosted Qdrant Instance | Time: 45 min | Level: Intermediate | ...",
    },
    # ... full list in the ipynb
]

각 청크에는 어떤 텍스트 정규화(normalization) 파이프라인이 마련되어 있다고 가정해요. 이유가 있어요:

텍스트의 노이즈는 임베딩 품질을 떨어뜨려요
노이즈는 필요 없는데 재-임베딩을 하게 만들어 비용이 들어요 (예: 누군가 끝에 공백을 하나 추가한 경우)

normalize(text):
  - remove invisible characters (zero-width spaces, byte-order mark, soft hyphen)
  - collapse any whitespace run into a single space
  - ...

컬렉션 구성하기

청크를 저장할 컬렉션을 구성해 볼게요.

sentence-transformers/all-MiniLM-L6-v2를 사용할 거예요. Qdrant Cloud Inference의 무료 임베딩 모델 중 하나예요. 출력 차원은 384이고, 컨텍스트 윈도우는 256 토큰이에요. 그래서 위에서 긴 섹션을 청킹한 거예요. 윈도우를 초과하는 입력은 조용히 잘려 나가거든요.

컬렉션 메타데이터

서로 다른 임베딩 모델이 만든 벡터, 또는 같은 모델이지만 다르게 준비된 텍스트로 만든 벡터는 한 컬렉션에 섞이면 안 돼요. 검색 품질이 떨어지고, 왜 그런지 원인을 찾기도 어려워지거든요. 간단한 안전장치가 있어요. 어떤 모델과 어떤 파이프라인 버전이 데이터 포인트를 만들었는지 컬렉션 메타데이터에 저장하는 거예요. 그리고 둘 중 하나라도 바뀌면 전체 컬렉션 재-임베딩을 실행하면 돼요.

MODEL = "sentence-transformers/all-MiniLM-L6-v2"
PIPELINE = "docs-prep-pipeline-v1"
COLLECTION = "docs-sync-tutorial"

client.create_collection(
    COLLECTION,
    vectors_config=models.VectorParams(
        size=384,  # all-MiniLM-L6-v2 output dimension
        distance=models.Distance.COSINE,
    ),
    metadata={
        "embedding_model": MODEL,
        "pipeline_version": PIPELINE,
    },
)

const MODEL = "sentence-transformers/all-MiniLM-L6-v2";
const PIPELINE = "docs-prep-pipeline-v1";
const COLLECTION = "docs-sync-tutorial";

await client.createCollection(COLLECTION, {
  vectors: {
    size: 384, // all-MiniLM-L6-v2 output dimension
    distance: "Cosine",
  },
});
await client.updateCollection(COLLECTION, {
  metadata: {
    embedding_model: MODEL,
    pipeline_version: PIPELINE,
  },
});

const MODEL: &str = "sentence-transformers/all-MiniLM-L6-v2";
const PIPELINE: &str = "docs-prep-pipeline-v1";
const COLLECTION: &str = "docs-sync-tutorial";

let mut metadata: HashMap<String, Value> = HashMap::new();
metadata.insert("embedding_model".to_string(), json!(MODEL));
metadata.insert("pipeline_version".to_string(), json!(PIPELINE));
client
    .create_collection(
        CreateCollectionBuilder::new(COLLECTION)
            .vectors_config(VectorParamsBuilder::new(
                384, // all-MiniLM-L6-v2 output dimension
                Distance::Cosine,
            ))
            .metadata(metadata),
    )
    .await?;

static final String MODEL = "sentence-transformers/all-MiniLM-L6-v2";
static final String PIPELINE = "docs-prep-pipeline-v1";
static final String COLLECTION = "docs-sync-tutorial";

static void createCollection() throws Exception {
    client.createCollectionAsync(
        CreateCollection.newBuilder()
            .setCollectionName(COLLECTION)
            .setVectorsConfig(
                VectorsConfig.newBuilder()
                    .setParams(
                        VectorParams.newBuilder()
                            .setSize(384) // all-MiniLM-L6-v2 output dimension
                            .setDistance(Distance.Cosine)
                            .build())
                    .build())
            .putAllMetadata(Map.of(
                "embedding_model", value(MODEL),
                "pipeline_version", value(PIPELINE)))
            .build()).get();
}

var MODEL = "sentence-transformers/all-MiniLM-L6-v2";
var PIPELINE = "docs-prep-pipeline-v1";
var COLLECTION = "docs-sync-tutorial";

await client.CreateCollectionAsync(
    collectionName: COLLECTION,
    vectorsConfig: new VectorParams {
        Size = 384, // all-MiniLM-L6-v2 output dimension
        Distance = Distance.Cosine
    },
    metadata: new() {
        ["embedding_model"] = MODEL,
        ["pipeline_version"] = PIPELINE
    });

MODEL := "sentence-transformers/all-MiniLM-L6-v2"
PIPELINE := "docs-prep-pipeline-v1"
COLLECTION := "docs-sync-tutorial"

client.CreateCollection(context.Background(), &qdrant.CreateCollection{
    CollectionName: COLLECTION,
    VectorsConfig: qdrant.NewVectorsConfig(&qdrant.VectorParams{
        Size:     384, // all-MiniLM-L6-v2 output dimension
        Distance: qdrant.Distance_Cosine,
    }),
    Metadata: qdrant.NewValueMap(map[string]any{
        "embedding_model": MODEL,
        "pipeline_version": PIPELINE,
    }),
})

그 다음 게이트(gate)는 매 실행 시작 시점의 단순한 검사예요:

def check_gate():
    # compare this pipeline's constants against what the collection records about itself
    meta = client.get_collection(COLLECTION).config.metadata or {}
    if meta.get("embedding_model") != MODEL or meta.get("pipeline_version") != PIPELINE:
        raise RuntimeError(
            f"collection was built by {meta}: full re-embed into a fresh collection required"
        )

async function checkGate() {
  // compare this pipeline's constants against what the collection records about itself
  const meta = (await client.getCollection(COLLECTION)).config.metadata ?? {} as Record<string, unknown>;
  if (meta.embedding_model !== MODEL || meta.pipeline_version !== PIPELINE) {
    throw new Error(`collection was built by ${JSON.stringify(meta)}: full re-embed into a fresh collection required`);
  }
}

async fn check_gate(client: &Qdrant) -> anyhow::Result<()> {
    // compare this pipeline's constants against what the collection records about itself
    let meta = client.collection_info(COLLECTION).await?.result
        .and_then(|info| info.config)
        .map(|config| config.metadata)
        .unwrap_or_default();
    if meta.get("embedding_model").and_then(|v| v.as_str()).map(String::as_str) != Some(MODEL)
        || meta.get("pipeline_version").and_then(|v| v.as_str()).map(String::as_str) != Some(PIPELINE)
    {
        anyhow::bail!("collection was built by {meta:?}: full re-embed into a fresh collection required");
    }
    Ok(())
}

static void checkGate() throws Exception {
    // compare this pipeline's constants against what the collection records about itself
    Map<String, Value> meta = client.getCollectionInfoAsync(COLLECTION).get().getConfig().getMetadataMap();
    Value model = meta.get("embedding_model");
    Value pipeline = meta.get("pipeline_version");
    if (model == null || !MODEL.equals(model.getStringValue())
            || pipeline == null || !PIPELINE.equals(pipeline.getStringValue())) {
        throw new RuntimeException("collection was built by " + meta + ": full re-embed into a fresh collection required");
    }
}

async Task CheckGate() {
    // compare this pipeline's constants against what the collection records about itself
    var meta = (await client.GetCollectionInfoAsync(COLLECTION)).Config.Metadata;
    var model = meta.GetValueOrDefault("embedding_model")?.StringValue;
    var pipeline = meta.GetValueOrDefault("pipeline_version")?.StringValue;
    if (model != MODEL || pipeline != PIPELINE)
        throw new InvalidOperationException($"collection was built by {model}/{pipeline}: full re-embed into a fresh collection required");
}

checkGate := func() {
    // compare this pipeline's constants against what the collection records about itself
    info, err := client.GetCollectionInfo(context.Background(), COLLECTION)
    meta := info.GetConfig().GetMetadata()
    if meta["embedding_model"].GetStringValue() != MODEL || meta["pipeline_version"].GetStringValue() != PIPELINE {
        panic(fmt.Sprintf("collection was built by %v: full re-embed into a fresh collection required", meta))
    }
}

문서 청크의 특징

문서에 보통 어떤 일이 일어날까요? 완전히 새로운 것이 나타나거나, 페이지의 정보가 수정되거나, 페이지가 재구성되면서 섹션이 그대로 이동하거나, 페이지가 삭제되거나 하죠.

문서 청크의 서로 독립된 두 가지 특징을 모니터링하는 게 의미가 있어요:

콘텐츠(Content): 검색 대상이자 임베딩을 만드는 바로 그 텍스트예요.
위치(Position): 청크가 사는 곳 — 우리의 경우 URL과 anchor, 번호예요.

그래서 모든 레코드는 두 가지 파생 값을 가져야 해요:

콘텐츠 지문(Content fingerprint): 예를 들어 텍스트의 SHA-256이에요. 한 글자만 바뀌어도 바뀌고, 그 외에는 절대 바뀌지 않아요. 지문을 비교하면 텍스트를 직접 비교하지 않고도 "같은 콘텐츠인가?"에 답할 수 있어요.
결정적 ID(Deterministic ID): 문서 내 위치를 위한 거예요. 예를 들어 url + "#" + anchor + "::" + chunk_num을 UUID로 바꾼 것으로, Qdrant가 받아들이는 두 포인트 ID 형식 중 하나예요. ID를 비교하면 "이 콘텐츠가 여전히 같은 위치에 있나?"에 답할 수 있어요.

import hashlib
import uuid
from datetime import datetime, timezone

def content_hash(text):
    return hashlib.sha256(text.encode()).hexdigest()

def point_id(url, anchor, num):
    # NAMESPACE_URL is a fixed constant uuid5 requires;
    # it marks the input as a URL-like name
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{url}#{anchor}::{num}"))

def prepare_chunks_for_sync(chunks):
    """Derive both values (and the section address) for every raw chunk."""
    out = []
    for c in chunks:
        text = normalize(c["text"])
        out.append({
            **c,
            "text": text,
            "section_url": f"{c['url']}#{c['anchor']}" if c["anchor"] else c["url"],
            "content_hash": content_hash(text),
            "point_id": point_id(c["url"], c["anchor"], c["chunk_num"]),
        })
    return out

import { createHash } from "node:crypto";

type RawChunk = { url: string; anchor: string; chunk_num: number; text: string };
type SyncChunk = RawChunk & { section_url: string; content_hash: string; point_id: string };

function contentHash(text: string): string {
  return createHash("sha256").update(text).digest("hex");
}

// NAMESPACE_URL is a fixed constant name-based (v5) UUIDs require;
// it marks the input as a URL-like name
function pointId(url: string, anchor: string, num: number): string {
  // Qdrant accepts any well-formed UUID as a point ID:
  // hash the address, format the digest as a UUID, and the same address always yields the same ID
  const hex = createHash("sha256").update(`${url}#${anchor}::${num}`).digest("hex");
  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20, 32)}`;
}

// Derive both values (and the section address) for every raw chunk.
function prepareChunksForSync(chunks: RawChunk[]): SyncChunk[] {
  return chunks.map((c) => {
    const text = normalize(c.text);
    return {
      ...c,
      text,
      section_url: c.anchor ? `${c.url}#${c.anchor}` : c.url,
      content_hash: contentHash(text),
      point_id: pointId(c.url, c.anchor, c.chunk_num),
    };
  });
}

fn content_hash(text: &str) -> String {
    Sha256::digest(text.as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

fn point_id(url: &str, anchor: &str, num: u32) -> String {
    // NAMESPACE_URL is a fixed constant uuid5 requires;
    // it marks the input as a URL-like name
    uuid::Uuid::new_v5(
        &uuid::Uuid::NAMESPACE_URL,
        format!("{url}#{anchor}::{num}").as_bytes(),
    ).to_string()
}

/// Derive both values (and the section address) for every raw chunk.
fn prepare_chunks_for_sync(chunks: &[Chunk]) -> Vec<Chunk> {
    chunks.iter().map(|c| {
        let text = normalize(&c.text);
        Chunk {
            text: text.clone(),
            section_url: if c.anchor.is_empty() {
                c.url.clone()
            } else {
                format!("{}#{}", c.url, c.anchor)
            },
            content_hash: content_hash(&text),
            point_id: point_id(&c.url, &c.anchor, c.chunk_num),
            ..c.clone()
        }
    }).collect()
}

static String contentHash(String text) throws Exception {
    byte[] digest = MessageDigest.getInstance("SHA-256").digest(text.getBytes(StandardCharsets.UTF_8));
    return String.format("%064x", new BigInteger(1, digest));
}

static String pointId(String url, String anchor, int num) {
    // name-based UUID (version 3); the same address always yields the same ID
    return UUID.nameUUIDFromBytes((url + "#" + anchor + "::" + num).getBytes(StandardCharsets.UTF_8)).toString();
}

// Derive both values (and the section address) for every raw chunk.
static List<Chunk> prepareChunksForSync(List<Chunk> chunks) throws Exception {
    List<Chunk> out = new ArrayList<>();
    for (Chunk c : chunks) {
        String text = normalize(c.text);
        Chunk prepared = new Chunk(c.url, c.anchor, c.chunkNum, text);
        prepared.sectionUrl = !c.anchor.isEmpty() ? c.url + "#" + c.anchor : c.url;
        prepared.contentHash = contentHash(text);
        prepared.pointId = pointId(c.url, c.anchor, c.chunkNum);
        out.add(prepared);
    }
    return out;
}

string ContentHash(string text) =>
    Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(text))).ToLowerInvariant();

// Qdrant accepts any well-formed UUID as a point ID:
// a Guid built from the first 16 bytes of the address hash, so the same address always yields the same ID
string PointIdFor(string url, string anchor, int num) =>
    new Guid(SHA256.HashData(Encoding.UTF8.GetBytes($"{url}#{anchor}::{num}")).AsSpan(0, 16)).ToString();

// Derive both values (and the section address) for every raw chunk.
List<Chunk> PrepareChunksForSync(List<Chunk> chunks) {
    var prepared = new List<Chunk>();
    foreach (var c in chunks) {
        var text = Normalize(c.Text);
        prepared.Add(c with {
            Text = text,
            SectionUrl = c.Anchor != "" ? $"{c.Url}#{c.Anchor}" : c.Url,
            ContentHash = ContentHash(text),
            PointId = PointIdFor(c.Url, c.Anchor, c.ChunkNum),
        });
    }
    return prepared;
}

contentHash := func(text string) string {
    sum := sha256.Sum256([]byte(text))
    return hex.EncodeToString(sum[:])
}

pointID := func(url, anchor string, num int) string {
    // NewSHA1 with a namespace is UUIDv5; NameSpaceURL is a fixed constant it requires,
    // marking the input as a URL-like name
    return uuid.NewSHA1(uuid.NameSpaceURL, []byte(fmt.Sprintf("%s#%s::%d", url, anchor, num))).String()
}

// derive both values (and the section address) for every raw chunk
prepareChunksForSync := func(chunks []Chunk) []Chunk {
    out := make([]Chunk, 0, len(chunks))
    for _, c := range chunks {
        c.Text = normalize(c.Text)
        c.SectionURL = c.URL
        if c.Anchor != "" {
            c.SectionURL = c.URL + "#" + c.Anchor
        }
        c.ContentHash = contentHash(c.Text)
        c.PointID = pointID(c.URL, c.Anchor, c.ChunkNum)
        out = append(out, c)
    }
    return out
}

예시:

point ID: 2ff5204a-0353-5991-...  # UUID(url + "#" + anchor + "::" + chunk_num)
text (to vectorize): Prerequisites - Docker and Docker Compose...
content_hash: 27d55e75b962f1d5...  # sha256(text)

추가로, 포인트는 다음 필드로도 설명할 수 있어요.

페이로드 (Payload):

url: 한 페이지의 모든 청크를 필터링하거나 그룹화하기
section_url: 한 섹션의 모든 청크를 필터링하거나 그룹화하기
last_updated: 이 청크의 콘텐츠가 마지막으로 바뀐(또는 생성된) 시각

payload() 구현:

def payload(chunk, last_updated=None):
    return {
        "url": chunk["url"],
        "anchor": chunk["anchor"],
        "chunk_num": chunk["chunk_num"],
        "section_url": chunk["section_url"],
        "text": chunk["text"],
        "content_hash": chunk["content_hash"],
        "last_updated": last_updated or datetime.now(timezone.utc).isoformat(timespec="seconds"),
    }

function payload(chunk: SyncChunk, lastUpdated?: string) {
  return {
    url: chunk.url,
    anchor: chunk.anchor,
    chunk_num: chunk.chunk_num,
    section_url: chunk.section_url,
    text: chunk.text,
    content_hash: chunk.content_hash,
    last_updated: lastUpdated ?? new Date().toISOString().replace(/\.\d+Z$/, "Z"),
  };
}

fn payload(chunk: &Chunk, last_updated: Option<String>) -> anyhow::Result<Payload> {
    let last_updated = last_updated.unwrap_or_else(|| {
        chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, false)
    });
    Ok(Payload::try_from(serde_json::json!({
        "url": chunk.url,
        "anchor": chunk.anchor,
        "chunk_num": chunk.chunk_num,
        "section_url": chunk.section_url,
        "text": chunk.text,
        "content_hash": chunk.content_hash,
        "last_updated": last_updated,
    }))?)
}

static Map<String, Value> payload(Chunk chunk, String lastUpdated) {
    Map<String, Value> p = new HashMap<>();
    p.put("url", value(chunk.url));
    p.put("anchor", value(chunk.anchor));
    p.put("chunk_num", value(chunk.chunkNum));
    p.put("section_url", value(chunk.sectionUrl));
    p.put("text", value(chunk.text));
    p.put("content_hash", value(chunk.contentHash));
    p.put("last_updated", value(lastUpdated != null ? lastUpdated : OffsetDateTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.SECONDS).toString()));
    return p;
}

Dictionary<string, Value> Payload(Chunk chunk, string? lastUpdated = null) => new() {
    ["url"] = chunk.Url,
    ["anchor"] = chunk.Anchor,
    ["chunk_num"] = chunk.ChunkNum,
    ["section_url"] = chunk.SectionUrl,
    ["text"] = chunk.Text,
    ["content_hash"] = chunk.ContentHash,
    ["last_updated"] = lastUpdated ?? DateTimeOffset.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssK"),
};

payload := func(c Chunk, lastUpdated string) map[string]any {
    if lastUpdated == "" {
        lastUpdated = time.Now().UTC().Format(time.RFC3339)
    }
    return map[string]any{
        "url":          c.URL,
        "anchor":       c.Anchor,
        "chunk_num":    c.ChunkNum,
        "section_url":  c.SectionURL,
        "text":         c.Text,
        "content_hash": c.ContentHash,
        "last_updated": lastUpdated,
    }
}

필터링이나 그룹화에 쓰이는 모든 페이로드 필드에 대해 페이로드 인덱스를 만들어야 해요.

for field in ("content_hash", "url", "section_url"):
    client.create_payload_index(COLLECTION, field, models.PayloadSchemaType.KEYWORD)

for (const field of ["content_hash", "url", "section_url"]) {
  await client.createPayloadIndex(COLLECTION, {
    field_name: field,
    field_schema: "keyword",
  });
}

for field in ["content_hash", "url", "section_url"] {
    client.create_field_index(
        CreateFieldIndexCollectionBuilder::new(COLLECTION, field, FieldType::Keyword),
    ).await?;
}

static void createPayloadIndexes() throws Exception {
    for (String field : List.of("content_hash", "url", "section_url")) {
        client.createPayloadIndexAsync(COLLECTION, field, PayloadSchemaType.Keyword, null, null, null, null).get();
    }
}

foreach (var field in new[] { "content_hash", "url", "section_url" })
    await client.CreatePayloadIndexAsync(COLLECTION, field, PayloadSchemaType.Keyword);

for _, field := range []string{"content_hash", "url", "section_url"} {
    client.CreateFieldIndex(context.Background(), &qdrant.CreateFieldIndexCollection{
        CollectionName: COLLECTION,
        FieldName: field,
        FieldType: qdrant.FieldType_FieldTypeKeyword.Enum(),
    })
}

컬렉션 채우기

전체 문서로 컬렉션을 채워요.

client.upsert(
    COLLECTION,
    points=[
        models.PointStruct(
            id=c["point_id"],
            vector=models.Document(text=c["text"], model=MODEL),
            payload=payload(c),
        )
        for c in prepare_chunks_for_sync(CHUNKS)
    ],
    wait=True,
)

await client.upsert(COLLECTION, {
  points: prepareChunksForSync(CHUNKS).map((c) => ({
    id: c.point_id,
    vector: { text: c.text, model: MODEL },
    payload: payload(c),
  })),
  wait: true,
});

let points: Vec<PointStruct> = prepare_chunks_for_sync(&chunks)
    .iter()
    .map(|c| {
        Ok(PointStruct::new(
            c.point_id.clone(),
            Document::new(&c.text, MODEL),
            payload(c, None)?,
        ))
    })
    .collect::<anyhow::Result<_>>()?;
client.upsert_points(UpsertPointsBuilder::new(COLLECTION, points).wait(true)).await?;

static void populate() throws Exception {
    List<PointStruct> points = new ArrayList<>();
    for (Chunk c : prepareChunksForSync(CHUNKS)) {
        points.add(PointStruct.newBuilder()
            .setId(id(UUID.fromString(c.pointId)))
            .setVectors(vectors(vector(Document.newBuilder().setText(c.text).setModel(MODEL).build())))
            .putAllPayload(payload(c, null))
            .build());
    }
    client.upsertAsync(COLLECTION, points).get();
}

await client.UpsertAsync(
    collectionName: COLLECTION,
    points: PrepareChunksForSync(CHUNKS).Select(c => new PointStruct {
        Id = new PointId { Uuid = c.PointId },
        Vectors = new Document { Text = c.Text, Model = MODEL },
        Payload = { Payload(c) },
    }).ToList(),
    wait: true);

var points []*qdrant.PointStruct
for _, c := range prepareChunksForSync(CHUNKS) {
    points = append(points, &qdrant.PointStruct{
        Id: qdrant.NewID(c.PointID),
        Vectors: qdrant.NewVectorsDocument(&qdrant.Document{Text: c.Text, Model: MODEL}),
        Payload: qdrant.NewValueMap(payload(c, "")),
    })
}
client.Upsert(context.Background(), &qdrant.UpsertPoints{
    CollectionName: COLLECTION,
    Points: points,
    Wait: qdrant.PtrOf(true),
})

그에 대한 검색을 테스트해 봐요:

QUERY = "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?"
client.query_points(
    COLLECTION,
    query=models.Document(text=QUERY, model=MODEL),
    limit=3,
    with_payload=["section_url", "text"],
)

const QUERY = "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?";
await client.query(COLLECTION, {
  query: { text: QUERY, model: MODEL },
  limit: 3,
  with_payload: ["section_url", "text"],
});

const QUERY: &str = "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?";
client.query(
    QueryPointsBuilder::new(COLLECTION)
        .query(Query::new_nearest(Document::new(QUERY, MODEL)))
        .limit(3)
        .with_payload(PayloadIncludeSelector::new(vec![
            "section_url".to_string(),
            "text".to_string(),
        ])),
).await?;

static final String QUERY = "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?";
static void search() throws Exception {
    client.queryAsync(
        QueryPoints.newBuilder()
            .setCollectionName(COLLECTION)
            .setQuery(nearest(Document.newBuilder().setText(QUERY).setModel(MODEL).build()))
            .setLimit(3)
            .setWithPayload(WithPayloadSelectorFactory.include(List.of("section_url", "text")))
            .build()).get();
}

var QUERY = "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?";
await client.QueryAsync(
    collectionName: COLLECTION,
    query: new Document { Text = QUERY, Model = MODEL },
    limit: 3,
    payloadSelector: new[] { "section_url", "text" });

QUERY := "Where exactly to set `QDRANT__SERVICE__API_KEY` variable to enable authentication for a self-hosted Qdrant?"
client.Query(context.Background(), &qdrant.QueryPoints{
    CollectionName: COLLECTION,
    Query: qdrant.NewQueryDocument(&qdrant.Document{Text: QUERY, Model: MODEL}),
    Limit: qdrant.PtrOf(uint64(3)),
    WithPayload: qdrant.NewWithPayloadInclude("section_url", "text"),
})

이런 결과가 나올 거예요:

0.675 https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/#secure-a-self-hosted-qdrant-instance Secure a Self-Hosted Qdrant Instance | Time: 45 min | Level: Intermediate | ...

문서 변경과 동기화하기

동기화 트리거는, 문서가 git에 있다면 머지(merge) 시점의 CI 작업일 수도 있고, 야간 크론 작업일 수도 있어요.

여기서 문서 컬렉션과의 동기화 입력은 문서의 현재 전체 청크 목록이에요. 단순하고 결정적인 데이터 준비 파이프라인이라면 하루에 한 번 전체 목록을 모으는 건 저렴해서, 원본 변경을 파생하는 번거로움을 덜 수 있어요.

들어오는 각 청크는 point ID(문서에서 청크의 주소)와 content_hash(청크의 정확한 콘텐츠, 지문)를 기준으로 현재 문서 컬렉션과 다음 방식 중 하나로 비교돼요:

incoming chunk
├─ ID found in the collection?
│   ├─ yes: fingerprint equal?
│   │   ├─ yes -> unchanged: the point stays as is
│   │   └─ no -> content changed: re-embed in place
│   └─ no: identical fingerprint under another ID?
│       ├─ yes -> address changed: reuse the vector, create a new point
│       └─ no -> new: embed and insert a new point
└─ stored point whose ID is absent from the incoming list -> gone: delete (last, after all writes)

참고: 선택적 안전망이 있어요. 동기화 전에 스냅샷을 찍어 두고, 모든 게 정상이면 나중에 삭제하는 거예요.

동기화 파이프라인의 입력

몇 가지 가능한 변경을 생각해 볼게요:

"Secure a Self-Hosted Qdrant Instance" 튜토리얼에 "Step 6: Rotate API keys"라는 새 작은 섹션을 추가하고, "Step 3" 섹션이 이것을 가리키도록 바꾼 경우
마이그레이션 페이지가 새 URL로 이동한 경우
"Time-based sharding" 튜토리얼이 삭제된 경우

이 세 가지 변경이 반영된 LATEST_CHUNKS 목록:

untouched_secure_qdrant = [
    c for c in CHUNKS
    if c["url"] == "https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/"
    and c["anchor"] != "step-3-enable-an-admin-api-key"
]

# now points to the new section
step_3 = {
    "url": "https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/",
    "anchor": "step-3-enable-an-admin-api-key",
    "chunk_num": 0,
    "text": "Step 3: Enable an Admin API Key ... Refer to Security > Authentication to learn more about admin API keys, including API key rotation. --- See also: rotating API keys.",
}

# the new section
step_6 = {
    "url": "https://qdrant.tech/documentation/tutorials-operations/secure-qdrant/",
    "anchor": "step-6-rotate-api-keys",
    "chunk_num": 0,
    "text": "Step 6: Rotate API keys Rotate the admin API key on a schedule and immediately after any suspected exposure. Update every client before revoking the old key.",
}

# the migration page moved: same texts, new addresses
moved = [
    {**c, "url": "https://qdrant.tech/documentation/tutorials-operations/migration-guide/"}
    for c in CHUNKS if c["url"] == "https://qdrant.tech/documentation/tutorials-operations/migration/"
]

# the time-based-sharding tutorial is absent from LATEST_CHUNKS - that is how a deletion arrives
LATEST_CHUNKS = prepare_chunks_for_sync(untouched_secure_qdrant + [step_3, step_6] + moved)

이제 들어오는 각 청크를 컬렉션과 대조해 봐요. ID(주소)가 존재하는지, content_hash(정확한 텍스트)가 일치하는지요.

retrieve는 ID로 포인트를 가져와요. 코퍼스 규모에서는 ID를 배치로 묶어서 처리하면 돼요.

def split_by_state(latest_chunks):
    """Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown."""
    incoming = {c["point_id"]: c for c in latest_chunks}
    stored = {}
    points = client.retrieve(
        COLLECTION,
        ids=list(incoming),
        with_payload=["content_hash"],
        with_vectors=False,
    )
    for p in points:
        stored[str(p.id)] = p.payload["content_hash"]

    unchanged, content_changed, unknown_ids = [], [], []
    for pid, c in incoming.items():
        if stored.get(pid) == c["content_hash"]:
            unchanged.append(c)
        elif pid in stored:
            content_changed.append(c)
        else:
            unknown_ids.append(c)
    return incoming, unchanged, content_changed, unknown_ids

incoming_ids, unchanged, content_changed, unknown_ids = split_by_state(LATEST_CHUNKS)

// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
async function splitByState(latestChunks: SyncChunk[]) {
  const incoming = new Map(latestChunks.map((c) => [c.point_id, c]));
  const stored = new Map<string, string>();
  const points = await client.retrieve(COLLECTION, {
    ids: [...incoming.keys()],
    with_payload: ["content_hash"],
    with_vector: false,
  });
  for (const p of points) {
    stored.set(String(p.id), p.payload?.content_hash as string);
  }
  const unchanged: SyncChunk[] = [];
  const contentChanged: SyncChunk[] = [];
  const unknownIds: SyncChunk[] = [];
  for (const [pid, c] of incoming) {
    if (stored.get(pid) === c.content_hash) {
      unchanged.push(c);
    } else if (stored.has(pid)) {
      contentChanged.push(c);
    } else {
      unknownIds.push(c);
    }
  }
  return { incoming, unchanged, contentChanged, unknownIds };
}
const { incoming, unchanged, contentChanged, unknownIds } = await splitByState(LATEST_CHUNKS);

/// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
async fn split_by_state(
    client: &Qdrant,
    latest_chunks: &[Chunk],
) -> anyhow::Result<(HashMap<String, Chunk>, Vec<Chunk>, Vec<Chunk>, Vec<Chunk>)> {
    let incoming: HashMap<String, Chunk> = latest_chunks
        .iter()
        .map(|c| (c.point_id.clone(), c.clone()))
        .collect();
    let ids: Vec<PointId> = incoming.keys().map(|id| id.as_str().into()).collect();
    let points = client.get_points(
        GetPointsBuilder::new(COLLECTION, ids)
            .with_payload(PayloadIncludeSelector::new(vec!["content_hash".to_string()]))
            .with_vectors(false),
    ).await?;
    let mut stored: HashMap<String, String> = HashMap::new();
    for p in points.result {
        let hash = p.get("content_hash").as_str().cloned();
        if let (Some(PointIdOptions::Uuid(id)), Some(hash)) = (p.id.and_then(|i| i.point_id_options), hash) {
            stored.insert(id, hash);
        }
    }
    let (mut unchanged, mut content_changed, mut unknown_ids) = (Vec::new(), Vec::new(), Vec::new());
    for (pid, c) in &incoming {
        if stored.get(pid) == Some(&c.content_hash) {
            unchanged.push(c.clone());
        } else if stored.contains_key(pid) {
            content_changed.push(c.clone());
        } else {
            unknown_ids.push(c.clone());
        }
    }
    Ok((incoming, unchanged, content_changed, unknown_ids))
}
let (incoming_ids, unchanged, content_changed, unknown_ids) = split_by_state(&client, &latest_chunks).await?;

static class SyncState {
    Map<String, Chunk> incoming = new LinkedHashMap<>();
    List<Chunk> unchanged = new ArrayList<>();
    List<Chunk> contentChanged = new ArrayList<>();
    List<Chunk> unknownIds = new ArrayList<>();
}

// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
static SyncState splitByState(List<Chunk> latestChunks) throws Exception {
    SyncState state = new SyncState();
    for (Chunk c : latestChunks) {
        state.incoming.put(c.pointId, c);
    }
    Map<String, String> stored = new HashMap<>();
    var points = client.retrieveAsync(
        COLLECTION,
        state.incoming.keySet().stream().map(pid -> id(UUID.fromString(pid))).collect(Collectors.toList()),
        WithPayloadSelectorFactory.include(List.of("content_hash")),
        WithVectorsSelectorFactory.enable(false),
        null).get();
    for (var p : points) {
        stored.put(p.getId().getUuid(), p.getPayloadMap().get("content_hash").getStringValue());
    }
    for (Map.Entry<String, Chunk> e : state.incoming.entrySet()) {
        String pid = e.getKey();
        Chunk c = e.getValue();
        if (c.contentHash.equals(stored.get(pid))) {
            state.unchanged.add(c);
        } else if (stored.containsKey(pid)) {
            state.contentChanged.add(c);
        } else {
            state.unknownIds.add(c);
        }
    }
    return state;
}

// Compare the incoming chunk list to the collection: who is unchanged, changed, or unknown.
async Task<(Dictionary<string, Chunk> incomingIds, List<Chunk> unchanged, List<Chunk> contentChanged, List<Chunk> unknownIds)> SplitByState(List<Chunk> latestChunks) {
    var incoming = latestChunks.ToDictionary(c => c.PointId);
    var stored = new Dictionary<string, string>();
    var points = await client.RetrieveAsync(
        COLLECTION,
        ids: incoming.Keys.Select(pid => new PointId { Uuid = pid }).ToList(),
        payloadSelector: new[] { "content_hash" },
        vectorSelector: false);
    foreach (var p in points) stored[p.Id.Uuid] = p.Payload["content_hash"].StringValue;
    var unchanged = new List<Chunk>();
    var contentChanged = new List<Chunk>();
    var unknownIds = new List<Chunk>();
    foreach (var (pid, c) in incoming) {
        if (stored.TryGetValue(pid, out var hash) && hash == c.ContentHash)
            unchanged.Add(c);
        else if (stored.ContainsKey(pid))
            contentChanged.Add(c);
        else
            unknownIds.Add(c);
    }
    return (incoming, unchanged, contentChanged, unknownIds);
}
var splitState = await SplitByState(LATEST_CHUNKS);

// compare the incoming chunk list to the collection: who is unchanged, changed, or unknown
splitByState := func(latestChunks []Chunk) (map[string]Chunk, []Chunk, []Chunk, []Chunk) {
    incoming := make(map[string]Chunk, len(latestChunks))
    ids := make([]*qdrant.PointId, 0, len(latestChunks))
    for _, c := range latestChunks {
        incoming[c.PointID] = c
        ids = append(ids, qdrant.NewID(c.PointID))
    }
    retrieved, err := client.Get(context.Background(), &qdrant.GetPoints{
        CollectionName: COLLECTION,
        Ids: ids,
        WithPayload: qdrant.NewWithPayloadInclude("content_hash"),
        WithVectors: qdrant.NewWithVectors(false),
    })
    stored := make(map[string]string, len(retrieved))
    for _, p := range retrieved {
        stored[p.GetId().GetUuid()] = p.GetPayload()["content_hash"].GetStringValue()
    }
    var unchanged, contentChanged, unknownIDs []Chunk
    for pid, c := range incoming {
        storedHash, found := stored[pid]
        switch {
        case found && storedHash == c.ContentHash:
            unchanged = append(unchanged, c)
        case found:
            contentChanged = append(contentChanged, c)
        default:
            unknownIDs = append(unknownIDs, c)
        }
    }
    return incoming, unchanged, contentChanged, unknownIDs
}
incomingIDs, unchanged, contentChanged, unknownIDs := splitByState(LATEST_CHUNKS)

사례 1: 바뀌지 않음 — 아무것도 하지 않기

이 청크들은 이전과 같은 지문을 갖고 있어요.

Step 3에 대한 청크는 알려진 ID 아래 존재해요(문서 웹사이트에서 위치가 바뀌지 않았죠). 하지만 새 정보를 담고 있어요. upsert를 쓰면 돼요. 기존 ID 아래에 포인트를 쓰면 그 포인트를 교체하거든요.

def re_embed_changed(content_changed):
    if not content_changed:
        return
    client.upsert(
        COLLECTION,
        points=[
            models.PointStruct(
                id=c["point_id"],
                vector=models.Document(text=c["text"], model=MODEL),
                payload=payload(c),
            )
            for c in content_changed
        ],
        wait=True,
    )

async function reEmbedChanged(contentChanged: SyncChunk[]) {
  if (contentChanged.length === 0) {
    return;
  }
  await client.upsert(COLLECTION, {
    points: contentChanged.map((c) => ({
      id: c.point_id,
      vector: { text: c.text, model: MODEL },
      payload: payload(c),
    })),
    wait: true,
  });
}

async fn re_embed_changed(client: &Qdrant, content_changed: &[Chunk]) -> anyhow::Result<()> {
    if content_changed.is_empty() {
        return Ok(());
    }
    let points: Vec<PointStruct> = content_changed
        .iter()
        .map(|c| {
            Ok(PointStruct::new(
                c.point_id.clone(),
                Document::new(&c.text, MODEL),
                payload(c, None)?,
            ))
        })
        .collect::<anyhow::Result<_>>()?;
    client.upsert_points(UpsertPointsBuilder::new(COLLECTION, points).wait(true)).await?;
    Ok(())
}

static void reEmbedChanged(List<Chunk> contentChanged) throws Exception {
    if (contentChanged.isEmpty()) {
        return;
    }
    List<PointStruct> points = new ArrayList<>();
    for (Chunk c : contentChanged) {
        points.add(PointStruct.newBuilder()
            .setId(id(UUID.fromString(c.pointId)))
            .setVectors(vectors(vector(Document.newBuilder().setText(c.text).setModel(MODEL).build())))
            .putAllPayload(payload(c, null))
            .build());
    }
    client.upsertAsync(COLLECTION, points).get();
}

async Task ReEmbedChanged(List<Chunk> contentChanged) {
    if (contentChanged.Count == 0) return;
    await client.UpsertAsync(
        collectionName: COLLECTION,
        points: contentChanged.Select(c => new PointStruct {
            Id = new PointId { Uuid = c.PointId },
            Vectors = new Document { Text = c.Text, Model = MODEL },
            Payload = { Payload(c) },
        }).ToList(),
        wait: true);
}

reEmbedChanged := func(contentChanged []Chunk) {
    if len(contentChanged) == 0 { return }
    points := make([]*qdrant.PointStruct, 0, len(contentChanged))
    for _, c := range contentChanged {
        points = append(points, &qdrant.PointStruct{
            Id: qdrant.NewID(c.PointID),
            Vectors: qdrant.NewVectorsDocument(&qdrant.Document{Text: c.Text, Model: MODEL}),
            Payload: qdrant.NewValueMap(payload(c, "")),
        })
    }
    client.Upsert(context.Background(), &qdrant.UpsertPoints{
        CollectionName: COLLECTION,
        Points: points,
        Wait: qdrant.PtrOf(true),
    })
}

사례 3과 4: ID가 컬렉션에 없음

여섯 개의 ID가 컬렉션에 없어요. 그런데 알 수 없는 ID라고 해서 반드시 새 콘텐츠인 건 아니에요. 페이지가 그대로 이동하면, 페이지의 모든 청크가 새 주소(새 ID)를 얻지만 텍스트는 정확히 같아요. 다시 임베딩해도 같은 벡터가 나오는데, 왜 그 비용을 치르겠어요.

content_hash에 대한 필터링된 scroll이 "이 정확한 텍스트가 다른 ID 아래 이미 존재하나?"라는 질문에 답해 줘요.

히트가 나면, 저장된 벡터를 새 포인트에 복사하고 콘텐츠가 안 바뀌었으므로 원본의 last_updated를 유지해요.
미스가 나면, 콘텐츠가 진짜 새로운 것이므로 임베딩해서 새 포인트를 삽입해요.

참고: 이 버전은 알 수 없는 청크마다 해시 조회를 한 번씩 수행해서 의사결정을 살펴보기 쉽게 만들었어요. 운영 환경에서는 해시 조회와 포인트 upsert를 배치로 묶어 처리하세요.

def reuse_or_add(unknown_ids):
    """Reuse an existing embedding when the same text is already stored; embed only what is new."""
    reused, added = 0, 0
    for c in unknown_ids:
        same_text = models.Filter(
            must=[
                models.FieldCondition(
                    key="content_hash",
                    match=models.MatchValue(value=c["content_hash"]),
                )
            ]
        )
        hits, _ = client.scroll(
            COLLECTION,
            scroll_filter=same_text,
            limit=1,
            with_payload=["last_updated"],
            with_vectors=True,
        )
        if hits:
            # same text, new address: copy the vector, keep its last_updated
            point = models.PointStruct(
                id=c["point_id"],
                vector=hits[0].vector,
                payload=payload(c, hits[0].payload["last_updated"]),
            )
            reused += 1
        else:
            # genuinely new content: embed and insert
            point = models.PointStruct(
                id=c["point_id"],
                vector=models.Document(text=c["text"], model=MODEL),
                payload=payload(c),
            )
            added += 1
        client.upsert(COLLECTION, points=[point], wait=True)
    return reused, added

// Reuse an existing embedding when the same text is already stored; embed only what is new.
async function reuseOrAdd(unknownIds: SyncChunk[]) {
  let reused = 0;
  let added = 0;
  for (const c of unknownIds) {
    const sameText = {
      must: [
        { key: "content_hash", match: { value: c.content_hash } },
      ],
    };
    const hits = (await client.scroll(COLLECTION, {
      filter: sameText,
      limit: 1,
      with_payload: ["last_updated"],
      with_vector: true,
    })).points;
    let point: Schemas["PointStruct"];
    if (hits.length > 0) {
      // same text, new address: copy the vector, keep its last_updated
      point = {
        id: c.point_id,
        vector: hits[0].vector as number[],
        payload: payload(c, hits[0].payload?.last_updated as string),
      };
      reused += 1;
    } else {
      // genuinely new content: embed and insert
      point = {
        id: c.point_id,
        vector: { text: c.text, model: MODEL },
        payload: payload(c),
      };
      added += 1;
    }
    await client.upsert(COLLECTION, { points: [point], wait: true });
  }
  return { reused, added };
}

/// Reuse an existing embedding when the same text is already stored; embed only what is new.
async fn reuse_or_add(client: &Qdrant, unknown_ids: &[Chunk]) -> anyhow::Result<(usize, usize)> {
    let (mut reused, mut added) = (0, 0);
    for c in unknown_ids {
        let same_text = Filter::must([Condition::matches("content_hash", c.content_hash.clone())]);
        let hits = client
            .scroll(
                ScrollPointsBuilder::new(COLLECTION)
                    .filter(same_text)
                    .limit(1)
                    .with_payload(PayloadIncludeSelector::new(vec!["last_updated".to_string()]))
                    .with_vectors(true),
            )
            .await?
            .result;
        let point = if let Some(hit) = hits.into_iter().next() {
            // same text, new address: copy the vector, keep its last_updated
            let last_updated = hit.get("last_updated").as_str().cloned();
            let vector: Vec<f32> = match hit.vectors.and_then(|v| v.vectors_options) {
                Some(vectors_output::VectorsOptions::Vector(v)) => match v.vector {
                    Some(vector_output::Vector::Dense(dense)) => dense.data,
                    _ => anyhow::bail!("expected a dense vector on the stored point"),
                },
                _ => anyhow::bail!("expected a dense vector on the stored point"),
            };
            reused += 1;
            PointStruct::new(c.point_id.clone(), vector, payload(c, last_updated)?)
        } else {
            // genuinely new content: embed and insert
            added += 1;
            PointStruct::new(c.point_id.clone(), Document::new(&c.text, MODEL), payload(c, None)?)
        };
        client.upsert_points(UpsertPointsBuilder::new(COLLECTION, vec![point]).wait(true)).await?;
    }
    Ok((reused, added))
}

// Reuse an existing embedding when the same text is already stored; embed only what is new.
static int[] reuseOrAdd(List<Chunk> unknownIds) throws Exception {
    int reused = 0;
    int added = 0;
    for (Chunk c : unknownIds) {
        Filter sameText = Filter.newBuilder()
            .addMust(matchKeyword("content_hash", c.contentHash))
            .build();
        var hits = client.scrollAsync(
            ScrollPoints.newBuilder()
                .setCollectionName(COLLECTION)
                .setFilter(sameText)
                .setLimit(1)
                .setWithPayload(WithPayloadSelectorFactory.include(List.of("last_updated")))
                .setWithVectors(WithVectorsSelectorFactory.enable(true))
                .build()).get().getResult();
        PointStruct point;
        if (!hits.isEmpty()) {
            // same text, new address: copy the vector, keep its last_updated
            point = PointStruct.newBuilder()
                .setId(id(UUID.fromString(c.pointId)))
                .setVectors(vectors(hits.get(0).getVectors()))
                .putAllPayload(payload(c, hits.get(0).getPayloadMap().get("last_updated").getStringValue()))
                .build();
            reused += 1;
        } else {
            // genuinely new content: embed and insert
            point = PointStruct.newBuilder()
                .setId(id(UUID.fromString(c.pointId)))
                .setVectors(vectors(vector(Document.newBuilder().setText(c.text).setModel(MODEL).build())))
                .putAllPayload(payload(c, null))
                .build();
            added += 1;
        }
        client.upsertAsync(COLLECTION, List.of(point), true, null).get();
    }
    return new int[]{reused, added};
}

// Reuse an existing embedding when the same text is already stored; embed only what is new.
async Task<(int reused, int added)> ReuseOrAdd(List<Chunk> unknownIds) {
    int reused = 0, added = 0;
    foreach (var c in unknownIds) {
        var sameText = new Filter {
            Must = { new FieldCondition {
                Key = "content_hash",
                Match = new Match { Keyword = c.ContentHash },
            } },
        };
        var hits = (await client.ScrollAsync(
            collectionName: COLLECTION,
            scrollFilter: sameText,
            limit: 1,
            payloadSelector: new[] { "last_updated" },
            withVectors: true)).Result;
        PointStruct point;
        if (hits.Count > 0) {
            // same text, new address: copy the vector, keep its last_updated
            point = new PointStruct {
                Id = new PointId { Uuid = c.PointId },
                Vectors = hits[0].Vectors,
                Payload = { Payload(c, hits[0].Payload["last_updated"].StringValue) },
            };
            reused += 1;
        } else {
            // genuinely new content: embed and insert
            point = new PointStruct {
                Id = new PointId { Uuid = c.PointId },
                Vectors = new Document { Text = c.Text, Model = MODEL },
                Payload = { Payload(c) },
            };
            added += 1;
        }
        await client.UpsertAsync(collectionName: COLLECTION, points: new[] { point }, wait: true);
    }
    return (reused, added);
}

// Reuse an existing embedding when the same text is already stored; embed only what is new.
reuseOrAdd := func(unknownIDs []Chunk) (int, int) {
    reused, added := 0, 0
    for _, c := range unknownIDs {
        sameText := &qdrant.Filter{
            Must: []*qdrant.Condition{
                qdrant.NewConditionMatch(&qdrant.Match{
                    MatchValue: qdrant.NewMatchText(c.ContentHash),
                }, &qdrant.FieldCondition{
                    Field: qdrant.PtrOf("content_hash"),
                }),
            },
        }
        // ... (batching recommended in production)
        hits := client.Scroll(context.Background(), &qdrant.ScrollPoints{
            CollectionName: COLLECTION,
            Filter: sameText,
            Limit: qdrant.PtrOf(uint64(1)),
            WithPayload: qdrant.NewWithPayloadInclude("last_updated"),
            WithVectors: qdrant.NewWithVectors(true),
        }).GetResult()
        if len(hits) > 0 {
            // same text, new address: copy the vector, keep its last_updated
            reused += 1
            // upsert point with copied vector ...
        } else {
            // genuinely new content: embed and insert
            added += 1
        }
    }
    return reused, added
}

삭제 처리하기

들어오는 목록에 없는 저장 포인트는 사라진(삭제된) 문서를 나타내요. 모든 쓰기가 끝난 다음, 마지막에 이를 처리해요.

def delete_gone(stored_ids, incoming_ids):
    gone = stored_ids - set(incoming_ids)
    if gone:
        client.delete(COLLECTION, points_selector=models.PointIdsList(points=list(gone)))

여기서 stored_ids는 컬렉션에서 조회한 ID 목록이고, incoming_ids는 들어오는 청크의 ID 목록이에요.

정리

원본 데이터의 변화를 추적해서 증분적으로 벡터를 갱신하는 파이프라인은 복잡해 보이지만, 두 가지 파생 값(콘텐츠 지문과 결정적 ID)만 잘 정의하면 각 청크의 상태를 명확하게 나눌 수 있어요. 바뀌지 않은 청크는 그대로 두고, 위치만 바뀐 청크는 벡터를 재사용하고, 진짜 새로운 콘텐츠만 임베딩하며, 사라진 청크는 마지막에 삭제하면 돼요. 이 패턴 덕분에 문서가 자주 바뀌어도 검색은 항상 최신 상태를 가리키게 할 수 있어요.

증분 임베딩 업데이트