OpenAI 없이 RAG: BentoML, OctoAI 및 Milvus
이 글은 원래 The New Stack에 게시되었으며, 허가를 받아 여기에 재게시되었습니다.
확장된 검색 증강 생성 옵션은 개발자의 OpenAI 의존성을 없앨 수 있습니다
ChatGPT는 2023년에 AI를 대중적 지식의 최전선으로 끌어올렸습니다. 하지만 이제 훨씬 더 많은 옵션이 있으므로, 우리는 더 이상 OpenAI에 묶여 있지 않습니다. 이것은 OpenAI의 GPT가 아닌 LLM을 사용해 검색 증강 생성 (RAG) 앱을 구축하는 방법에 관한 블로그 시리즈의 세 번째 글입니다. 1부와 2부는 여기에서 확인할 수 있습니다. 이 프로젝트의 GitHub repo는 여기에서 찾을 수 있습니다.
이 튜토리얼에서는 BentoML을 사용해 embeddings를 제공하고, OctoAI를 사용해 LLM을 가져오며, Milvus를 vector database로 사용할 것입니다. 다루는 내용은 다음과 같습니다:
BentoML로 embeddings 제공하기
RAG를 위해 vector database에 데이터 삽입하기
Milvus collection 만들기
삽입을 위해 데이터 파싱 및 embedding하기
RAG를 위한 LLM 설정하기
LLM에 지시사항 제공하기
RAG 예시
RAG를 위한 BOM dot COM: BentoML, OctoAI, Milvus 요약
BentoML로 Embeddings 제공하기
BentoML의 Sentence Transformers Embeddings repository를 사용해 제공되는 sentence embeddings를 사용할 수 있습니다. 이 repo에서 어떤 일이 일어나는지 간단히 살펴보겠습니다. 주로 볼 것은 service.py 파일입니다. 기본적으로 서버를 띄우고 그 위에 API endpoint를 올립니다. API endpoint 내부에서는 Hugging Face의 all-MiniLM-L6-v2를 로드하고 이를 사용해 embeddings를 생성합니다.
해당 repository는 서버를 띄우고 <http://localhost:3000>에서 호출할 endpoint를 제공합니다. 이 endpoint를 사용하려면 bentoml을 import하고 SyncHTTPClient native object type을 사용해 HTTP client를 실행합니다.
import bentoml
bento_client = bentoml.SyncHTTPClient("http://localhost:3000")
client에 연결한 후, 문자열 목록에서 embeddings 목록을 가져오는 함수를 만듭니다. 한 가지 주목할 점은 목록을 한 번에 25개의 문자열로 나눈다는 것입니다. 이는 주로 synchronous endpoint를 사용하고 있기 때문입니다. 문자열 목록을 나누면 호출의 계산 비용이 줄어들고 timeout을 피할 수 있습니다.
목록을 25개 섹션으로 나눈 후, 위에서 만든 bento_client를 호출하여 이 문장들을 encode합니다. BentoML client는 vector 목록, 사실상 list of lists를 반환합니다. 우리는 이 vector 각각을 가져와 비어 있는 embedding 목록에 append합니다. 이 loop가 끝나면 최종 embeddings 목록을 반환합니다.
텍스트 목록에 문자열이 25개를 넘지 않으면, 전달된 문자열 목록에 대해 client의 encode method를 단순히 호출합니다.
def get_embeddings(texts: list) -> list:
if len(texts) > 25:
splits = [texts[x:x+25] for x in range(0, len(texts), 25)]
embeddings = []
for split in splits:
embedding_split = bento_client.encode(
sentences = split
)
for embedding in embedding_split:
embeddings.append(embedding)
return embeddings
return bento_client.encode(
sentences=texts,
)
RAG를 위해 Vector Database에 데이터 삽입하기
임베딩 함수가 준비되었으므로, RAG 애플리케이션을 위해 Milvus에 삽입할 데이터를 준비할 수 있습니다. 이 섹션의 첫 번째 단계는 Milvus를 시작하고 연결하는 것입니다. 위에 링크된 저장소에는 docker-compose.yml 파일이 있습니다. 이 문서 페이지에서도 Milvus Docker Compose를 찾을 수 있습니다.
Docker가 설치되어 있고 해당 repo를 다운로드했다면, docker compose up -d를 실행하여 Milvus를 시작할 수 있습니다. Milvus 서버가 올라오면, 이제 여기에 연결할 차례입니다. 이 부분에서는 단순히 connections 모듈을 import하고 호스트(localhost 또는 127.0.0.1)와 포트(19530)를 사용해 connect를 호출합니다. 아래 코드 블록은 두 개의 상수, 즉 컬렉션 이름과 차원도 정의합니다. 원하는 컬렉션 이름은 무엇이든 만들 수 있습니다. 차원 크기는 임베딩 모델 all-MiniLM-L6-v2의 크기에서 나옵니다.
from pymilvus import connections
COLLECTION_NAME = "bmo_test"
DIMENSION = 384
connections.connect(host="localhost", port=19530)
Milvus 컬렉션 생성하기
Milvus에서 컬렉션을 생성하는 과정은 두 단계로 이루어집니다. 먼저 스키마를 정의하고, 두 번째로 인덱스를 정의합니다. 이 섹션에는 네 개의 모듈이 필요합니다. FieldSchema는 필드를 정의하고, CollectionSchema는 컬렉션을 정의하며, DataType은 필드에 어떤 유형의 데이터가 들어갈지 알려주고, Collection은 Milvus가 컬렉션을 생성하는 데 사용하는 객체입니다.
여기서 컬렉션의 전체 스키마를 정의할 수 있습니다. 또는 필요한 두 가지 요소인 id와 embedding만 간단히 정의할 수도 있습니다. 그런 다음 스키마를 정의할 때 enabled_dynamic_field라는 매개변수를 전달하면, id와 embedding 필드만 함께 있다면 원하는 필드를 무엇이든 삽입할 수 있습니다. 이를 통해 Milvus에 데이터를 삽입하는 방식을 MongoDB 같은 NoSQL 데이터베이스를 다루는 방식과 동일하게 취급할 수 있습니다. 다음으로, 앞서 지정한 이름과 스키마로 컬렉션을 간단히 생성합니다.
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection
# id와 embedding은 정의해야 합니다
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
# "enable_dynamic_field"를 사용하면 어떤 메타데이터 필드로든 데이터를 삽입할 수 있습니다
schema = CollectionSchema(fields=fields, enable_dynamic_field=True)
# 컬렉션 이름을 정의하고 스키마를 전달합니다
collection = Collection(name=COLLECTION_NAME, schema=schema)
이제 컬렉션을 생성했으므로 인덱스를 정의해야 합니다. 검색 측면에서 “index”는 검색을 위해 데이터를 어떻게 매핑할 것인지를 정의합니다. 이 프로젝트에서는 데이터를 인덱싱하기 위해 HSNW(hierarchical navigable small worlds)를 사용합니다. 또한 벡터 거리를 측정하는 방식을 정의해야 합니다. 이 예제에서는 내적, 즉 IP를 사용합니다.
Milvus에서 제공하는 11가지 인덱스 유형은 각각 서로 다른 매개변수 집합을 가지고 있습니다. HNSW의 경우 조정할 매개변수는 “M”과 “efConstruction” 두 가지입니다. “M”은 각 그래프에서 노드 차수의 상한이며, “efConstruction”은 인덱스 생성 중 사용되는 탐색 계수입니다.
실용적인 관점에서 볼 때, 더 높은 “M”과 “efConstruction”은 둘 다 더 나은 검색으로 이어집니다. 더 높은 “M” 값은 인덱스가 더 많은 메모리를 차지한다는 의미입니다. 더 높은 “efConstruction” 값은 인덱스를 빌드하는 데 더 오래 걸린다는 의미입니다. 최적의 값을 찾으려면 이것저것 조정해 보아야 합니다.
인덱스가 정의되면, 선택한 필드(이 경우 embedding)에 인덱스를 생성합니다. 그런 다음 load를 호출하여 컬렉션을 메모리에 로드합니다.
index_params = {
"index_type": "HNSW", # 11개의 Milvus 인덱스 중 하나
"metric_type": "IP", # L2, Cosine 또는 IP
"params": {
"M": 8, # M이 높을수록 = 더 많은 메모리를 소비하지만 검색 품질이 더 좋음
"efConstruction": 64 # efConstruction이 높을수록 = 빌드는 느려지고, 검색은 더 좋아짐
},
}
# 인덱스를 만들 필드와 인덱스에 사용할 매개변수를 전달
collection.create_index(field_name="embedding", index_params=index_params)
# 컬렉션을 메모리에 로드
collection.load()
삽입을 위한 데이터 파싱 및 임베딩
Milvus가 준비되고 연결이 완료되면, 데이터를 벡터 데이터베이스에 삽입할 수 있습니다. 하지만 먼저 삽입할 데이터를 준비해야 합니다. 이 예제에서는 repo의 data 폴더에서 사용할 수 있는 여러 txt 파일이 있습니다. 이 데이터를 청크로 나누고, 임베딩한 뒤 Milvus에 저장합니다.
이 텍스트를 청크로 나누는 함수를 만드는 것부터 시작해 보겠습니다. 청킹을 수행하는 방법은 많지만, 이 예제에서는 단순한 방식으로 진행하겠습니다. 아래 함수는 파일을 받아 문자열로 읽은 다음 모든 새 줄에서 분할합니다. 새로 생성된 문자열 목록을 반환합니다.
# 단순하게 새 줄 기준으로 청킹
def chunk_text(filename: str) -> list:
with open(filename, "r") as f:
text = f.read()
sentences = text.split("n")
return sentences
다음으로, 우리가 가진 각 파일을 처리합니다. 모든 파일 이름의 목록을 가져오고 청크된 정보를 담을 빈 목록을 만듭니다. 그런 다음 모든 파일을 반복하며 위의 함수를 각각에 실행하여 각 파일을 단순하게 청킹합니다. 청크를 저장하기 전에 정리해야 합니다.
개별 파일이 어떻게 청킹되는지 보면 빈 줄이 많다는 것을 알 수 있으며, 우리는 빈 줄을 원하지 않습니다. 일부 줄은 탭이나 기타 특수 문자뿐입니다. 이를 피하기 위해 빈 목록을 만들고 특정 길이보다 긴 청크만 저장합니다. 단순화를 위해 일곱 글자를 사용할 수 있습니다.
각 문서에서 정리된 청크 목록을 얻으면 데이터를 저장할 수 있습니다. 각 청크 목록을 문서 이름, 이 경우 도시 이름에 매핑하는 딕셔너리를 만듭니다. 그런 다음 이 모든 것을 위에서 만든 빈 목록에 추가합니다.
import os
cities = os.listdir("data")
# 각 도시의 청크된 텍스트를 dict 목록에 저장
city_chunks = []
for city in cities:
chunked = chunk_text(f"data/{city}")
cleaned = []
for chunk in chunked:
if len(chunk) > 7:
cleaned.append(chunk)
mapped = {
"city_name": city.split(".")[0],
"chunks": cleaned
}
city_chunks.append(mapped)
각 도시별 청크된 텍스트 세트가 준비되었으므로 이제 임베딩을 가져올 차례입니다. Milvus는 컬렉션에 삽입할 딕셔너리 목록을 받을 수 있으므로, 또 다른 빈 목록으로 시작할 수 있습니다. 위에서 만든 각 딕셔너리에 대해 문장 목록과 일치하는 임베딩 목록을 가져와야 합니다.
우리는 섹션에서 BentoML을 사용해 만든 get_embeddings 함수를 청크 목록의 각 항목에 직접 호출하여 이를 수행합니다. 이제 이것들을 서로 매칭해야 합니다. 목록 임베딩과 문장 목록은 인덱스별로 일치해야 하므로, 어느 목록이든 enumerate를 사용해 순회하면서 매칭할 수 있습니다.
Milvus에 들어갈 단일 항목을 나타내는 딕셔너리를 만들어 이들을 매칭합니다. 각 항목에는 임베딩, 관련 문장, 그리고 도시가 포함됩니다. 도시를 포함하는 것은 선택 사항이지만, 사용할 수 있도록 포함해 봅시다. 이 항목에 id를 포함할 필요가 없다는 점에 주목하세요. 위에서 스키마를 만들 때 id를 자동 증가하도록 선택했기 때문입니다.
이들을 반복 처리하면서 각 항목을 목록에 추가합니다. 마지막에는 각 딕셔너리가 Milvus의 단일 행 항목을 나타내는 딕셔너리 목록이 생깁니다. 그런 다음 이 항목들을 Milvus 컬렉션에 간단히 삽입할 수 있습니다. 여기서 마지막 단계는 항목들을 인덱싱하기 시작할 수 있도록 flush하는 것입니다.
entries = []
for city_dict in city_chunks:
embedding_list = get_embeddings(city_dict["chunks"]) # returns a list of lists
# now match texts with embeddings and city name
for i, embedding in enumerate(embedding_list):
entry = {"embedding": embedding,
"sentence": city_dict["chunks"][i], # poorly named cuz it's really a bunch of sentences, but meh
"city": city_dict["city_name"]}
entries.append(entry)
collection.insert(entries)
collection.flush()
RAG를 위한 LLM 설정
이제 LLM을 가져와 본격적으로 시작할 준비를 해봅시다. 여기서 본격적으로 시작한다는 것은 RAG를 한다는 뜻입니다. 이 섹션을 그대로 따라 하려면 OctoAI 계정이 필요합니다. 대체로 원하는 어떤 LLM이든 선택해 사용할 수도 있습니다.
이 첫 번째 코드 블록에서는 환경 변수를 로드하고, OctoAI API 토큰을 추출한 다음, 클라이언트를 실행합니다.
from dotenv import load_dotenv
load_dotenv()
os.environ["OCTOAI_TOKEN"] = os.getenv("OCTOAI_API_TOKEN")
from octoai.client import Client
octo_client = Client()
LLM에 지시 사항 제공하기
LLM이 RAG를 수행하기 위해 알아야 할 것은 두 가지입니다: 질문과 컨텍스트입니다. 두 개의 문자열, 즉 질문과 컨텍스트를 받는 함수를 만들어 이 둘을 한 번에 전달할 수 있습니다. 이 함수를 사용해 OctoAI 클라이언트의 채팅 완성을 통해 LLM을 호출합니다. 이 예제에서는 Nous Research가 파인튜닝한 Mixtral 모델을 사용합니다.
이 모델에는 어떻게 동작해야 하는지를 나타내는 두 개의 “메시지”를 제공합니다. 먼저 LLM에게 주어진 컨텍스트만을 기반으로 사용자의 질문에 답변한다는 메시지를 전달합니다. 다음으로 사용자가 있을 것이라고 알려주고, 질문을 그대로 전달합니다.
다른 매개변수들은 모델 동작을 튜닝하기 위한 것입니다. 최대 토큰 수와 모델이 얼마나 “창의적으로” 행동할지를 제어할 수 있습니다.
그런 다음 함수는 클라이언트의 출력을 JSON 형식으로 반환합니다.
def dorag(question: str, context: str):
completion = octo_client.chat.completions.create(
messages=[
{
"role": "system",
"content": f"You are a helpful assistant. The user has a question. Answer the user question based only on the context: {context}"
},
{
"role": "user",
"content": f"{question}"
}
],
model="nous-hermes-2-mixtral-8x7b-dpo",
max_tokens=512,
presence_penalty=0,
temperature=0.1,
top_p=0.9,
)
return completion.model_dump()
RAG 예제
이제 준비가 되었습니다. 질문을 할 시간입니다. 아마 함수를 만들지 않고도 할 수 있겠지만, 함수를 만들면 깔끔하고 반복하기 좋습니다. 이 함수는 단순히 질문을 입력받은 다음 RAG를 수행해 답변합니다.
먼저 문서를 임베딩할 때 사용한 것과 동일한 임베딩 모델을 사용해 질문을 임베딩합니다. 다음으로 Milvus에서 검색을 실행합니다. 질문을 리스트 형식으로 get_embeddings 함수에 전달한 다음, 출력된 리스트를 Milvus 검색의 data 섹션에 바로 전달한다는 점에 주목하세요. 이는 단지 함수 시그니처가 그렇게 설정되어 있기 때문이며, 여러 함수를 다시 작성하는 것보다 재사용하는 편이 더 쉽기 때문입니다.
검색 호출 안에서는 몇 가지 매개변수를 추가로 제공해야 합니다. anns_field는 Milvus에 어떤 필드에서 근사 최근접 이웃 검색 (ANNS)을 수행할지 알려줍니다. 또한 인덱스에 대한 몇 가지 매개변수도 전달해야 합니다. metric type이 인덱스를 만들 때 사용한 것과 일치하는지 확인하세요. 이 경우에는 IP입니다. 또한 일치하는 인덱스 매개변수도 사용해야 하는데, 이 경우에는 exploratory factor인 ef입니다.
ef가 높을수록 검색 시간은 길어지지만 recall은 높아집니다. 이를 실험해 볼 수 있으며, ef는 2048까지 올라갈 수 있습니다. 여기서는 속도와 단순성을 위해 16을 사용합니다. 이 데이터셋에는 수천 개의 항목만 있습니다. 다음으로 limit 매개변수도 전달하는데, 이는 Milvus에서 몇 개의 결과를 가져올지 알려줍니다. 이 예제에서는 다섯 개로 하면 됩니다.
마지막 검색 매개변수는 검색에서 어떤 필드를 반환받을지 정의합니다. 이 예제에서는 텍스트 청크를 저장하는 데 사용한 필드인 sentence만 가져오면 됩니다. 검색 결과를 받은 뒤에는 이를 처리해야 합니다. Milvus는 내부에 hits가 있는 엔티티를 반환하므로, 다섯 개의 hit 모두에서 “sentence”를 가져와 마침표로 연결해 리스트 형태의 문단이 되도록 합니다.
그런 다음 사용자가 질문한 내용과 그 문단을 위에서 만든 dorag 함수에 전달하고 응답만 반환합니다.
def ask_a_question(question):
embeddings = get_embeddings([question])
res = collection.search(
data=embeddings, # search for the one (1) embedding returned as a list of lists
anns_field="embedding", # Search across embeddings
param={"metric_type": "IP",
"params": {"ef": 16}},
limit = 5, # get me the top 5 results
output_fields=["sentence"] # get the sentence/chunk and city
)
sentences = []
for hits in res:
for hit in hits:
sentences.append(hit.entity.get("sentence"))
context = ". ".join(sentences)
return dorag(question, context)
print(ask_a_question("What state is Cambridge in?")["choices"][0]["message"]["content"])
Cambridge가 어느 주에 있는지 묻는 예제 질문의 경우, OctoAI의 전체 응답을 그냥 출력하면 됩니다. 하지만 시간을 들여 이를 파싱하면 더 보기 좋고, Cambridge가 Massachusetts에 위치해 있다고 알려줄 것입니다.
BOM dot COM 요약: RAG를 위한 BentoML, OctoAI, Milvus
이 예제에서는 OpenAI나 프레임워크 없이 RAG를 수행하는 방법을 다뤘습니다. 이전의 일부 예제와 달리 이번에는 LangChain이나 LlamaIndex도 사용하지 않았다는 점에 주목하세요. 이번 스택은 BOM.COM — BentoML, OctoAI, Milvus였습니다. BentoML의 서빙 기능을 사용해 임베딩 모델 엔드포인트를 제공했고, OctoAI의 LLM 엔드포인트를 사용해 오픈 소스 모델에 접근했으며, Milvus를 벡터 데이터베이스로 사용했습니다.
이러한 다양한 퍼즐 조각들을 사용하는 순서를 구조화하는 방법은 많습니다. 이 예제에서는 먼저 BentoML로 로컬 서버를 띄워 Hugging Face의 임베딩 모델을 호스팅했습니다. 다음으로, Docker Compose를 사용해 로컬 Milvus 인스턴스를 띄웠습니다.
우리는 Wikipedia에서 스크래핑한 데이터를 청크로 나누기 위해 간단한 방법을 사용했습니다. 그런 다음, 그 청크들을 BentoML에서 호스팅되는 임베딩 모델에 전달하여 Milvus에 넣을 벡터 임베딩을 얻었습니다. 모든 벡터 임베딩이 Milvus에 들어간 상태에서, RAG를 수행할 준비가 완전히 되었습니다.
이번에 선택한 LLM은 OctoAI에서 사용할 수 있는 많은 오픈 소스 모델 중 하나인 Nous Hermes 미세 조정 Mixtral 모델이었습니다. 우리는 RAG를 가능하게 하기 위해 두 개의 함수를 만들었습니다. 하나는 질문과 컨텍스트를 LLM에 전달하는 함수인 dorag였고, 다른 하나는 사용자 질문을 임베딩하고 Milvus를 검색한 다음 검색 결과를 질문과 함께 원래의 RAG 함수에 전달하는 함수였습니다. 마지막에는 정상 동작 확인을 위해 간단한 질문으로 RAG를 테스트했습니다.
계속 읽기
VDBBench Adds Cost-Aware Benchmarking for Vector Databases
Compare Zilliz Cloud, Pinecone, and turbopuffer with VDBBench cost-aware vector database benchmarks across latency, freshness, multitenancy, and cold starts.
Smarter Autoscaling in Zilliz Cloud: Always Optimized for Every Workload
With the latest upgrade, Zilliz Cloud introduces smarter autoscaling—a fully automated, more streamlined, elastic resource management system.
Zilliz Cloud Enterprise Vector Search Powers High-Performance AI on AWS
Zilliz Cloud on AWS powers secure, scalable, ultra-fast vector search for enterprise AI apps, with BYOC, sub-10ms latency, and zero-DevOps simplicity.