Integrações
Dynamiq and Zilliz Cloud Integration

Dynamiq and Zilliz Cloud Integration

Dynamiq and Zilliz Cloud integrate to power fast, secure, and context-aware AI agents, combining Dynamiq's enterprise agentic AI platform with visual workflow builder and Python SDK alongside Zilliz Cloud's high-performance vector storage for scalable RAG retrieval.

Use esta integração gratuitamente

O que é Dynamiq
Dynamiq é a plataforma operacional empresarial para IA agêntica. Ela permite que equipes construam, coordenem e implementem agentes de IA e fluxos de trabalho RAG de nível de produção sem integrar várias ferramentas separadas. Ela oferece tanto um construtor visual de arrastar e soltar quanto um SDK Python open-source, dando a desenvolvedores e usuários de negócios a flexibilidade para escalar aplicações agênticas em SaaS, nuvem privada ou ambientes totalmente air-gapped.

Ao integrar-se com a Zilliz Cloud (Milvus totalmente gerenciado), a Dynamiq obtém acesso a armazenamento e recuperação vetorial de alto desempenho que impulsionam seus fluxos de trabalho RAG agênticos, combinando a infraestrutura vetorial escalável da Zilliz com a camada de orquestração da Dynamiq para permitir agentes responsivos e cientes do contexto que operam em bases de conhecimento empresariais.
Benefícios da integração Dynamiq + Zilliz Cloud
- Fluxos de trabalho RAG agênticos escaláveis: A camada de orquestração da Dynamiq, combinada com o armazenamento vetorial de alto desempenho do Zilliz Cloud, permite agentes de IA de nível de produção capazes de recuperar contexto de bases de conhecimento empresariais em grande escala em tempo real.
- Flexibilidade com dois construtores: As equipes podem criar fluxos de trabalho RAG usando o construtor visual de arrastar e soltar da Dynamiq ou o SDK Python open-source, com o Zilliz Cloud gerenciando o armazenamento vetorial de forma transparente em qualquer uma das abordagens.
- Implantação empresarial segura: A integração oferece suporte a ambientes SaaS, nuvem privada e totalmente isolados da internet, com observabilidade completa e guardrails em torno das chamadas a LLMs e do uso de vetores.
- Indexação e recuperação eficientes de documentos: O Writer Node da Dynamiq ingere e indexa documentos no Zilliz Cloud, enquanto o Retriever Node conecta agentes a bases de conhecimento com tecnologia Milvus para recuperação de contexto em tempo real.
Como a integração funciona
Dynamiq atua como a plataforma de orquestração, fornecendo o framework de IA agêntica com construtor visual de fluxos de trabalho, Python SDK e recursos de agente LLM. Ele lida com processamento de documentos, orquestração de fluxos de trabalho, gerenciamento de prompts e geração de respostas por meio de sua arquitetura modular baseada em nós.

Zilliz Cloud atua como a camada de banco de dados vetorial, armazenando e indexando embeddings de documentos por meio do nó MilvusDocumentWriter da Dynamiq e fornecendo busca rápida por similaridade por meio do nó MilvusDocumentRetriever. Ele permite a recuperação eficiente de contexto relevante a partir de grandes bases de conhecimento.

Juntos, Dynamiq e Zilliz Cloud criam uma solução RAG agêntica de ponta a ponta: os documentos são processados, incorporados e armazenados no Zilliz Cloud por meio do fluxo de trabalho de indexação da Dynamiq. Quando os usuários interagem com agentes de IA, o fluxo de trabalho de recuperação da Dynamiq consulta o Zilliz Cloud em busca de embeddings de documentos relevantes e passa o contexto para o LLM gerar respostas detalhadas e sensíveis ao contexto — tudo sem código personalizado ou configuração de infraestrutura.

Guia passo a passo

1. Instale as bibliotecas necessárias

$ pip install dynamiq pymilvus

2. Configure o agente LLM

Usaremos OpenAI como o LLM neste exemplo. Prepare a chave de API como uma variável de ambiente:

import os

os.environ["OPENAI_API_KEY"] = "sk-***********"

3. Fluxo de indexação de documentos — Importe bibliotecas e inicialize o fluxo de trabalho

from io import BytesIO
from dynamiq import Workflow
from dynamiq.nodes import InputTransformer
from dynamiq.connections import (
    OpenAI as OpenAIConnection,
    Milvus as MilvusConnection,
    MilvusDeploymentType,
)
from dynamiq.nodes.converters import PyPDFConverter
from dynamiq.nodes.splitters.document import DocumentSplitter
from dynamiq.nodes.embedders import OpenAIDocumentEmbedder
from dynamiq.nodes.writers import MilvusDocumentWriter

# Initialize the workflow
rag_wf = Workflow()

4. Defina os nós do pipeline de indexação

Defina os nós de conversor de PDF, divisor de documentos, incorporação e armazenamento vetorial Milvus:

# PDF Converter
converter = PyPDFConverter(document_creation_mode="one-doc-per-page")
converter_added = rag_wf.flow.add_nodes(converter)

