Esta arquitectura de referencia describe cómo crear flujos de trabajo de IA duraderos en MongoDB Atlas con Temporal para la ingesta basada en eventos, la recuperación semántica y la ejecución basada en agentes.
Brinda soporte a equipos que necesitan generación aumentada por recuperación (RAG) y flujos de trabajo de IA de varios pasos para continuar de manera confiable a pesar de fallas, reintentos y operaciones de larga duración.
La arquitectura admite dos patrones de ingesta. En el patrón de flujo de eventos, las actualizaciones de origen fluyen a través de Kafka y Atlas Stream Processing antes de invocar Temporal. En el patrón directo, las actualizaciones de origen activan los flujos de trabajo de Temporal de inmediato. En ambos casos, MongoDB Atlas almacena datos operativos, conocimiento semántico y el estado de la aplicación, mientras que Temporal coordina las operaciones de extracción, segmentación, incrustación, indexación y recuperación.
Solución 1: Ingesta a través de Kafka y Atlas Stream Processing
Este patrón es adecuado para entornos que ya utilizan Kafka para el transporte de eventos, la propagación de cambios o la integración de sistemas desacoplados. Kafka proporciona una capa de entrada estándar para actualizaciones de alto volumen o de diversas fuentes, mientras que Atlas Stream Processing transforma y enruta los eventos antes de invocar a Temporal. Esta separación es importante cuando la ingesta debe escalar independientemente de la ejecución del flujo de trabajo, o cuando los equipos desean una estructura de eventos común para múltiples consumidores posteriores, además del pipeline de IA. Las actualizaciones de contenido se originan en diversas plataformas, como S3, API o una base de datos o sistema de mensajería.
Diagrama
El siguiente diagrama muestra este flujo.
Figura 1. Ingesta a través de Kafka y procesamiento de flujos de Atlas.
Flujo de datos
Los siguientes pasos describen este flujo:
Los sistemas de origen producen o exponen contenido.
El contenido proviene de sistemas de origen como Amazon S3, plataformas de IoT y bases de datos operativas. Estos sistemas proporcionan los documentos, registros o eventos sin procesar que procesa la canalización. Cuando hay contenido nuevo o actualizado disponible, el sistema de origen emite un evento al conector Kafka Sink, que lo escribe en la colección MongoDB
sources.Conector para fregadero Kafka
El conector Kafka Sink es el mecanismo de transferencia de mensajes persistente entre la capa de transmisión y MongoDB. Consume eventos de temas de Kafka y los escribe en MongoDB Atlas como documentos. Este componente ofrece dos ventajas que el disparador directo no proporciona: integra múltiples fuentes variadas en un único flujo ordenado y gestiona la contrapresión mediante búfer.
MongoDB (Atlas, Procesamiento de flujos y Búsqueda vectorial)
MongoDB desempeña un doble papel en esta arquitectura, apareciendo en dos puntos diferentes del flujo.
MongoDB Atlas es la zona de destino para los eventos del conector Kafka Sink. Los documentos fuente sin procesar que escribe el conector Sink se almacenan en Atlas como registro intermedio de lo que llega del exterior.
Atlas Stream Processing actúa como el desencadenante entre esa zona de aterrizaje y Temporal. Monitorea los documentos entrantes a través de flujos de cambios e inicia el flujo de trabajo posterior. Esto convierte a MongoDB de un almacén pasivo en una fuente de eventos activa, eliminando la necesidad de un bucle de sondeo y haciendo que la transferencia a Temporal se base en eventos en lugar de programarse.
Atlas Vector Search sirve como ruta de lectura para el agente, ejecutando la búsqueda de similitud semántica como una etapa nativa de la canalización de agregación, sin una base de datos vectorial separada ni una ramificación de consultas entre sistemas.
MongoDB interviene en ambos extremos del flujo de datos: recibe los eventos sin procesar y proporciona los vectores incrustados finales. Atlas Stream Processing conecta ambos procesos.
Integración de IA temporal y de viaje
Atlas Stream Processing invoca a Temporal, en lugar de que Temporal reciba el evento directamente. El núcleo de procesamiento es el mismo que en la Solución 2: los flujos de trabajo de Temporal proporcionan una orquestación duradera y reanudable, y la integración de Voyage AI se ejecuta como actividades reintentables de forma independiente dentro de esos flujos de trabajo.
En esta solución, Temporal consume eventos de MongoDB en lugar de recibirlos directamente. Solo recibe un disparador limpio y normalizado de Atlas Stream Processing, independientemente de si se originó en S3, IoT o una base de datos. Esta etapa es bidireccional: Temporal escribe los fragmentos indexados e integrados de nuevo en la colección de conocimiento de MongoDB una vez que finaliza el procesamiento.
Solución 2: Ingerir fuentes de datos directamente en Temporal
Este patrón es adecuado para sistemas que pueden emitir eventos de objetos, API o aplicaciones confiables directamente a un activador de flujo de trabajo. Reduce las capas arquitectónicas y simplifica la ruta de ingesta, a la vez que conserva la capacidad de reanudación para pasos de extracción e incrustación de larga duración. Dado que el flujo de trabajo se activa con una referencia de origen en lugar de contenido sin procesar, la canalización posterior permanece independiente del origen y puede admitir sistemas ascendentes adicionales con cambios mínimos en la orquestación.
Diagrama
El siguiente diagrama muestra este flujo.
Figura 2. Ingesta de fuentes de datos directamente en Temporal
Flujo de datos
Los siguientes pasos describen este flujo:
Los sistemas de origen producen o exponen contenido.
El contenido proviene de sistemas de origen como Amazon S3, plataformas de IoT y bases de datos operativas. Estos sistemas proporcionan los documentos, registros o eventos sin procesar que procesa la canalización. Cuando hay contenido nuevo o actualizado disponible, el sistema de origen emite un evento a un adaptador ligero, como AWS Lambda, un webhook o un conector. El adaptador inicia el flujo de trabajo temporal y solo pasa la referencia de origen y los metadatos necesarios. Esto mantiene la lógica específica del origen fuera del flujo de trabajo, lo que permite agregar nuevos tipos de origen sin modificar la canalización posterior.
Temporal gestiona el ciclo de vida de la ingestión.
Una vez iniciado el flujo de trabajo, Temporal coordina la ejecución, los reintentos y la recuperación a lo largo de las etapas de ingesta. Esto garantiza la durabilidad del proceso, de modo que las operaciones de larga duración continúen de forma fiable incluso en caso de fallos o reinicios. El flujo de trabajo recupera el contenido de origen y lo convierte a un formato normalizado para su posterior procesamiento mediante IA, creando una representación coherente entre los distintos tipos de fuente antes de que comience la transformación semántica.
Voyage AI genera incrustaciones dentro del flujo de trabajo.
La integración de Voyage AI se ejecuta como parte del flujo de trabajo, en lugar de como un paso externo que se ejecuta una sola vez. Esto permite que la integración sea observable y recuperable dentro de la misma ruta de ejecución, y mantiene la transformación semántica estrechamente vinculada al ciclo de vida de la ingesta.
MongoDB Atlas almacena contenido, metadatos e incrustaciones.
MongoDB Atlas almacena el contenido procesado, los metadatos relacionados y los vectores de incrustación en una única plataforma. Esto crea una capa de conocimiento duradera que admite tanto la ruta de escritura de ingesta como la ruta de recuperación posterior.
Atlas Vector Search hace que el conocimiento sea recuperable.
Atlas Vector Search indexa el contenido incrustado para que puedas consultarlo por similitud semántica. Dado que las incrustaciones y los metadatos operativos permanecen en la misma plataforma, las solicitudes posteriores de la aplicación y del agente recuperan el contexto relevante sin necesidad de un almacén de vectores o una capa de sincronización independientes.
Flujo de solicitud del agente
Este flujo se aplica a ambas soluciones de ingesta. Utiliza los mismos principios de durabilidad que la ingesta: en lugar de tratar la ejecución del agente como una solicitud API de corta duración, la arquitectura ejecuta la recuperación y el razonamiento como operaciones basadas en flujos de trabajo que se pueden observar, reintentar y reanudar. Esto es importante cuando el agente realiza múltiples llamadas de recuperación, invoca herramientas externas o necesita devolver actualizaciones de estado progresivas antes de obtener un resultado final.
Diagrama
El siguiente diagrama muestra los componentes del agente.
Figura 3. Arquitectura del agente de investigación
Flujo de datos
Los siguientes pasos describen este flujo:
Iniciar una solicitud de investigación de agente
Cuando la API del agente recibe una consulta del usuario a través de la interfaz de usuario, inicia un flujo de trabajo temporal persistente. El sistema devuelve inmediatamente un identificador de flujo de trabajo, de modo que la interfaz de usuario puede realizar un seguimiento del progreso mientras el agente de investigación recupera el contexto y las razones de una respuesta.
Recuperar contexto de MongoDB Atlas Vector Search
MongoDB Atlas aloja los metadatos, las incrustaciones vectoriales y los índices semánticos. Durante la interacción del usuario, el agente incrusta la consulta y utiliza MongoDB Atlas Vector Search para recuperar el contexto relevante, aplicando a menudo una capa de reordenamiento antes de la síntesis final. Para obtener información sobre los patrones de indexación y consulta, consulte la documentación de Atlas Vector Search.
Devuelva un resultado de investigación duradero.
El flujo de trabajo de investigación coordina la ejecución de herramientas, las interacciones con el modelo y la síntesis final. Temporal mantiene el proceso mediante reintentos y persistencia de estado, de modo que la interfaz de usuario proporciona una respuesta validada. Para obtener más información sobre la ejecución persistente, consulte la documentación de Temporal.
Componentes
Los siguientes componentes implementan esta arquitectura.
MongoDB Atlas
Utilice MongoDB Atlas para almacenar los fragmentos preparados e integrados del proceso de ingesta, el índice de búsqueda vectorial de MongoDB Atlas y el estado del agente en una única base de datos. Dado que el agente lee los mismos datos que escribe el proceso de ingesta, no es necesario mantener sincronizado un almacenamiento de memoria independiente.
Atlas Stream Processing
Utilice Atlas Stream Processing para proporcionar una ruta de integración opcional basada en eventos entre fuentes de streaming y flujos de trabajo temporales. Gestiona la transformación y el enrutamiento en tiempo real de los eventos entrantes para arquitecturas de alto rendimiento. Esto le permite implementar la ingesta casi en tiempo real sin necesidad de Kafka si opta por una conexión directa entre la fuente y los flujos de trabajo temporales.
Voyage AI
Utilice Voyage AI para generar las incrustaciones y realizar la reclasificación de resultados para esta arquitectura. Dado que la incrustación y la reclasificación son independientes de la orquestación del flujo de trabajo, los equipos pueden actualizar el modelo independientemente de las canalizaciones de ingesta y recuperación.
Atlas Vector Search
Utilice MongoDB Atlas Vector Search para recuperar el contexto del documento y la memoria del agente mediante indexación semántica. El agente de investigación consulta el mismo clúster de MongoDB Atlas que se utiliza para el almacenamiento principal, por lo que la recuperación utiliza los datos operativos actuales.
Temporal
Utilice Temporal como capa de ejecución robusta para la ingesta y los flujos de trabajo basados en agentes en sistemas de origen, MongoDB Atlas y servicios de IA externos. Coordina pasos de larga duración como la extracción, la fragmentación, la incrustación y la indexación, y proporciona reintentos, puntos de control y recuperación integrados. Esto permite que los flujos de trabajo se reanuden desde su último estado exitoso tras un fallo o interrupción, en lugar de reiniciarse. Al hacer que la orquestación sea robusta, Temporal ayuda a los equipos a operar, reponer y evolucionar de forma fiable las canalizaciones de IA en producción.
Excepciones, salvedades y compensaciones
Antes de adoptar esta arquitectura, considere las ventajas y desventajas de la ingesta directa y la basada en Kafka. Una conexión directa entre la fuente y el repositorio temporal requiere menos componentes. Una ruta basada en Kafka añade un intermediario de mensajes, lo que implica infraestructura adicional, pero se adapta perfectamente si su organización ya enruta las actualizaciones de la fuente a través de Kafka.
Implementación y más información
Consulte el repositorio de GitHub mdb-temporal-pra para obtener instrucciones de implementación y documentación técnica.