KubeMQ로 GenAI 애플리케이션 강화: RAG(검색 증강 생성)를 효율적으로 확장
- Linda Hamilton원래의
- 2024-12-26 09:00:15114검색
산업 전반에서 GenAI(제너레이티브 AI) 채택이 급증함에 따라 조직에서는 상황에 맞는 실시간 AI 모델을 강화하기 위해 검색 증강 세대(RAG) 기술을 점점 더 많이 활용하고 있습니다. 데이터. 이러한 애플리케이션에서 복잡한 정보 흐름을 관리하는 것은 특히 대규모로 지속적으로 생성되는 데이터를 처리할 때 중요한 과제를 안겨줍니다. 강력한 메시지 브로커인 KubeMQ는 여러 RAG 프로세스의 라우팅을 간소화하여 GenAI 애플리케이션에서 효율적인 데이터 처리를 보장하는 솔루션으로 등장합니다.
RAG 워크플로의 효율성과 확장성을 더욱 향상하려면 FalkorDB와 같은 고성능 데이터베이스를 통합하는 것이 필수적입니다. FalkorDB는 RAG 시스템이 의존하는 동적 지식 기반을 위한 안정적이고 확장 가능한 스토리지 솔루션을 제공하여 빠른 데이터 검색과 KubeMQ와 같은 메시징 시스템과의 원활한 통합을 보장합니다.
GenAI 워크플로의 RAG 이해
RAG는 검색 메커니즘을 통합하여 모델이 추론 중에 외부 지식 기반에 액세스할 수 있도록 하여 생성 AI 모델을 향상시키는 패러다임입니다. 이 접근 방식은 이용 가능한 가장 최근의 관련 정보를 바탕으로 생성된 응답의 정확성, 관련성 및 적시성을 크게 향상시킵니다.
RAG를 사용하는 일반적인 GenAI 워크플로에서 프로세스에는 여러 단계가 포함됩니다.
쿼리 처리: 사용자 입력을 해석하여 의도와 맥락을 이해
검색: FalkorDB와 같은 동적 지식 기반에서 관련 문서나 데이터를 가져오므로 가장 최근의 관련 정보에 빠르고 효율적으로 액세스할 수 있습니다.
세대: 입력 데이터와 검색된 데이터를 모두 사용하여 응답 생성
응답 전달: 최종적이고 풍부한 출력을 사용자에게 다시 제공
특히 데이터가 지속적으로 생성되고 업데이트되는 환경에서 이러한 단계를 확장하려면 RAG 파이프라인의 다양한 구성 요소 간의 데이터 흐름을 위한 효율적이고 안정적인 메커니즘이 필요합니다.
RAG 처리에서 KubeMQ의 중요한 역할
대규모 연속 데이터 스트림 처리
IoT 네트워크, 소셜 미디어 플랫폼 또는 실시간 분석 시스템과 같은 시나리오에서는 새로운 데이터가 끊임없이 생성되며 AI 모델은 이 정보를 통합하기 위해 신속하게 적응해야 합니다. 기존의 요청-응답 아키텍처는 처리량이 많은 조건에서 병목 현상이 발생하여 지연 시간 문제와 성능 저하로 이어질 수 있습니다.
KubeMQ는 서비스 간 효율적인 데이터 라우팅을 위한 확장 가능하고 강력한 인프라를 제공하여 처리량이 높은 메시징 시나리오를 관리합니다. KubeMQ를 RAG 파이프라인에 통합하면 각각의 새로운 데이터 포인트가 메시지 대기열이나 스트림에 게시되므로 검색 구성 요소가 시스템에 부담을 주지 않고 최신 정보에 즉시 액세스할 수 있습니다. 이 실시간 데이터 처리 기능은 GenAI 출력의 관련성과 정확성을 유지하는 데 매우 중요합니다.
최적의 라우터 역할
KubeMQ는 큐, 스트림, 게시-구독(pub/sub), 원격 프로시저 호출(RPC)을 포함한 다양한 메시징 패턴을 제공하여 RAG 파이프라인 내에서 다재다능하고 강력한 라우터가 됩니다. 짧은 대기 시간과 고성능 특성은 신속한 메시지 전달을 보장하며, 이는 지연이 사용자 경험과 시스템 효율성에 큰 영향을 미칠 수 있는 실시간 GenAI 애플리케이션에 필수적입니다.
게다가 복잡한 라우팅 로직을 처리하는 KubeMQ의 능력은 정교한 데이터 배포 전략을 가능하게 합니다. 이를 통해 AI 시스템의 다양한 구성 요소가 불필요한 중복이나 지연 없이 필요할 때 필요한 데이터를 정확하게 수신할 수 있습니다.
향상된 데이터 관리를 위한 FalkorDB 통합
KubeMQ는 서비스 간에 메시지를 효율적으로 라우팅하는 반면, FalkorDB는 RAG 프로세스에 필요한 방대한 양의 데이터를 저장하고 검색하기 위한 확장 가능한 고성능 그래프 데이터베이스 솔루션을 제공하여 이를 보완합니다. 이러한 통합을 통해 새로운 데이터가 KubeMQ를 통해 흐르면 FalkorDB에 원활하게 저장되므로 지연 시간이나 병목 현상 없이 검색 작업에 쉽게 사용할 수 있습니다.
확장성 및 신뢰성 향상
GenAI 애플리케이션의 사용자 기반과 데이터 양이 모두 증가함에 따라 확장성이 가장 중요한 관심사가 되었습니다. KubeMQ는 확장 가능하며 증가된 로드를 원활하게 수용할 수 있도록 수평 확장을 지원합니다. RAG 프로세스 수가 증가하거나 데이터 생성이 가속화됨에 따라 메시징 인프라가 강력하고 응답성이 뛰어난 상태로 유지됩니다.
또한 KubeMQ는 메시지 지속성과 내결함성을 제공합니다. 시스템 오류나 네트워크 중단이 발생하는 경우 KubeMQ는 메시지가 손실되지 않고 시스템이 정상적으로 복구될 수 있도록 보장합니다. 이러한 신뢰성은 사용자가 적시에 정확한 정보를 얻기 위해 의존하는 AI 애플리케이션의 무결성을 유지하는 데 매우 중요합니다.
전용 라우팅 서비스의 필요성 제거
RAG 파이프라인에서 데이터 처리를 위한 사용자 정의 라우팅 서비스를 구현하는 것은 리소스 집약적이고 복잡할 수 있습니다. 이러한 서비스를 구축, 유지 관리 및 확장하려면 상당한 개발 노력이 필요한 경우가 많아 핵심 AI 애플리케이션 개발에 집중하지 못하게 됩니다.
KubeMQ를 채택하면 조직에서는 맞춤형 라우팅 솔루션을 만들 필요가 없습니다. KubeMQ는 복잡한 라우팅 패턴, 메시지 필터링, 우선순위 처리 등 RAG 프로세스의 라우팅 요구 사항을 해결하는 즉시 사용 가능한 기능을 제공합니다. 이는 개발 및 유지 관리 오버헤드를 줄일 뿐만 아니라 GenAI 솔루션의 출시 기간을 가속화합니다.
REST 및 SDK를 통한 통합 액세스
KubeMQ는 메시지 브로커 기능과 상호작용하기 위한 여러 인터페이스를 제공합니다.
REST API: 언어에 구애받지 않는 통합을 가능하게 하여 모든 프로그래밍 언어로 작성된 서비스가 HTTP를 통해 메시지를 보내고 받을 수 있도록 합니다
SDK: 다양한 프로그래밍 언어(예: Python, Java, Go, .NET)용 클라이언트 라이브러리를 제공하여 기본 통합을 통해 보다 효율적인 통신 패턴과 향상된 성능을 촉진합니다
이러한 유연성을 통해 개발자는 특정 사용 사례에 가장 적합한 방법을 선택하여 아키텍처를 단순화하고 개발 주기를 가속화할 수 있습니다. 데이터 라우팅을 위한 단일 터치포인트는 RAG 파이프라인의 다양한 구성 요소 간의 통신을 간소화하여 전체 시스템 일관성을 향상시킵니다.
RAG 파이프라인에서 KubeMQ 구현: 자세한 예
코드 예제는 KubeMQ를 RAG 파이프라인에 통합하여 영화 정보 검색 시스템을 구축하는 방법을 보여줍니다. GPT-4를 사용하여 지식 그래프를 구축하기 위해 Rotten Tomatoes에서 영화 URL을 수집하는 서버를 설정합니다. 사용자는 채팅 클라이언트를 통해 이 시스템과 상호 작용하여 영화 관련 쿼리를 보내고 AI 생성 응답을 받을 수 있습니다. 이 사용 사례는 영화 맥락 내에서 효율적인 메시지 처리 및 서비스 간 통신을 위해 KubeMQ를 활용하여 실제 애플리케이션에서 지속적인 데이터 수집 및 실시간 쿼리 처리를 처리하는 방법을 보여줍니다.
아키텍처 개요
데이터 수집 서비스: 새로운 데이터가 나오면 캡처하여 KubeMQ 스트림에 게시합니다
검색 서비스: KubeMQ 스트림을 구독하여 업데이트를 받고 지식 기반을 새로 고치세요
생성 서비스: 쿼리 요청을 수신하고 AI 모델과 상호작용하며 응답을 생성합니다
응답 서비스: 생성된 응답을 적절한 채널을 통해 사용자에게 다시 보냅니다
KubeMQ 설정
Docker를 사용하여 배포하여 KubeMQ가 작동하는지 확인하세요.
docker run -d --rm \ -p 8080:8080 \ -p 50000:50000 \ -p 9090:9090 \ -e KUBEMQ_TOKEN="your token"
이 명령은 REST 및 gRPC 통신에 필요한 포트가 노출된 KubeMQ를 시작합니다.
RAG 서버 측
이 코드(GitHub repo)는 메시지 처리를 위해 KubeMQ를 사용하여 채팅 쿼리를 처리하고 지식 소스를 관리하는 RAG 서버를 구현합니다.
docker run -d --rm \ -p 8080:8080 \ -p 50000:50000 \ -p 9090:9090 \ -e KUBEMQ_TOKEN="your token"
서버는 두 개의 기본 스레드를 실행합니다. 하나는 'rag-chat-query'라는 채널을 통해 채팅 쿼리를 구독하고 GPT-4의 지식 그래프를 사용하여 이를 처리하는 스레드이고, 다른 하나는 'rag'라는 대기열에서 지속적으로 가져오는 스레드입니다. -sources-queue'를 사용하여 지식 그래프에 새로운 소스를 추가합니다. 지식 그래프는 JSON 파일에서 로드된 사용자 정의 온톨로지로 초기화되고 처리를 위해 OpenAI의 GPT-4 모델을 사용합니다. 서버는 정상적인 종료 처리 및 오류 관리를 구현하여 서버가 중지될 때 모든 스레드가 적절하게 종료되도록 합니다.
RAG 지식 그래프로 수집하기 위해 소스 데이터 보내기
# server.py
import json
import threading
from typing import List
from dotenv import load_dotenv
load_dotenv()
import time
from kubemq.common import CancellationToken
from kubemq.cq import Client as CQClient, QueryMessageReceived, QueryResponseMessage, QueriesSubscription
from kubemq.queues import Client as QueuesClient
from graphrag_sdk.models.openai import OpenAiGenerativeModel
from graphrag_sdk.model_config import KnowledgeGraphModelConfig
from graphrag_sdk import KnowledgeGraph, Ontology
from graphrag_sdk.source import URL
class RAGServer:
def __init__(self):
self.cq_client = CQClient(address="localhost:50000")
self.queues_client = QueuesClient(address="localhost:50000")
model = OpenAiGenerativeModel(model_name="gpt-4o")
with open("ontology.json", "r") as f:
ontology = json.load(f)
ontology = Ontology.from_json(ontology)
self.kg = KnowledgeGraph(
name="movies",
model_config=KnowledgeGraphModelConfig.with_model(model),
ontology=ontology)
self.chat = self.kg.chat_session()
self.shutdown_event = threading.Event()
self.threads: List[threading.Thread] = []
def handle_chat(self, request: QueryMessageReceived):
try:
message = request.body.decode('utf-8')
print(f"Received chat message: {message}")
result= self.chat.send_message(message)
answer = result.get("response","No answer")
print(f"Chat response: {answer}")
response = QueryResponseMessage(
query_received=request,
is_executed=True,
body=answer.encode('utf-8')
)
self.cq_client.send_response_message(response)
except Exception as e:
print(f"Error processing chat message: {str(e)}")
self.cq_client.send_response_message(QueryResponseMessage(
query_received=request,
is_executed=False,
error=str(e)
))
def pull_from_queue(self):
while not self.shutdown_event.is_set():
try:
result = self.queues_client.pull("rag-sources-queue", 10, 1)
if result.is_error:
print(f"Error pulling message from queue: {result.error}")
continue
sources = []
for message in result.messages:
source = message.body.decode('utf-8')
print(f"Received source: {source}, adding to knowledge graph")
sources.append(URL(message.body.decode('utf-8')))
if sources:
self.kg.process_sources(sources)
except Exception as e:
if not self.shutdown_event.is_set(): # Only log if not shutting down
print(f"Error processing sources: {str(e)}")
def subscribe_to_chat_queries(self):
def on_error(err: str):
if not self.shutdown_event.is_set(): # Only log if not shutting down
print(f"Error: {err}")
cancellation_token = CancellationToken()
try:
self.cq_client.subscribe_to_queries(
subscription=QueriesSubscription(
channel="rag-chat-query",
on_receive_query_callback=self.handle_chat,
on_error_callback=on_error,
),
cancel=cancellation_token
)
# Wait for shutdown signal
while not self.shutdown_event.is_set():
time.sleep(0.1)
# Cancel subscription when shutdown is requested
cancellation_token.cancel()
except Exception as e:
if not self.shutdown_event.is_set():
print(f"Error in subscription thread: {str(e)}")
def run(self):
chat_thread = threading.Thread(target=self.subscribe_to_chat_queries)
queue_thread = threading.Thread(target=self.pull_from_queue)
self.threads.extend([chat_thread, queue_thread])
for thread in self.threads:
thread.daemon = True # Make threads daemon so they exit when main thread exits
thread.start()
print("RAG server started")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\nShutting down gracefully...")
self.shutdown()
self.cq_client.close()
self.queues_client.close()
def shutdown(self):
print("Initiating shutdown sequence...")
self.shutdown_event.set() # Signal all threads to stop
for thread in self.threads:
thread.join(timeout=5.0) # Wait up to 5 seconds for each thread
if thread.is_alive():
print(f"Warning: Thread {thread.name} did not shutdown cleanly")
print("Shutdown complete")
if __name__ == "__main__":
rag_server = RAGServer()
rag_server.run()
이 코드는 KubeMQ의 대기열 시스템을 통해 영화 URL을 RAG 서버로 보내는 간단한 클라이언트를 구현합니다. 특히 KubeMQ에 연결하고 RAG 서버가 모니터링하는 것과 동일한 대기열인 'rag-sources-queue' 채널로 메시지를 보내는 SourceClient 클래스를 생성합니다. 메인 프로그램으로 실행하면 RAG 서버에서 처리하고 지식 그래프에 추가할 Rotten Tomatoes 영화 URL(Matrix 영화, John Wick 및 Speed 포함) 목록을 보냅니다.
질문과 답변 보내기 및 받기
# sources_client.py
from kubemq.queues import *
class SourceClient:
def __init__(self, address="localhost:50000"):
self.client = Client(address=address)
def send_source(self, message: str) :
send_result = self.client.send_queues_message(
QueueMessage(
channel="rag-sources-queue",
body=message.encode("utf-8"),
)
)
if send_result.is_error:
print(f"message send error, error:{send_result.error}")
if __name__ == "__main__":
client = SourceClient()
urls = ["https://www.rottentomatoes.com/m/side_by_side_2012",
"https://www.rottentomatoes.com/m/matrix",
"https://www.rottentomatoes.com/m/matrix_revolutions",
"https://www.rottentomatoes.com/m/matrix_reloaded",
"https://www.rottentomatoes.com/m/speed_1994",
"https://www.rottentomatoes.com/m/john_wick_chapter_4"]
for url in urls:
client.send_source(url)
print("done")
이 코드는 KubeMQ의 쿼리 시스템을 통해 RAG 서버와 통신하는 채팅 클라이언트를 구현합니다. ChatClient 클래스는 'rag-chat-query' 채널에 메시지를 보내고 응답을 기다립니다. 각 쿼리에 대한 제한 시간은 30초입니다. 메인 프로그램으로 실행하면 The Matrix의 감독과 Keanu Reeves와의 관계에 대한 두 가지 관련 질문을 보내고 수신되는 각 응답을 인쇄하여 클라이언트의 기능을 보여줍니다.
코드 저장소
모든 코드 예제는 원래 GitHub 저장소의 포크에서 찾을 수 있습니다.
결론
GenAI 애플리케이션용 RAG 파이프라인에 KubeMQ를 통합하면 지속적인 데이터 스트림과 복잡한 프로세스 간 통신을 처리하기 위한 확장 가능하고 안정적이며 효율적인 메커니즘이 제공됩니다. 다양한 메시징 패턴을 갖춘 통합 라우터 역할을 함으로써 KubeMQ는 전체 아키텍처를 단순화하고 맞춤형 라우팅 솔루션의 필요성을 줄이며 개발 주기를 가속화합니다.
또한 FalkorDB를 통합하면 KubeMQ와 원활하게 통합되는 고성능 지식 기반을 제공하여 데이터 관리가 향상됩니다. 이러한 조합은 최적화된 데이터 검색 및 저장을 보장하여 RAG 프로세스의 동적 요구 사항을 지원합니다.
지속성 및 내결함성과 같은 기능과 결합하여 높은 처리량 시나리오를 처리하는 능력은 과부하가 발생하거나 시스템 중단이 발생하는 경우에도 GenAI 애플리케이션이 응답성과 안정성을 유지하도록 보장합니다.
KubeMQ와 FalkorDB를 활용함으로써 조직은 데이터 라우팅 인프라가 강력하고 최신 AI 워크플로의 요구 사항을 충족할 수 있다는 확신을 갖고 AI 모델을 강화하고 귀중한 통찰력과 서비스를 제공하는 데 집중할 수 있습니다.
위 내용은 KubeMQ로 GenAI 애플리케이션 강화: RAG(검색 증강 생성)를 효율적으로 확장의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!