# Document Splitter
document_splitter = DocumentSplitter(
    split_by="sentence",
    split_length=10,
    split_overlap=1,
    input_transformer=InputTransformer(
        selector={
            "documents": f"${[converter.id]}.output.documents",
        },
    ),
).depends_on(converter)
splitter_added = rag_wf.flow.add_nodes(document_splitter)

# Embedding Node
embedder = OpenAIDocumentEmbedder(
    connection=OpenAIConnection(api_key=os.environ["OPENAI_API_KEY"]),
    input_transformer=InputTransformer(
        selector={
            "documents": f"${[document_splitter.id]}.output.documents",
        },
    ),
).depends_on(document_splitter)
document_embedder_added = rag_wf.flow.add_nodes(embedder)

# Milvus Vector Store
vector_store = (
    MilvusDocumentWriter(
        connection=MilvusConnection(
            deployment_type=MilvusDeploymentType.FILE, uri="./milvus.db"
        ),
        index_name="my_milvus_collection",
        dimension=1536,
        create_if_not_exist=True,
        metric_type="COSINE",
    )
    .inputs(documents=embedder.outputs.documents)
    .depends_on(embedder)
)
milvus_writer_added = rag_wf.flow.add_nodes(vector_store)

Milvus oferece dois tipos de implantação: MilvusDeploymentType.FILE é ideal para prototipagem local ou armazenamento de dados em pequena escala, definindo o uri como um caminho de arquivo local (por exemplo, ./milvus.db) para aproveitar o Milvus Lite. MilvusDeploymentType.HOST foi projetado para cenários de dados em larga escala. Você pode implantar um servidor Milvus usando Docker ou Kubernetes, ou usar Zilliz Cloud definindo o uri e o token como o Public Endpoint e a API Key no Zilliz Cloud.

5. Execute o fluxo de trabalho de indexação

file_paths = ["./pdf_files/WhatisMilvus.pdf"]
input_data = {
    "files": [BytesIO(open(path, "rb").read()) for path in file_paths],
    "metadata": [{"filename": path} for path in file_paths],
}

inserted_data = rag_wf.run(input_data=input_data)

6. Fluxo de recuperação de documentos — Inicialize e defina nós

from dynamiq import Workflow
from dynamiq.connections import (
    OpenAI as OpenAIConnection,
    Milvus as MilvusConnection,
    MilvusDeploymentType,
)
from dynamiq.nodes.embedders import OpenAITextEmbedder
from dynamiq.nodes.retrievers import MilvusDocumentRetriever
from dynamiq.nodes.llms import OpenAI
from dynamiq.prompts import Message, Prompt

# Initialize the workflow
retrieval_wf = Workflow()

# OpenAI connection and text embedder
openai_connection = OpenAIConnection(api_key=os.environ["OPENAI_API_KEY"])
embedder = OpenAITextEmbedder(
    connection=openai_connection,
    model="text-embedding-3-small",
)
embedder_added = retrieval_wf.flow.add_nodes(embedder)

# Recuperador de documentos Milvus
document_retriever = (
    MilvusDocumentRetriever(
        connection=MilvusConnection(
            deployment_type=MilvusDeploymentType.FILE, uri="./milvus.db"
        ),
        index_name="my_milvus_collection",
        dimension=1536,
        top_k=5,
    )
    .inputs(embedding=embedder.outputs.embedding)
    .depends_on(embedder)
)
milvus_retriever_added = retrieval_wf.flow.add_nodes(document_retriever)

7. Defina o Prompt e o Gerador de Respostas

prompt_template = """
Please answer the question based on the provided context.

Question: {{ query }}

Context:
{% for document in documents %}
- {{ document.content }}
{% endfor %}
"""

prompt = Prompt(messages=[Message(content=prompt_template, role="user")])

answer_generator = (
    OpenAI(
        connection=openai_connection,
        model="gpt-4o",
        prompt=prompt,
    )
    .inputs(
        documents=document_retriever.outputs.documents,
        query=embedder.outputs.query,
    )
    .depends_on([document_retriever, embedder])
)
answer_generator_added = retrieval_wf.flow.add_nodes(answer_generator)

8. Execute o Fluxo de Trabalho de Recuperação

sample_query = "What is the Advanced Search Algorithms in Milvus?"

result = retrieval_wf.run(input_data={"query": sample_query})

answer = result.output.get(answer_generator.id).get("output", {}).get("content")
print(answer)

Saiba Mais
- Primeiros passos com Dynamiq e Milvus — Tutorial oficial da Milvus para criar RAG com Dynamiq
- Documentação do Dynamiq — Documentação oficial do Dynamiq
- Repositório GitHub do Dynamiq — SDK de código aberto do Dynamiq
- Documentação do Zilliz Cloud — Documentação do Zilliz Cloud para Milvus gerenciado
- Crie apps de IA com Milvus: Tutoriais e Notebooks — Coleção da Zilliz de tutoriais e notebooks sobre Milvus

Dynamiq and Zilliz Cloud Integration

O que é Dynamiq

Benefícios da integração Dynamiq + Zilliz Cloud

Como a integração funciona

Guia passo a passo

Saiba Mais

Recursos Relacionados

Como avaliar aplicações RAG

O que é Geração Aumentada por Recuperação (RAG)?

Otimizando aplicações RAG