Vector Retrieval Service for Milvus:Crie um pipeline de produção de ativos e recuperação multimodal para minisséries com o Alibaba Cloud Milvus

Visão geral da solução

Minisséries AIGC, dramas verticais curtos e vídeos promocionais são produzidos em ritmo acelerado. Uma equipe pode entregar dezenas ou até centenas de vídeos finalizados por semana, e cada vídeo exige a geração e triagem de um grande volume de keyframes, storyboards e ativos candidatos. Sem busca semântica em uma biblioteca acumulada, a localização de ativos anteriores depende apenas de nomes de arquivos e da memória da equipe.

O Alibaba Cloud Milvus orquestra esse pipeline com quatro funções de IA, divididas em duas fases — geração e recuperação — e cinco etapas. A tabela a seguir mapeia os quatro estágios típicos de um pipeline de produção para essas etapas.

As quatro funções de IA utilizadas nessas etapas são:

• AI_IMAGE_EDIT (Etapa 1) — Geração e edição de imagens. Suporta edição de imagem única e fusão de múltiplas imagens de referência. Converte uma imagem de referência do storyboard em uma imagem de keyframe com base no roteiro.

• AI_VIDEO_EDIT (Etapa 2) — Geração de vídeo a partir de imagem e de texto. Executa como uma tarefa assíncrona. Gera uma cena com movimento usando um keyframe como primeiro quadro.

• AI_MULTI_MODAL_GENERATE (Etapa 3) — Compreensão de conteúdo multimodal. Gera títulos, legendas e tags para imagens e vídeos. Produz as legendas objetivas usadas na recuperação.

• AI_EMBEDDING (Etapas 4 e 5) — Vetorização na escrita. Mapeia textos, imagens e vídeos no mesmo espaço vetorial, o que viabiliza a busca entre modalidades.

  Três mecanismos sustentam este pipeline:

• Ingestão durante a geração — Keyframes e cenas com movimento são ingeridos junto com suas legendas e tags, garantindo que todo item gerado possua informações estruturadas e um vetor associado.

• Espaço vetorial único — Imagens e vídeos compartilham o mesmo modelo multimodal e o mesmo campo vetorial, permitindo buscas cruzadas como texto para imagem e imagem para vídeo.

• Vetorização na escrita e inferência na consulta — Sua aplicação não precisa chamar o modelo de embedding diretamente.

Pré-requisitos

• Uma instância do Milvus 2.6 criada. As funções de IA exigem o kernel 2.6. Não é necessário vincular um service de modelo separadamente após criar a instância.

• pymilvus instalado. Os exemplos deste tutorial foram validados com pymilvus 3.0.0.

• Para acessar a instância pela Internet, ative Public Network Access na aba Security Configuration da página de detalhes da instância e adicione o IP de saída do seu cliente à lista de permissões de acesso à rede pública.

• (Recomendado para pipelines de produção) Um bucket próprio do Object Storage Service (OSS) para persistir imagens e vídeos gerados antes da ingestão.

Considerações

Antes de executar o pipeline, observe os pontos abaixo:

• Porta do endpoint — A API RESTful e o gRPC compartilham a porta 19530. Especifique explicitamente a porta no endpoint, por exemplo http://c-xxx.milvus.aliyuncs.com:19530. Se omitir a porta, a requisição será direcionada à porta 80 por padrão e a conexão atingirá o tempo limite.

• Validade das URLs geradas — As URLs retornadas pela geração de imagens e vídeos são assinadas pelo OSS e, em ambiente de teste, permanecem válidas por cerca de 24 horas. Persista os ativos gerados em seu próprio OSS antes da ingestão e grave a URL de longa duração em media_ref.

• URLs de mídia no lado da consulta — Certifique-se de que a URL de mídia passada na consulta esteja acessível.

• Um campo vetorial, um modelo — Imagens e vídeos devem residir no mesmo campo vetorial e usar o mesmo modelo para permitir buscas cruzadas no mesmo espaço vetorial.

• Parâmetros de função multimodal — Uma função multimodal deve incluir "is_multimodal": "true", que é um valor do tipo string. O dim do campo vetorial deve corresponder ao dim nos parâmetros da função.

• Flush antes da busca — Chame flush() após gravar dados. Caso contrário, uma busca executada imediatamente depois poderá retornar um resultado vazio.

