새 임베딩 모델로 마이그레이션
새 임베딩 모델로 마이그레이션 (tutorials-operations-embedding-model-migration)
의미적 검색 애플리케이션을 구축할 때는 임베딩 모델을 선택해야 해요. 시간이 지나면서 더 나은 품질이나 비용 효율을 위해 다른 모델로 전환하고 싶어질 수 있어요. 애플리케이션이 프로덕션에 있다면, 사용자를 방해하지 않으려면 이 작업을 제로 다운타임으로 수행해야 해요. 모델 전환은 컬렉션의 모든 벡터를 다시 임베딩해야 하므로 시간이 걸릴 수 있어요.
이 튜토리얼에서는 제로 다운타임으로 새 모델로 마이그레이션하는 두 가지 옵션을 단계별로 안내할게요.
| 시간: 40분 | 난이도: 중급 |
|---|
다시 임베딩하려면 임베딩을 만드는 데 사용한 원본 데이터에 대한 접근이 필요해요. 이 데이터는 기본 데이터베이스에서 오거나, Qdrant 포인트의 payload에 저장되어 있을 수 있어요. 이 튜토리얼에서는 필요한 데이터가 payload에 저장되어 있다고 가정해요. 보통 payload가 임베딩을 생성하는 데 사용된 텍스트나 다른 데이터를 담고 있기 때문에, 이는 일반적으로 사실이에요.
이 튜토리얼의 코드 예시는 Qdrant Cloud Inference를 사용해 벡터 임베딩을 생성해요. 자체 임베딩 인프라를 관리한다면 같은 원리를 적용할 수 있지만, 임베딩 서비스에 맞게 코드 예시를 조정해야 해요.
두 가지 옵션 (Two Options)
새 임베딩 모델로 마이그레이션하는 가장 좋은 방법은 컬렉션이 어떻게 구성되어 있는지에 따라 달라져요. 블루-그린 마이그레이션(옵션 1)은 모든 컬렉션 유형에서 작동해요. 반면 명명된 벡터(named vectors)를 사용하고 배포 버전이 1.18 이상이라면, 옵션 2가 더 쉽고 빠르며 리소스를 덜 사용해요.
옵션 1: 블루-그린 마이그레이션
블루-그린 마이그레이션은 두 개의 병렬 컬렉션을 사용해요. 새 임베딩 모델에 맞게 구성된 새 컬렉션을 먼저 만들어요. 그런 다음 들어오는 모든 upsert가 두 컬렉션 모두에 동시에 쓰이도록 이중 쓰기(dual writes)를 활성화해요. 백그라운드 스크롤로 각 포인트를 새 모델로 다시 임베딩해 새 컬렉션에 써요. 마이그레이션이 완료되면 검색 트래픽을 새 컬렉션으로 전환(가능하면 alias를 플립)하고 이중 쓰기를 비활성화해요. 이 옵션은 명명되지 않은 벡터든 명명된 벡터든 모든 컬렉션 유형에서 작동해요.
이 방법에는 몇 가지 단점이 있어요.
- 두 컬렉션 모두에 payload가 중복돼요. payload가 큰 텍스트 중심 컬렉션에서는 이 영향이 상당할 수 있어요.
- 마이그레이션 중 삭제나 부분 업데이트를 일시 중지해야 하거나, 이를 처리하는 추가 로직을 구현해야 해요.
옵션 2: 명명된 벡터
명명된 벡터 접근 방식은 모든 것을 단일 컬렉션에 유지해요. 먼저 새 모델을 추가 명명된 벡터로 추가해요. 이는 기존 데이터에 영향을 주지 않는 스키마 전용 작업이에요. 다음으로 이중 쓰기를 활성화해 들어오는 모든 upsert가 두 모델 모두로 임베딩되도록 해요. 그런 다음 백그라운드 스크롤로 기존 각 포인트의 새 명명된 벡터를 업데이트하고, 기존 벡터와 payload는 그대로 둬요. 모든 포인트가 다시 임베딩되면 검색 쿼리의 using 매개변수를 새 벡터로 전환한 뒤, 기존 명명된 벡터를 삭제해요.
이 방법의 단점은 명명된 벡터로 생성된 컬렉션에서만 작동한다는 거예요.
블루-그린 마이그레이션과 비교해 이 방법은:
- 두 번째 컬렉션이나 데이터 복사가 필요 없어요.
- 마이그레이션 전체에 걸쳐 모든 포인트 ID, payload, 기타 명명된 벡터를 그대로 유지해요.
- 롤백이 간단해요. 명시적으로 삭제하기 전까지 기존 명명된 벡터가 컬렉션에 남아 있어요.
옵션 1과 달리, 이 마이그레이션 동안 포인트 삭제는 안전해요. 포인트를 삭제하면 컬렉션에서 완전히 제거되므로, 마이그레이션 프로세스가 다시 추가할 위험이 없어요. 벡터를 업데이트할 때는 이중 쓰기 로직이 동시에 새 명명된 벡터도 업데이트하도록 해야 해요. 하나만 업데이트하면 두 벡터가 분기될 거예요.
블루-그린 마이그레이션 (Blue-Green Migration)
블루-그린 마이그레이션은 두 개의 컬렉션을 사용해요. 첫 번째 컬렉션에는 기존 임베딩이 있고, 두 번째 컬렉션은 새 임베딩을 저장하는 데 사용돼요. 마이그레이션 프로세스는 기존 컬렉션에서 새 컬렉션으로 데이터를 복사하면서 새 모델로 벡터를 다시 임베딩해요. 마이그레이션 중에는 기존 컬렉션을 계속 검색하면서 데이터 업데이트는 두 컬렉션 모두에 써요. 모든 벡터가 다시 임베딩되면 검색이 새 컬렉션을 사용하도록 전환해요.
여기서 설명한 해법은 upsert 작업에만 그대로 작동해요. 삭제나 부분 업데이트를 사용한다면, 마이그레이션 중 해당 작업을 일시 중지하거나 이를 처리하는 추가 로직을 구현해야 해요.
Step 1: 새 컬렉션 만들기
첫 번째 단계는 새 임베딩을 저장할 새 컬렉션을 만드는 거예요. 벡터 크기와 유사도 함수가 새 모델과 호환되어야 해요.
client.create_collection(
collection_name=NEW_COLLECTION,
vectors_config=(
models.VectorParams(
size=512, # 새 임베딩 벡터의 크기
distance=models.Distance.COSINE # 새 모델의 유사도 함수
)
)
)
await client.createCollection(NEW_COLLECTION, {
vectors: {
size: 512, // 새 임베딩 벡터의 크기
distance: "Cosine", // 새 모델의 유사도 함수
},
});
client
.create_collection(
CreateCollectionBuilder::new(new_collection)
.vectors_config(VectorParamsBuilder::new(512, Distance::Cosine)), // 새 임베딩 벡터의 크기
)
.await?;
client.createCollectionAsync(NEW_COLLECTION,
VectorParams.newBuilder()
.setSize(512) // 새 임베딩 벡터의 크기
.setDistance(Distance.Cosine) // 새 모델의 유사도 함수
.build()).get();
await client.CreateCollectionAsync(
collectionName: NEW_COLLECTION,
vectorsConfig: new VectorParams { Size = 512, Distance = Distance.Cosine }
);
client.CreateCollection(context.Background(), &qdrant.CreateCollection{
CollectionName: NEW_COLLECTION,
VectorsConfig: qdrant.NewVectorsConfig(&qdrant.VectorParams{
Size: 512, // 새 임베딩 벡터의 크기
Distance: qdrant.Distance_Cosine,
}),
})
지금은 커스텀 샤딩, 복제 계수 등 컬렉션의 다른 설정을 바꿀 좋은 기회이기도 해요. 모델을 전환하는 것은 검색 성능을 개선할 좋은 기회가 될 수 있어요.
새로 만든 컬렉션은 비어 있고, 새 임베딩을 저장할 준비가 됐어요.
Step 2: 이중 쓰기 활성화
마이그레이션 중 두 컬렉션을 모두 최신 상태로 유지하려면, 모든 변경 사항을 두 컬렉션에 동시에 써요. 이렇게 하면 새 데이터나 기존 데이터의 업데이트가 두 컬렉션 모두에 반영돼요.
이상적으로 Qdrant의 데이터는 업데이트 큐를 읽는 업데이트 서비스에 의해 업데이트돼요. 이 서비스는 문서를 임베딩하고 Qdrant에 쓰는 책임이 있어요. 다음과 유사한 코드를 사용해요.
client.upsert(
collection_name=OLD_COLLECTION,
points=[
models.PointStruct(
id=1,
vector=models.Document(
text="Example document",
model=OLD_MODEL,
),
payload={"text": "Example document"}
)
]
)
await client.upsert(OLD_COLLECTION, {
points: [
{
id: 1,
vector: {
text: "Example document",
model: OLD_MODEL,
},
payload: { text: "Example document" },
},
],
});
client
.upsert_points(UpsertPointsBuilder::new(
old_collection,
vec![PointStruct::new(
1,
Document::new("Example document", old_model),
[("text", "Example document".into())],
)],
))
.await?;
client.upsertAsync(OLD_COLLECTION, List.of(
PointStruct.newBuilder()
.setId(id(1))
.setVectors(
vectors(
vector(
Document.newBuilder()
.setText("Example document")
.setModel(OLD_MODEL)
.build())))
.putAllPayload(Map.of("text", value("Example document")))
.build())).get();
await client.UpsertAsync(
collectionName: OLD_COLLECTION,
points: new List<PointStruct>
{
new()
{
Id = 1,
Vectors = new Document
{
Text = "Example document",
Model = OLD_MODEL
},
Payload = { ["text"] = "Example document" }
}
}
);
client.Upsert(context.Background(), &qdrant.UpsertPoints{
CollectionName: OLD_COLLECTION,
Points: []*qdrant.PointStruct{
{
Id: qdrant.NewIDNum(1),
Vectors: qdrant.NewVectorsDocument(&qdrant.Document{
Text: "Example document",
Model: OLD_MODEL,
}),
Payload: qdrant.NewValueMap(map[string]any{"text": "Example document"}),
},
},
})
새 컬렉션을 업데이트하려면 기존 컬렉션과 병렬로 새 컬렉션을 업데이트하는 두 번째 서비스를 배포해요. 이 서비스는 새 임베딩 모델로 문서를 인코딩해 새 컬렉션에 써요.
client.upsert(
collection_name=NEW_COLLECTION,
points=[
models.PointStruct(
id=1,
# 새 임베딩 모델로 문서를 인코딩
vector=models.Document(
text="Example document",
model=NEW_MODEL,
),
payload={"text": "Example document"}
)
]
)
await client.upsert(NEW_COLLECTION, {
points: [
{
id: 1,
// 새 임베딩 모델로 문서를 인코딩
vector: {
text: "Example document",
model: NEW_MODEL,
},
payload: { text: "Example document" },
},
],
});
client
.upsert_points(UpsertPointsBuilder::new(
new_collection,
vec![PointStruct::new(
1,
// 새 임베딩 모델로 문서를 인코딩
Document::new("Example document", new_model),
[("text", "Example document".into())],
)],
))
.await?;
client.upsertAsync(NEW_COLLECTION, List.of(
PointStruct.newBuilder()
.setId(id(1))
// 새 임베딩 모델로 문서를 인코딩
.setVectors(
vectors(
vector(
Document.newBuilder()
.setText("Example document")
.setModel(NEW_MODEL)
.build())))
.putAllPayload(Map.of("text", value("Example document")))
.build())).get();
await client.UpsertAsync(
collectionName: NEW_COLLECTION,
points: new List<PointStruct>
{
new()
{
Id = 1,
// 새 임베딩 모델로 문서를 인코딩
Vectors = new Document
{
Text = "Example document",
Model = NEW_MODEL
},
Payload = { ["text"] = "Example document" }
}
}
);
client.Upsert(context.Background(), &qdrant.UpsertPoints{
CollectionName: NEW_COLLECTION,
Points: []*qdrant.PointStruct{
{
Id: qdrant.NewIDNum(1),
// 새 임베딩 모델로 문서를 인코딩
Vectors: qdrant.NewVectorsDocument(&qdrant.Document{
Text: "Example document",
Model: NEW_MODEL,
}),
Payload: qdrant.NewValueMap(map[string]any{"text": "Example document"}),
},
},
})
두 작업이 모두 성공하도록 항상 보장하는 것이 좋은 습관이에요. 오류는 클라이언트 쪽에서 처리해야 해요. 오류를 로그나 "dead letter queue"에 저장해 나중에 처리할 수 있어요. 일시적인 오류는 나중에 재시도할 수 있어요. 다른 오류는 분석해 적절히 대응해야 해요.
업데이트 서비스 대신 모놀리식 애플리케이션을 사용한다면, 전환 기간 동안 두 컬렉션 모두에 쓰도록 애플리케이션 코드를 수정해야 해요. 문서 임베딩을 처리하는 코드에서 두 컬렉션 모두에 쓰는 로직을 추가해야 해요.
이 튜토리얼에서 설명한 방법은 upsert 작업에서만 작동해요. 예를 들어, delete 작업은 포인트가 아직 존재하지 않으면 새 컬렉션에서 실패하고, 그 포인트는 나중에 마이그레이션 프로세스에 의해 잘못 추가될 수 있어요. 컬렉션의 포인트를 수정하기 위해 다음 방법 중 하나를 사용한다면, 마이그레이션 중 해당 작업을 일시 중지하거나 처리하는 추가 로직을 구현해야 해요.
.delete- 지정된 포인트를 컬렉션에서 제거.update_vectors- 포인트의 지정된 벡터 업데이트.delete_vectors- 포인트의 지정된 벡터 삭제.set_payload- 지정된 포인트에 payload 값 설정.overwrite_payload- 지정된 포인트의 전체 payload를 새 payload로 덮어쓰기.delete_payload- 포인트의 지정된 키 payload 삭제.clear_payload- 지정된 포인트의 전체 payload 제거.batch_update_points- 포인트(해당 벡터와 payload 포함)에 대한 일괄 업데이트
정확한 메서드 이름은 언어마다 다를 수 있으므로 사용 중인 SDK 문서, 또는 HTTP/gRPC 정의를 참고해요.
이 변경 후에는 이중 쓰기 모드가 되어, 모든 변경 사항이 기존·새 컬렉션 모두에 쓰여요. 이를 통해 마이그레이션 중 두 컬렉션을 최신 상태로 유지할 수 있어요.
Step 3: 기존 포인트를 새 컬렉션으로 마이그레이션
이제 이중 쓰기 모드이므로, 기존 포인트를 기존 컬렉션에서 새 컬렉션으로 마이그레이션할 차례예요. 이는 일반적인 upsert 서비스와 병렬로 실행되는 별도의 프로세스로 수행할 수 있어요.
마이그레이션 프로세스는 기존 컬렉션에서 포인트를 읽고, 새 모델로 다시 임베딩한 뒤, 업데이트 서비스가 삽입한 기존 포인트를 덮어쓰지 않도록 주의하면서 새 컬렉션에 써요. 이러한 마이그레이션 프로세스의 코드는 다음과 같을 수 있어요.
last_offset = None
batch_size = 100 # 각 배치에서 읽을 포인트 수
reached_end = False
while not reached_end:
# 기존 컬렉션에서 다음 배치의 포인트 가져오기
records, last_offset = client.scroll(
collection_name=OLD_COLLECTION,
limit=batch_size,
offset=last_offset,
# 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
with_payload=True,
# 기존 벡터는 필요 없으므로 대역폭 절약
with_vectors=False,
)
# 새 모델로 포인트 다시 임베딩
points = [
models.PointStruct(
# 일관성 보장을 위해 원본 ID 유지
id=record.id,
# 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
# 새 임베딩 모델로 인코딩
vector=models.Document(
text=(record.payload or {}).get("text", ""),
model=NEW_MODEL,
),
# 원본 payload 유지
payload=record.payload
)
for record in records
]
# 다시 임베딩된 포인트를 새 컬렉션에 upsert
client.upsert(
collection_name=NEW_COLLECTION,
points=points,
# 이 ID의 포인트가 아직 없을 때만 삽입
update_mode=models.UpdateMode.INSERT_ONLY
)
# 컬렉션 끝에 도달했는지 확인
reached_end = (last_offset == None)
let lastOffset: number | string | undefined = undefined;
const batchSize = 100; // 각 배치에서 읽을 포인트 수
let reachedEnd = false;
while (!reachedEnd) {
// 기존 컬렉션에서 다음 배치의 포인트 가져오기
const scrollResult = await client.scroll(OLD_COLLECTION, {
limit: batchSize,
offset: lastOffset,
// 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
with_payload: true,
// 기존 벡터는 필요 없으므로 대역폭 절약
with_vector: false,
});
const records = scrollResult.points;
lastOffset = scrollResult.next_page_offset as number | string | undefined;
// 새 모델로 포인트 다시 임베딩
const points = records.map((record) => ({
// 일관성 보장을 위해 원본 ID 유지
id: record.id,
// 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
// 새 임베딩 모델로 인코딩
vector: {
text: ((record.payload?.text as string) ?? ""),
model: NEW_MODEL,
},
// 원본 payload 유지
payload: record.payload,
}));
// 다시 임베딩된 포인트를 새 컬렉션에 upsert
await client.upsert(NEW_COLLECTION, {
points,
// 이 ID의 포인트가 아직 없을 때만 삽입
update_mode: "insert_only" as const,
});
// 컬렉션 끝에 도달했는지 확인
reachedEnd = lastOffset == null;
}
let mut last_offset = None;
let batch_size = 100; // 각 배치에서 읽을 포인트 수
loop {
// 기존 컬렉션에서 다음 배치의 포인트 가져오기
let mut scroll_builder = ScrollPointsBuilder::new(old_collection)
.limit(batch_size)
// 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
.with_payload(true)
// 기존 벡터는 필요 없으므로 대역폭 절약
.with_vectors(false);
if let Some(offset) = last_offset {
scroll_builder = scroll_builder.offset(offset);
}
let scroll_result = client.scroll(scroll_builder).await?;
let records = scroll_result.result;
last_offset = scroll_result.next_page_offset;
// 새 모델로 포인트 다시 임베딩
let points: Vec<PointStruct> = records
.iter()
.map(|record| {
PointStruct::new(
// 일관성 보장을 위해 원본 ID 유지
record.id.clone().unwrap(),
// 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
// 새 임베딩 모델로 인코딩
Document::new(
record.payload.get("text")
.and_then(|v| v.as_str())
.map_or("", |v| v),
new_model,
),
// 원본 payload 유지
record.payload.clone(),
)
})
.collect();
// 다시 임베딩된 포인트를 새 컬렉션에 upsert
client
.upsert_points(
// 이 ID의 포인트가 아직 없을 때만 삽입
UpsertPointsBuilder::new(new_collection, points)
.update_mode(UpdateMode::InsertOnly),
)
.await?;
// 컬렉션 끝에 도달했는지 확인
if last_offset.is_none() {
break;
}
}
int batchSize = 100; // 각 배치에서 읽을 포인트 수
boolean reachedEnd = false;
// 기존 컬렉션에서 다음 배치의 포인트 가져오기
var scrollBuilder = ScrollPoints.newBuilder()
.setCollectionName(OLD_COLLECTION)
.setLimit(batchSize)
// 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
.setWithPayload(WithPayloadSelectorFactory.enable(true))
// 기존 벡터는 필요 없으므로 대역폭 절약
.setWithVectors(WithVectorsSelectorFactory.enable(false));
while (!reachedEnd) {
var scrollResult = client.scrollAsync(scrollBuilder.build()).get();
var records = scrollResult.getResultList();
// 새 모델로 포인트 다시 임베딩
List<PointStruct> points = new ArrayList<>();
for (var record : records) {
String text = record.getPayloadMap().containsKey("text")
? record.getPayloadMap().get("text").getStringValue()
: "";
points.add(
PointStruct.newBuilder()
// 일관성 보장을 위해 원본 ID 유지
.setId(record.getId())
// 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
// 새 임베딩 모델로 인코딩
.setVectors(
vectors(
vector(
Document.newBuilder()
.setText(text)
.setModel(NEW_MODEL)
.build())))
// 원본 payload 유지
.putAllPayload(record.getPayloadMap())
.build());
}
// 다시 임베딩된 포인트를 새 컬렉션에 upsert
client.upsertAsync(
UpsertPoints.newBuilder()
.setCollectionName(NEW_COLLECTION)
.addAllPoints(points)
// 이 ID의 포인트가 아직 없을 때만 삽입
.setUpdateMode(UpdateMode.InsertOnly)
.build()).get();
// 컬렉션 끝에 도달했는지 확인
if (scrollResult.hasNextPageOffset()) {
scrollBuilder.setOffset(scrollResult.getNextPageOffset());
} else {
reachedEnd = true;
}
}
PointId? lastOffset = null;
uint limit = 100; // 각 배치에서 읽을 포인트 수
bool reachedEnd = false;
while (!reachedEnd)
{
// 기존 컬렉션에서 다음 배치의 포인트 가져오기
var scrollResult = await client.ScrollAsync(
collectionName: OLD_COLLECTION,
limit: limit,
offset: lastOffset,
// 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
payloadSelector: true,
// 기존 벡터는 필요 없으므로 대역폭 절약
vectorsSelector: false
);
var records = scrollResult.Result;
lastOffset = scrollResult.NextPageOffset;
// 새 모델로 포인트 다시 임베딩
var points = new List<PointStruct>();
foreach (var record in records)
{
var text = record.Payload.ContainsKey("text")
? record.Payload["text"].StringValue
: "";
points.Add(new PointStruct
{
// 일관성 보장을 위해 원본 ID 유지
Id = record.Id,
// 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
// 새 임베딩 모델로 인코딩
Vectors = new Document
{
Text = text,
Model = NEW_MODEL
},
// 원본 payload 유지
Payload = { record.Payload }
});
}
// 다시 임베딩된 포인트를 새 컬렉션에 upsert
await client.UpsertAsync(
new()
{
CollectionName = NEW_COLLECTION,
Points = { points },
// 이 ID의 포인트가 아직 없을 때만 삽입
UpdateMode = UpdateMode.InsertOnly
}
);
// 컬렉션 끝에 도달했는지 확인
reachedEnd = (lastOffset == null);
}
var lastOffset *qdrant.PointId
batchSize := uint32(100) // 각 배치에서 읽을 포인트 수
reachedEnd := false
for !reachedEnd {
// 기존 컬렉션에서 다음 배치의 포인트 가져오기
scrollResult, err := client.Scroll(context.Background(), &qdrant.ScrollPoints{
CollectionName: OLD_COLLECTION,
Limit: qdrant.PtrOf(batchSize),
Offset: lastOffset,
// 벡터를 다시 임베딩하는 데 필요하므로 응답에 payload 포함
WithPayload: qdrant.NewWithPayload(true),
// 기존 벡터는 필요 없으므로 대역폭 절약
WithVectors: qdrant.NewWithVectors(false),
})
records := scrollResult
// 새 모델로 포인트 다시 임베딩
points := make([]*qdrant.PointStruct, len(records))
for idx, record := range records {
text := ""
if val, ok := record.Payload["text"]; ok {
text = val.GetStringValue()
}
points[idx] = &qdrant.PointStruct{
// 일관성 보장을 위해 원본 ID 유지
Id: record.Id,
// 임베딩의 원본 소스라고 가정하고 payload의 텍스트를
// 새 임베딩 모델로 인코딩
Vectors: qdrant.NewVectorsDocument(&qdrant.Document{
Text: text,
Model: NEW_MODEL,
}),
// 원본 payload 유지
Payload: record.Payload,
}
}
// 다시 임베딩된 포인트를 새 컬렉션에 upsert
client.Upsert(context.Background(), &qdrant.UpsertPoints{
CollectionName: NEW_COLLECTION,
Points: points,
// 이 ID의 포인트가 아직 없을 때만 삽입
UpdateMode: qdrant.UpdateMode_InsertOnly.Enum(),
})
// 컬렉션 끝에 도달했는지 확인
reachedEnd = (lastOffset == nil)
}
이 코드를 단계별로 분해하면:
- scroll을 사용해 기존 컬렉션에서 포인트를 100개씩 배치로 읽어요.
- 각 포인트 배치에 대해 새 임베딩 모델로 벡터를 다시 임베딩해요. 임베딩에 사용된 원본 텍스트가 payload의
text키에 저장되어 있다고 가정해요. - 다시 임베딩된 벡터로 포인트를 새 컬렉션에 upsert하며, 원본 ID와 payload를 유지해요. upsert는 insert-only 모드(버전 1.16 이상에서 사용 가능)를 사용해 포인트가 새 컬렉션에 이미 없을 때만 삽입되도록 보장해요. 이는 일반 업데이트 서비스의 최신 업데이트를 덮어쓰는 것을 방지해요.
마이그레이션 프로세스는 시간이 걸릴 수 있으며, 오프셋을 영속적으로 저장해 실패 시 마이그레이션을 재개할 수 있어요. 데이터베이스, 파일, 기타 영속 스토리지를 사용해 마지막 오프셋을 추적할 수 있어요. 그렇지만 조건부 upsert가 새 컬렉션의 어떤 포인트도 덮어쓰지 않기 때문에, 필요하면 마이그레이션을 처음부터 안전하게 다시 시작할 수 있어요.
마이그레이션 프로세스가 완료되고 기존 컬렉션의 모든 포인트가 새 컬렉션에 다시 임베딩·저장되면, 백엔드 애플리케이션의 구성 변경을 배포할 수 있어요. 두 가지 주요 변경이 필요해요.
- 컬렉션 이름. 이를 기존 컬렉션에서 새 컬렉션으로 바꿔요. 컬렉션 alias를 사용한다면 alias를 새 컬렉션을 가리키도록 전환해요.
- 임베딩 모델. 이를 기존 임베딩 모델에서 새 임베딩 모델로 바꿔요.
이 값들이 애플리케이션에 하드코딩되어 있다면 코드에서 직접 변경하고 애플리케이션의 새 버전을 배포해야 해요. 예를 들어 현재 검색 코드가 다음과 같다면:
results = client.query_points(
collection_name=OLD_COLLECTION,
query=models.Document(text="my query", model=OLD_MODEL),
limit=10,
)
const results = await client.query(OLD_COLLECTION, {
query: {
text: "my query",
model: OLD_MODEL,
},
limit: 10,
});
let results = client
.query(
QueryPointsBuilder::new(old_collection)
.query(Query::new_nearest(Document::new("my query", old_model)))
.limit(10),
)
.await?;
QueryPoints oldRequest =
QueryPoints.newBuilder()
.setCollectionName(OLD_COLLECTION)
.setQuery(
nearest(
Document.newBuilder()
.setText("my query")
.setModel(OLD_MODEL)
.build()))
.setLimit(10)
.build();
var results = client.queryAsync(oldRequest).get();
var results = await client.QueryAsync(
collectionName: OLD_COLLECTION,
query: new Document
{
Text = "my query",
Model = OLD_MODEL
},
limit: 10
);
results, err := client.Query(context.Background(), &qdrant.QueryPoints{
CollectionName: OLD_COLLECTION,
Query: qdrant.NewQueryDocument(&qdrant.Document{
Text: "my query",
Model: OLD_MODEL,
}),
Limit: qdrant.PtrOf(uint64(10)),
})
다음과 같이 변경해야 해요.
results = client.query_points(
collection_name=NEW_COLLECTION,
query=models.Document(text="my query", model=NEW_MODEL),
limit=10,
)
const resultsNew = await client.query(NEW_COLLECTION, {
query: {
text: "my query",
model: NEW_MODEL,
},
limit: 10,
});
let results = client
.query(
QueryPointsBuilder::new(new_collection)
.query(Query::new_nearest(Document::new("my query", new_model)))
.limit(10),
)
.await?;
QueryPoints newRequest =
QueryPoints.newBuilder()
.setCollectionName(NEW_COLLECTION)
.setQuery(
nearest(
Document.newBuilder()
.setText("my query")
.setModel(NEW_MODEL)
.build()))
.setLimit(10)
.build();
results = client.queryAsync(newRequest).get();
results = await client.QueryAsync(
collectionName: NEW_COLLECTION,
query: new Document
{
Text = "my query",
Model = NEW_MODEL
},
limit: 10
);
results, err = client.Query(context.Background(), &qdrant.QueryPoints{
CollectionName: NEW_COLLECTION,
Query: qdrant.NewQueryDocument(&qdrant.Document{
Text: "my query",
Model: NEW_MODEL,
}),
Limit: qdrant.PtrOf(uint64(10)),
})
Step 5: 마무리 (Wrapping Up)
애플리케이션이 새 컬렉션으로 전환되면, Step 2에서 구현한 이중 쓰기 모드를 비활성화해요. 이제부터 애플리케이션은 새 컬렉션에만 써야 해요.
이제 모든 검색이 새 임베딩을 사용해 수행돼요. 기존 컬렉션이 더 이상 필요하지 않다면 안전하게 삭제할 수 있어요. 필요할 때 롤백할 수 있도록 기존 컬렉션의 스냅샷을 유지하는 것이 좋아요.
명명된 벡터로 마이그레이션 (Migrate Using Named Vectors)
컬렉션이 명명된 벡터를 사용하고 배포 버전이 1.18 이상이라면, 두 번째 컬렉션을 만들지 않고 새 임베딩 모델로 마이그레이션할 수 있어요. 대신 새 모델을 기존 컬렉션 스키마의 추가 명명된 벡터로 추가하고, 백그라운드에서 포인트를 다시 임베딩한 뒤, 검색 쿼리의 using 매개변수를 전환하고, 기존 명명된 벡터를 삭제하면 돼요.
이 방법은 컬렉션이 명명된 벡터로 생성되었고 배포 버전이 1.18 이상일 때만 작동해요. 그렇지 않다면 블루-그린 마이그레이션을 대신 사용해요.
Step 1: 새 명명된 벡터 추가
새 모델의 벡터 스키마를 기존 컬렉션에 추가해요. 이는 스키마 전용 작업이에요. 세그먼트가 재구축되지도 않고 기존 포인트 데이터도 수정되지 않아요. 새 벡터는 즉시 쿼리할 수 있지만, 포인트에 값이 채워질 때까지는 검색이 결과를 반환하지 않아요.
client.create_vector_name(
collection_name=COLLECTION,
vector_name=NEW_VECTOR,
vector_name_config=models.DenseVectorNameConfig(
dense=models.DenseVectorConfig(
size=512, # 새 임베딩 벡터의 크기
distance=models.Distance.COSINE # 새 모델의 유사도 함수
)
),
)
await client.createVectorName(COLLECTION, NEW_VECTOR, {
dense: {
size: 512, // 새 임베딩 벡터의 크기
distance: "Cosine", // 새 모델의 유사도 함수
},
});
client
.create_vector_name(
CreateVectorNameRequestBuilder::new(
collection,
new_vector,
DenseVectorCreationConfigBuilder::new(512, Distance::Cosine), // 새 임베딩 벡터의 크기
),
)
.await?;
client
.createVectorNameAsync(
CreateVectorNameRequest.newBuilder()
.setCollectionName(COLLECTION)
.setVectorName(NEW_VECTOR)
.setDenseConfig(
DenseVectorCreationConfig.newBuilder()
.setSize(512) // 새 임베딩 벡터의 크기
.setDistance(Distance.Cosine) // 새 모델의 유사도 함수
.build())
.build())
.get();
await client.CreateVectorNameAsync(new()
{
CollectionName = COLLECTION,
VectorName = NEW_VECTOR,
DenseConfig = new() { Size = 512, Distance = Distance.Cosine }
});
client.CreateVectorName(context.Background(), &qdrant.CreateVectorNameRequest{
CollectionName: COLLECTION,
VectorName: NEW_VECTOR,
VectorConfig: &qdrant.CreateVectorNameRequest_DenseConfig{
DenseConfig: &qdrant.DenseVectorCreationConfig{
Size: 512, // 새 임베딩 벡터의 크기
Distance: qdrant.Distance_Cosine,
},
},
})
Step 2: 이중 쓰기 활성화
upsert 서비스를 업데이트해 각 문서를 두 모델 모두로 임베딩하고, 매 upsert마다 두 명명된 벡터를 모두 써요.
client.upsert(
collection_name=COLLECTION,
points=[
models.PointStruct(
id=1,
vector={
OLD_VECTOR: models.Document(
text="Example document",
model=OLD_MODEL,
),
NEW_VECTOR: models.Document(
text="Example document",
model=NEW_MODEL,
),
},
payload={"text": "Example document"}
)
]
)
await client.upsert(COLLECTION, {
points: [
{
id: 1,
vector: {
[OLD_VECTOR]: {
text: "Example document",
model: OLD_MODEL,
},
[NEW_VECTOR]: {
text: "Example document",
model: NEW_MODEL,
},
},
payload: { text: "Example document" },
},
],
});
client
.upsert_points(UpsertPointsBuilder::new(
collection,
vec![PointStruct::new(
1,
NamedVectors::default()
.add_vector(
old_vector,
Document {
text: "Example document".into(),
model: old_model.into(),
..Default::default()
},
)
.add_vector(
new_vector,
Document {
text: "Example document".into(),
model: new_model.into(),
..Default::default()
},
),
[("text", "Example document".into())],
)],
))
.await?;
client.upsertAsync(COLLECTION, List.of(
PointStruct.newBuilder()
.setId(id(1))
.setVectors(
namedVectors(
Map.of(
OLD_VECTOR, vector(
Document.newBuilder()
.setText("Example document")
.setModel(OLD_MODEL)
.build()),
NEW_VECTOR, vector(
Document.newBuilder()
.setText("Example document")
.setModel(NEW_MODEL)
.build()))))
.putAllPayload(Map.of("text", value("Example document")))
.build())).get();
await client.UpsertAsync(
collectionName: COLLECTION,
points: new List<PointStruct>
{
new()
{
Id = 1,
Vectors = new Dictionary<string, Vector>
{
[OLD_VECTOR] = new Document { Text = "Example document", Model = OLD_MODEL },
[NEW_VECTOR] = new Document { Text = "Example document", Model = NEW_MODEL },
},
Payload = { ["text"] = "Example document" }
}
}
);
client.Upsert(context.Background(), &qdrant.UpsertPoints{
CollectionName: COLLECTION,
Points: []*qdrant.PointStruct{
{
Id: qdrant.NewIDNum(1),
Vectors: qdrant.NewVectorsMap(map[string]*qdrant.Vector{
OLD_VECTOR: qdrant.NewVectorDocument(&qdrant.Document{
Text: "Example document",
Model: OLD_MODEL,
}),
NEW_VECTOR: qdrant.NewVectorDocument(&qdrant.Document{
Text: "Example document",
Model: NEW_MODEL,
}),
}),
Payload: qdrant.NewValueMap(map[string]any{"text": "Example document"}),
},
},
})
이 시점부터 모든 새 포인트나 업데이트된 포인트는 두 임베딩을 모두 갖고 있어요.
Step 3: 기존 포인트 다시 임베딩
컬렉션을 스크롤하고 각 기존 포인트의 새 명명된 벡터만 업데이트하는 백그라운드 프로세스를 실행해요. upsert 대신 update_vectors를 사용하므로 각 포인트의 기존 명명된 벡터와 payload는 변경되지 않아요.
last_offset = None
batch_size = 100
reached_end = False
while not reached_end:
records, last_offset = client.scroll(
collection_name=COLLECTION,
limit=batch_size,
offset=last_offset,
with_payload=True,
with_vectors=False,
)
# 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
client.update_vectors(
collection_name=COLLECTION,
points=[
models.PointVectors(
id=record.id,
vector={
NEW_VECTOR: models.Document(
text=(record.payload or {}).get("text", ""),
model=NEW_MODEL,
)
},
)
for record in records
],
)
reached_end = last_offset is None
let reEmbedLastOffset: number | string | undefined = undefined;
const reEmbedBatchSize = 100;
let reEmbedReachedEnd = false;
while (!reEmbedReachedEnd) {
const reEmbedScrollResult = await client.scroll(COLLECTION, {
limit: reEmbedBatchSize,
offset: reEmbedLastOffset,
with_payload: true,
with_vector: false,
});
const records = reEmbedScrollResult.points;
reEmbedLastOffset = reEmbedScrollResult.next_page_offset as number | string | undefined;
// 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
await client.updateVectors(COLLECTION, {
points: records.map((record) => ({
id: record.id,
vector: {
[NEW_VECTOR]: {
text: ((record.payload?.text as string) ?? ""),
model: NEW_MODEL,
},
},
})),
});
reEmbedReachedEnd = reEmbedLastOffset == null;
}
let mut last_offset = None;
let batch_size = 100;
loop {
let mut scroll_builder = ScrollPointsBuilder::new(collection)
.limit(batch_size)
.with_payload(true)
.with_vectors(false);
if let Some(offset) = last_offset {
scroll_builder = scroll_builder.offset(offset);
}
let scroll_result = client.scroll(scroll_builder).await?;
let records = scroll_result.result;
last_offset = scroll_result.next_page_offset;
// 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
let point_vectors: Vec<PointVectors> = records
.iter()
.map(|record| PointVectors {
id: record.id.clone(),
vectors: Some(
HashMap::<String, Document>::from([(
new_vector.to_string(),
Document::new(
record.payload.get("text")
.and_then(|v| v.as_str())
.map_or("", |v| v),
new_model,
),
)])
.into(),
),
})
.collect();
client
.update_vectors(UpdatePointVectorsBuilder::new(collection, point_vectors))
.await?;
if last_offset.is_none() {
break;
}
}
int reEmbedBatchSize = 100;
boolean reEmbedReachedEnd = false;
var reEmbedScrollBuilder = ScrollPoints.newBuilder()
.setCollectionName(COLLECTION)
.setLimit(reEmbedBatchSize)
.setWithPayload(WithPayloadSelectorFactory.enable(true))
.setWithVectors(WithVectorsSelectorFactory.enable(false));
while (!reEmbedReachedEnd) {
var reEmbedScrollResult = client.scrollAsync(reEmbedScrollBuilder.build()).get();
var reEmbedRecords = reEmbedScrollResult.getResultList();
List<PointVectors> pointVectors = new ArrayList<>();
for (var record : reEmbedRecords) {
String text = record.getPayloadMap().containsKey("text")
? record.getPayloadMap().get("text").getStringValue()
: "";
// 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
pointVectors.add(
PointVectors.newBuilder()
.setId(record.getId())
.setVectors(
namedVectors(
Map.of(
NEW_VECTOR, vector(
Document.newBuilder()
.setText(text)
.setModel(NEW_MODEL)
.build()))))
.build());
}
client.updateVectorsAsync(COLLECTION, pointVectors).get();
if (reEmbedScrollResult.hasNextPageOffset()) {
reEmbedScrollBuilder.setOffset(reEmbedScrollResult.getNextPageOffset());
} else {
reEmbedReachedEnd = true;
}
}
PointId? reEmbedLastOffset = null;
uint reEmbedBatchSize = 100;
bool reEmbedReachedEnd = false;
while (!reEmbedReachedEnd)
{
var reEmbedScrollResult = await client.ScrollAsync(
collectionName: COLLECTION,
limit: reEmbedBatchSize,
offset: reEmbedLastOffset,
payloadSelector: true,
vectorsSelector: false
);
var reEmbedRecords = reEmbedScrollResult.Result;
reEmbedLastOffset = reEmbedScrollResult.NextPageOffset;
var pointVectors = new List<PointVectors>();
foreach (var record in reEmbedRecords)
{
var text = record.Payload.ContainsKey("text")
? record.Payload["text"].StringValue
: "";
// 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
pointVectors.Add(new PointVectors
{
Id = record.Id,
Vectors = new Dictionary<string, Vector>
{
[NEW_VECTOR] = new Document { Text = text, Model = NEW_MODEL }
}
});
}
await client.UpdateVectorsAsync(collectionName: COLLECTION, points: pointVectors);
reEmbedReachedEnd = (reEmbedLastOffset == null);
}
var reEmbedLastOffset *qdrant.PointId
reEmbedBatchSize := uint32(100)
reEmbedReachedEnd := false
for !reEmbedReachedEnd {
reEmbedScrollResult, err := client.Scroll(context.Background(), &qdrant.ScrollPoints{
CollectionName: COLLECTION,
Limit: qdrant.PtrOf(reEmbedBatchSize),
Offset: reEmbedLastOffset,
WithPayload: qdrant.NewWithPayload(true),
WithVectors: qdrant.NewWithVectors(false),
})
reEmbedRecords := reEmbedScrollResult
pointVectors := make([]*qdrant.PointVectors, len(reEmbedRecords))
for idx, record := range reEmbedRecords {
text := ""
if val, ok := record.Payload["text"]; ok {
text = val.GetStringValue()
}
// 각 포인트의 새 벡터만 업데이트; 기존 벡터와 payload는 건드리지 않음
pointVectors[idx] = &qdrant.PointVectors{
Id: record.Id,
Vectors: qdrant.NewVectorsMap(map[string]*qdrant.Vector{
NEW_VECTOR: qdrant.NewVectorDocument(&qdrant.Document{
Text: text,
Model: NEW_MODEL,
}),
}),
}
}
client.UpdateVectors(context.Background(), &qdrant.UpdatePointVectors{
CollectionName: COLLECTION,
Points: pointVectors,
})
reEmbedReachedEnd = (reEmbedLastOffset == nil)
}
upsert 서비스와 마이그레이션 프로세스의 동시 쓰기는 안전해요. 두 프로세스 모두 같은 모델을 사용해 같은 payload 텍스트에서 새 벡터를 파생하므로, 포인트를 동시에 처리해도 같은 결과를 만들어요.
Step 4: 검색을 새 벡터로 전환
모든 포인트에 새 명명된 벡터 값이 있으면 쿼리 로직을 변경해요.
using매개변수를 기존 벡터에서 새 벡터로 전환해요.- 임베딩 모델을 기존 모델에서 새 모델로 전환해요.
이전:
results = client.query_points(
collection_name=COLLECTION,
query=models.Document(text="my query", model=OLD_MODEL),
using=OLD_VECTOR,
limit=10,
)
const oldVectorResults = await client.query(COLLECTION, {
query: {
text: "my query",
model: OLD_MODEL,
},
using: OLD_VECTOR,
limit: 10,
});
let old_vector_results = client
.query(
QueryPointsBuilder::new(collection)
.query(Query::new_nearest(Document::new("my query", old_model)))
.using(old_vector)
.limit(10),
)
.await?;
var oldVectorResults = client.queryAsync(
QueryPoints.newBuilder()
.setCollectionName(COLLECTION)
.setQuery(
nearest(
Document.newBuilder()
.setText("my query")
.setModel(OLD_MODEL)
.build()))
.setUsing(OLD_VECTOR)
.setLimit(10)
.build()).get();
var oldVectorResults = await client.QueryAsync(
collectionName: COLLECTION,
query: new Document { Text = "my query", Model = OLD_MODEL },
usingVector: OLD_VECTOR,
limit: 10
);
oldVectorResults, err := client.Query(context.Background(), &qdrant.QueryPoints{
CollectionName: COLLECTION,
Query: qdrant.NewQueryDocument(&qdrant.Document{
Text: "my query",
Model: OLD_MODEL,
}),
Using: qdrant.PtrOf(OLD_VECTOR),
Limit: qdrant.PtrOf(uint64(10)),
})
이후:
results = client.query_points(
collection_name=COLLECTION,
query=models.Document(text="my query", model=NEW_MODEL),
using=NEW_VECTOR,
limit=10,
)
const newVectorResults = await client.query(COLLECTION, {
query: {
text: "my query",
model: NEW_MODEL,
},
using: NEW_VECTOR,
limit: 10,
});
let new_vector_results = client
.query(
QueryPointsBuilder::new(collection)
.query(Query::new_nearest(Document::new("my query", new_model)))
.using(new_vector)
.limit(10),
)
.await?;
var newVectorResults = client.queryAsync(
QueryPoints.newBuilder()
.setCollectionName(COLLECTION)
.setQuery(
nearest(
Document.newBuilder()
.setText("my query")
.setModel(NEW_MODEL)
.build()))
.setUsing(NEW_VECTOR)
.setLimit(10)
.build()).get();
var newVectorResults = await client.QueryAsync(
collectionName: COLLECTION,
query: new Document { Text = "my query", Model = NEW_MODEL },
usingVector: NEW_VECTOR,
limit: 10
);
newVectorResults, err := client.Query(context.Background(), &qdrant.QueryPoints{
CollectionName: COLLECTION,
Query: qdrant.NewQueryDocument(&qdrant.Document{
Text: "my query",
Model: NEW_MODEL,
}),
Using: qdrant.PtrOf(NEW_VECTOR),
Limit: qdrant.PtrOf(uint64(10)),
})
Step 5: 이중 쓰기 비활성화 및 기존 명명된 벡터 삭제
모든 검색 트래픽이 새 벡터를 사용하면, upsert 서비스를 변경해 앞으로는 새 벡터에만 쓰도록 해요. 다음으로 컬렉션에서 기존 명명된 벡터를 삭제해요.
client.delete_vector_name(
collection_name=COLLECTION,
vector_name=OLD_VECTOR,
)
await client.deleteVectorName(COLLECTION, OLD_VECTOR);
client
.delete_vector_name(DeleteVectorNameRequestBuilder::new(
collection,
old_vector,
))
.await?;
client
.deleteVectorNameAsync(
DeleteVectorNameRequest.newBuilder()
.setCollectionName(COLLECTION)
.setVectorName(OLD_VECTOR)
.build())
.get();
await client.DeleteVectorNameAsync(new()
{
CollectionName = COLLECTION,
VectorName = OLD_VECTOR
});
client.DeleteVectorName(context.Background(), &qdrant.DeleteVectorNameRequest{
CollectionName: COLLECTION,
VectorName: OLD_VECTOR,
})
기존 벡터의 스토리지는 다음 optimizer 실행 후 회수돼요. 모든 포인트 ID, payload, 새 명명된 벡터는 그대로 유지돼요.