Integraciones
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.

Usa esta integración gratis

Qué es Dynamiq
Dynamiq es la plataforma operativa empresarial para IA agéntica. Permite a los equipos construir, coordinar e implementar agentes de IA de nivel de producción y flujos de trabajo RAG sin integrar múltiples herramientas separadas. Ofrece tanto un constructor visual de arrastrar y soltar como un SDK de Python de código abierto, brindando a desarrolladores y usuarios empresariales la flexibilidad para escalar aplicaciones agénticas en entornos SaaS, de nube privada o completamente aislados.

Al integrarse con Zilliz Cloud (Milvus totalmente gestionado), Dynamiq obtiene acceso a almacenamiento y recuperación vectorial de alto rendimiento que impulsan sus flujos de trabajo RAG agénticos, combinando la infraestructura vectorial escalable de Zilliz con la capa de orquestación de Dynamiq para habilitar agentes receptivos y conscientes del contexto que operan en bases de conocimiento empresariales.
Beneficios de la integración de Dynamiq + Zilliz Cloud
- Flujos de trabajo RAG agénticos escalables: La capa de orquestación de Dynamiq combinada con el almacenamiento vectorial de alto rendimiento de Zilliz Cloud permite agentes de IA de nivel de producción que pueden recuperar contexto de bases de conocimiento empresariales a gran escala en tiempo real.
- Flexibilidad de doble constructor: Los equipos pueden crear flujos de trabajo RAG usando el constructor visual de arrastrar y soltar de Dynamiq o el SDK de Python de código abierto, con Zilliz Cloud gestionando el almacenamiento vectorial sin problemas en cualquiera de los dos enfoques.
- Despliegue empresarial seguro: La integración admite SaaS, nube privada y entornos completamente aislados de Internet, con observabilidad completa y medidas de protección en torno a las llamadas a LLM y el uso vectorial.
- Indexación y recuperación eficiente de documentos: Writer Node de Dynamiq ingiere e indexa documentos en Zilliz Cloud, mientras que Retriever Node conecta agentes a bases de conocimiento impulsadas por Milvus para la recuperación de contexto en tiempo real.
Cómo funciona la integración
Dynamiq funciona como la plataforma de orquestación, proporcionando el marco de IA agéntica con un generador visual de flujos de trabajo, SDK de Python y capacidades de agentes LLM. Gestiona el procesamiento de documentos, la orquestación de flujos de trabajo, la gestión de prompts y la generación de respuestas mediante su arquitectura modular basada en nodos.

Zilliz Cloud funciona como la capa de base de datos vectorial, almacenando e indexando embeddings de documentos mediante el nodo MilvusDocumentWriter de Dynamiq y proporcionando una búsqueda rápida por similitud mediante el nodo MilvusDocumentRetriever. Permite la recuperación eficiente de contexto relevante desde grandes bases de conocimiento.

Juntos, Dynamiq y Zilliz Cloud crean una solución RAG agéntica de extremo a extremo: los documentos se procesan, se convierten en embeddings y se almacenan en Zilliz Cloud mediante el flujo de trabajo de indexación de Dynamiq. Cuando los usuarios interactúan con agentes de IA, el flujo de trabajo de recuperación de Dynamiq consulta Zilliz Cloud en busca de embeddings de documentos relevantes y pasa el contexto al LLM para generar respuestas detalladas y conscientes del contexto, todo sin código personalizado ni configuración de infraestructura.

Guía paso a paso

1. Instalar las bibliotecas requeridas

$ pip install dynamiq pymilvus

2. Configurar el agente LLM

Usaremos OpenAI como el LLM en este ejemplo. Prepara la clave de API como una variable de entorno:

import os

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

3. Flujo de indexación de documentos — Importar bibliotecas e inicializar el flujo de trabajo

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

# Inicializar el flujo de trabajo
rag_wf = Workflow()

4. Definir los nodos de la canalización de indexación

Define los nodos del convertidor PDF, divisor de documentos, incrustación y almacén vectorial Milvus:

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

# Divisor de documentos
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)

# Nodo de incrustación
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)

# Almacén vectorial Milvus
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 ofrece dos tipos de implementación: MilvusDeploymentType.FILE es ideal para prototipos locales o almacenamiento de datos a pequeña escala, configurando el uri como una ruta de archivo local (p. ej., ./milvus.db) para aprovechar Milvus Lite. MilvusDeploymentType.HOST está diseñado para escenarios de datos a gran escala. Puedes desplegar un servidor Milvus usando Docker o Kubernetes, o usar Zilliz Cloud configurando el uri y el token como el Public Endpoint y la API Key en Zilliz Cloud.

5. Ejecutar el flujo de trabajo de indexación

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. Flujo de recuperación de documentos — Inicializar y definir nodos

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

# Inicializar el flujo de trabajo
retrieval_wf = Workflow()

# Conexión de OpenAI e incrustador de texto
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 de 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. Define la plantilla de prompt y el generador de respuestas

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. Ejecuta el flujo de trabajo de recuperación

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)

Más información
- Primeros pasos con Dynamiq y Milvus — Tutorial oficial de Milvus para crear RAG con Dynamiq
- Documentación de Dynamiq — Documentación oficial de Dynamiq
- Repositorio de GitHub de Dynamiq — SDK de código abierto de Dynamiq
- Documentación de Zilliz Cloud — Documentación de Zilliz Cloud para Milvus gestionado
- Crea aplicaciones de IA con Milvus: tutoriales y notebooks — Colección de Zilliz de tutoriales y notebooks de Milvus

Dynamiq and Zilliz Cloud Integration

Qué es Dynamiq

Beneficios de la integración de Dynamiq + Zilliz Cloud

Cómo funciona la integración

Guía paso a paso

Más información

Recursos Relacionados

Cómo evaluar aplicaciones RAG

¿Qué es la generación aumentada por recuperación (RAG)?

Optimización de aplicaciones RAG