• Valores medidos — As durações, pontuações de similaridade e legendas mostradas neste tutorial foram medidas em ambiente de teste. Considere-os valores de referência, não especificações do product.

Prepare o código compartilhado

Este tutorial utiliza dois métodos de acesso. As etapas 1 a 3 chamam as funções de geração e compreensão via API RESTful em /v2/vectordb/ai/* através do utilitário post_json. As etapas 4 e 5 criam a coleção, ingerem dados e realizam buscas com o MilvusClient do pymilvus.

Os blocos de código Python neste tutorial formam um único script que deve ser executado em ordem numa sessão Python. Etapas posteriores consomem variáveis produzidas pelas anteriores: keyframe_url, video_url, image_caption, video_caption, client e collection_name. Para executar uma etapa isoladamente, forneça manualmente os valores dessas variáveis anteriores.

O código a seguir contém as configurações de conexão e o utilitário de chamada REST. Substitua MILVUS_ENDPOINT e MILVUS_TOKEN pelas informações da sua própria instância.

from __future__ import annotations

import json
import time
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient

# ==== Global settings (REST and gRPC share port 19530) ====
MILVUS_ENDPOINT = "c-xxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = f"http://{MILVUS_ENDPOINT}"
MILVUS_URI = MILVUS_REST_BASE_URL

def post_json(path: str, body: dict[str, Any], timeout: int = 200) -> tuple[int, dict[str, Any]]:
    """REST call utility shared by Steps 1 to 3."""
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        return exc.code, json.loads(exc.read().decode("utf-8"))

O escopo deste bloco varia conforme o elemento. As configurações de conexão MILVUS_ENDPOINT, MILVUS_TOKEN e MILVUS_URI são usadas em todas as cinco etapas, incluindo o MilvusClient na Etapa 4. A função post_json é usada apenas nas Etapas 1 a 3. Seu parâmetro timeout=200 define o tempo limite HTTP do lado do cliente, em segundos, para cada chamada REST.

Etapa 1: Gere a imagem de keyframe

Pegue uma imagem de referência do storyboard, substitua o fundo ou ajuste o estilo com base no roteiro e produza uma imagem de keyframe que servirá como primeiro quadro para a geração de vídeo. O roteiro do storyboard é uma entrada manual: no código abaixo, corresponde ao valor de params.prompt.

# ========== Step 1: Keyframe image generation AI_IMAGE_EDIT (wan2.7-image) ==========
IMAGE_URL = "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"
PROMPT = (
    "On a rainy night, a woman stands under an umbrella at the entrance of a convenience store, "
    "with warm light from inside the store falling on her face, cinematic composition, keep the main subject."
)
status, data = post_json(
    "/v2/vectordb/ai/image_edit",
    {
        "model_name": "wan2.7-image",
        "texts": [IMAGE_URL],
        "params": {"prompt": PROMPT, "n": 1, "size": "1024*1024",
                   "watermark": False, "timeout_sec": 180},
    },
)
assert status == 200 and data.get("code") == 0, data
keyframe_url = data["data"]["output"]["outputs"][0]
print(f"Keyframe image URL: {keyframe_url}")

Para entrada de imagem única, passe a URL da imagem de referência em texts e escreva a instrução de edição em params.prompt. n especifica a quantidade de imagens a gerar, sendo 1 neste exemplo.

O prompt deste exemplo termina com keep the main subject, preservando os elementos principais da imagem de referência no keyframe. Como a imagem original contém uma mulher e um cachorro, as legendas geradas na Etapa 3 mencionam o animal, mesmo que o prompt não o descreva explicitamente.

Etapa 2: Gere a cena com movimento a partir do keyframe

Utilize o keyframe como primeiro quadro para gerar uma cena com movimento. A geração de vídeo é uma tarefa assíncrona de longa duração: crie a tarefa para obter um task_id e depois consulte o status periodicamente.

AI_VIDEO_EDIT suporta geração tanto de imagem para vídeo quanto de texto para vídeo. Este tutorial aborda apenas a geração de imagem para vídeo, motivo pelo qual o keyframe é passado em media com type definido como first_frame.

# ========== Step 2: Image-to-video generation AI_VIDEO_EDIT (happyhorse-1.1-i2v), asynchronous task ==========
VIDEO_MODEL = "happyhorse-1.1-i2v"  # The creation and query requests must use the same model name

# Use the keyframe from Step 1 as the first frame (media.type=first_frame)
status, data = post_json(
    "/v2/vectordb/ai/video_edit",
    {
        "model_name": VIDEO_MODEL,
        "prompt": "The shot slowly pushes in, rain falls, warm light flickers gently on the umbrella, keep the main subject from the first frame.",
        "media": [{"type": "first_frame", "url": keyframe_url}],
        "params": {"resolution": "720P", "audio_setting": "none",
                   "watermark": False, "timeout_sec": 180},
    },
)
assert status == 200 and data.get("code") == 0, data
task_id = data["data"]["output"]["task_id"]
print(f"Task created  task_id = {task_id}")

# Poll for the result: tasks/describe queries one task_id at a time, with provider and the same model_name
video_url = None
for attempt in range(60):  # Local polling limit. When it is reached, only record task_id; do not treat it as a server-side failure
    status, data = post_json(
        "/v2/vectordb/ai/tasks/describe",
        {"provider": "aliyun_milvus", "model_name": VIDEO_MODEL, "task_id": task_id},
    )
    task_status = data.get("data", {}).get("output", {}).get("task_status")
    print(f"  Poll #{attempt:<2} ->  {task_status}")
    if task_status == "SUCCEEDED":
        video_url = data["data"]["output"]["video_url"]
        break
    if task_status in ("FAILED", "CANCELED"):
        raise RuntimeError(f"task ended in {task_status}: {json.dumps(data, ensure_ascii=False)}")
    time.sleep(10)
assert video_url, f"Local polling limit reached. Query again later with task_id={task_id}"
print(f"Motion shot video_url: {video_url}")

A requisição para tasks/describe deve conter o mesmo model_name e provider da requisição de criação, consultando um task_id por vez. O corpo da resposta contém três campos: task_id, task_status e video_url. A tabela a seguir descreve os valores de task_status retornados durante a consulta.

Uma tarefa de imagem para vídeo em 720P com 5 segundos leva cerca de 2 a 3 minutos em ambiente de teste, período em que task_status continua retornando UNKNOWN. O loop de consulta deste exemplo permite até 60 tentativas com intervalos de 10 segundos, configurando um limite local de aproximadamente 10 minutos, e não um tempo limite do servidor.

Se o loop atingir esse limite local, a tarefa ainda estará em execução no servidor. Não considere a interrupção como falha: mantenha o task_id impresso pela chamada de criação e chame /v2/vectordb/ai/tasks/describe posteriormente com o mesmo provider, o mesmo model_name e esse task_id para recuperar a video_url.

params.resolution especifica uma faixa de contagem de pixels, não dimensões fixas. As dimensões reais de saída dependem da proporção do primeiro quadro. Neste exemplo, o primeiro quadro é um quadrado de 1024×1024, então 720P produz uma saída real de 960×960 (contagem total de pixels equivalente a 1280×720), com duração de cerca de 5 segundos. Para saída paisagem 16:9, utilize um primeiro quadro com a proporção correspondente.

Etapa 3: Gere legendas e tags para os ativos

Antes de ingerir ativos, utilize um modelo grande multimodal para gerar uma legenda objetiva de recuperação para cada imagem e vídeo. Use a legenda como source para título, resumo ou tags e ingira-a juntamente com o ativo na Etapa 4.

# ========== Step 3: Asset understanding AI_MULTI_MODAL_GENERATE (qwen3.7-plus) ==========
def describe_media(url: str, media_type: str, prompt: str) -> str:
    status, data = post_json(
        "/v2/vectordb/ai/multi_modal_generate",
        {
            "model_name": "qwen3.7-plus",
            "texts": [url],
            "params": {"media_type": media_type, "prompt": prompt, "temperature": 0},
        },
    )
    assert status == 200 and data.get("code") == 0, data
    return data["data"]["output"]["outputs"][0]

image_caption = describe_media(
    keyframe_url, "image",
    "Describe the subject, scene, and color tone of this keyframe image in one objective sentence, for text-to-image search.",
)
video_caption = describe_media(
    video_url, "video",
    "Describe the subject, action, and visual mood of this video in one objective sentence, for shot-level retrieval.",
)
print(f"[Image caption] {image_caption}")
print(f"[Video caption] {video_caption}")

Use media_type=image para imagens e media_type=video para vídeos. O campo prompt é obrigatório. As legendas a seguir foram geradas em ambiente de teste:

[Image caption] On a rainy street at night, a woman in a plaid shirt stands under an umbrella at the
                entrance of a convenience store, with a golden retriever sitting beside her. The overall
                tone is cool blue, and the warm light reflects off the wet ground, creating a quiet,
                slightly melancholy mood.
[Video caption] The video shows a rainy street at night, where a woman in a plaid shirt holding a
                transparent umbrella stands with a yellow dog outside a lit storefront. The shot moves
                from a wide view to a close-up of the woman's face, and the overall tone is cool with
                wet reflections.

A qualidade da legenda determina diretamente o recall da busca subsequente de texto para imagem. No prompt, solicite explicitamente uma "descrição objetiva do sujeito, cena, tonalidade de cor e ação", além de outras dimensões usadas no momento da recuperação, evitando legendas com julgamentos subjetivos.

Etapa 4: Ingera os ativos em uma coleção multimodal

Crie uma coleção de ativos multimodais com vetorização na escrita e grave a URL da imagem e a URL do vídeo juntamente com as legendas e tags. O Milvus gera vetores automaticamente para os ativos e realiza a ingestão.

O exemplo exclui qualquer coleção existente com o mesmo nome antes de criar a nova. Portanto, reexecutar o script recria a coleção do zero e remove os ativos ingeridos em execuções anteriores.

# ========== Step 4: Create a multimodal collection and ingest vectors AI_EMBEDDING (qwen3-vl-embedding) ==========
EMBED_MODEL = "qwen3-vl-embedding"
VECTOR_DIM = 2560  # Multimodal dimension of qwen3-vl-embedding. The dimension of the vector field must match dim

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "aigc_assets_multimodal"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("media_type", DataType.VARCHAR, max_length=16)     # image / video
schema.add_field("media_ref", DataType.VARCHAR, max_length=4096)    # Image or video URL used for vectorization
schema.add_field("caption", DataType.VARCHAR, max_length=2048)      # Caption generated in Step 3
schema.add_field("tags", DataType.VARCHAR, max_length=512)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)

# When media_ref is written, qwen3-vl-embedding is called automatically to generate a 2560-dimensional vector
schema.add_function(
    Function(
        name="embed_media",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["media_ref"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": EMBED_MODEL,
            "dim": VECTOR_DIM,
            "is_multimodal": "true",
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

# Vectorize and ingest the output of Steps 1 to 3 (url + caption + tags)
client.insert(
    collection_name,
    [
        {"media_type": "image", "media_ref": keyframe_url,
         "caption": image_caption, "tags": "keyframe,generated image"},
        {"media_type": "video", "media_ref": video_url,
         "caption": video_caption, "tags": "motion shot,generated video"},
    ],
)
client.flush(collection_name)
client.load_collection(collection_name)
print(f"Images and videos are vectorized and ingested. Collection: {collection_name}")

Os parâmetros da função definem is_multimodal como "true" e dim como 2560, correspondendo à dimensão do campo embedding. A imagem e o vídeo são gravados no mesmo campo embedding através do mesmo modelo, e o script chama flush() antes de carregar a coleção. Para entender as restrições por trás dessas três configurações, consulte a seção Considerações.

Efeitos de uma URL gerada expirada — As URLs produzidas nas Etapas 1 e 2 são URLs assinadas temporárias, conforme descrito em Considerações. Se você gravar tal URL em media_ref, a expiração causará dois efeitos.

Portanto, em ambiente de produção, persista os ativos gerados em seu próprio OSS antes da ingestão e grave a URL de longa duração em media_ref.

Resultado esperado: O script imprime Images and videos are vectorized and ingested. Collection: aigc_assets_multimodal. Nesse ponto, a coleção está carregada e contém duas entidades: uma imagem e um vídeo. Não prossiga para a Etapa 5 até que essa mensagem apareça. Se a Etapa 5 retornar um resultado vazio posteriormente, verifique se flush() foi executado antes da busca.

Etapa 5: Pesquise na biblioteca de ativos

Após a ingestão dos ativos, use o mesmo modelo multimodal para mapear a consulta no mesmo espaço vetorial e recuperar os ativos mais similares para reutilização. Como a coleção já está vinculada à função qwen3-vl-embedding, passe texto bruto ou uma URL de mídia diretamente no parâmetro data de search, e o Milvus vetorizará a consulta automaticamente.

# ========== Step 5: Content search, text-to-image / text-to-video / image-to-video ==========
def search_assets(query: str, top_k: int = 5, media_type: str | None = None,
                  min_score: float = 0.0):
    """query can be a Chinese or English text description, or the URL of a query image.
    min_score: the similarity threshold. Results below this value are considered irrelevant and discarded."""
    filter_expr = f'media_type == "{media_type}"' if media_type else ""
    results = client.search(
        collection_name=collection_name,
        data=[query],                 # Text or image URL, both vectorized by qwen3-vl-embedding
        anns_field="embedding",
        limit=top_k,
        filter=filter_expr,
        output_fields=["media_type", "media_ref", "caption", "tags"],
    )
    for rank, hit in enumerate(results[0], 1):
        if hit["distance"] < min_score:
            continue
        e = hit["entity"]
        url = e["media_ref"].split("?")[0]   # Remove the OSS signature parameters for cleaner display
        print(f"  {rank}. [Similarity {hit['distance']:.4f}] [{e['media_type']}] {e['caption'][:42]}")
        print(f"       {url}")

# Text-to-image search: restrict the search to images
print("[5.1] Text-to-image search:")
search_assets("A woman under an umbrella at the entrance of a convenience store on a rainy night", media_type="image")

# Text-to-video search: restrict the search to videos
print("[5.2] Text-to-video search:")
search_assets("A motion shot that slowly pushes in on a rainy street at night", media_type="video")

# Image-to-video search: use a keyframe image URL to recall video assets with a similar style
print("[5.3] Image-to-video search (recall videos from the same source with a keyframe image):")
search_assets(keyframe_url, media_type="video")

# Cross-modal hybrid search: remove the filter to search a mixed collection of images and videos
print("[5.4] Cross-modal hybrid search:")
search_assets("A warm-light shot of a convenience store on a rainy night")

A tabela a seguir mostra os resultados de recuperação e as pontuações de similaridade dos quatro métodos de busca em ambiente de teste.

A busca imagem para vídeo atinge 0,8715, a maior pontuação entre os quatro métodos, porque aquele vídeo foi gerado a partir deste keyframe: ativos da mesma source apresentam similaridade muito alta no mesmo espaço vetorial. Como imagens e vídeos compartilham um único espaço vetorial, uma única chamada de search cobre todos os quatro métodos: substitua o texto da consulta por uma URL de imagem para recuperar ativos a partir de uma consulta de mídia e remova o filter para pesquisar em uma coleção mista de imagens e vídeos.

Mantenha o texto da consulta relevante ao tema dos ativos e aplique um limiar de similaridade. A busca multimodal retorna os resultados Top-K ordenados por similaridade, entregando resultados mesmo quando a consulta não tem relação alguma com os ativos. A comparação a seguir foi medida em ambiente de teste.

A diferença é de quase sete vezes. Portanto, filtre os resultados no lado da aplicação com um limiar de similaridade (parâmetro min_score de search_assets neste exemplo) e descarte resultados claramente irrelevantes da cauda longa, evitando que pontuações como 0,07 sejam exibidas aos usuários como recuperações válidas. Ajuste o limiar exato com base na distribuição dos ativos do seu negócio.

Dois detalhes do exemplo afetam o que você vê e o que deve reutilizar:

• min_score tem como padrão 0,0 — Nenhuma das quatro chamadas de exemplo passa min_score, então nenhum resultado é descartado e ocorrências de baixa similaridade, como 0,07, aparecem na saída. Passe seu próprio limiar ao integrar a busca em uma aplicação. A filtragem ocorre após a seleção Top-K, e o exemplo pula linhas filtradas sem renumerar, portanto os números de classificação impressos podem ser não contíguos.

• A URL impressa está truncada — O exemplo remove os parâmetros de assinatura do OSS de media_ref para manter a saída impressa legível. Use o valor completo de media_ref em sua aplicação, não a forma truncada.

Solução de problemas

Limpeza

Ao concluir o tutorial, exclua a coleção criada na Etapa 4.

client.drop_collection(collection_name)

Benefícios da solução

A tabela a seguir compara este pipeline com a integração de cada modelo generativo do zero e a montagem de um sistema de recuperação próprio.

Todo item gerado é armazenado com sua legenda, suas tags e um vetor, transformando cada geração em um ativo pesquisável em vez de uma saída descartável. Quando um novo roteiro exigir uma cena semelhante, execute primeiro uma busca semântica na biblioteca histórica de ativos, reutilize os ativos recuperados e gere apenas o que falta.