Amazon Bedrock으로 그래프 RAG 구현하기

개요

대규모 언어 모델들은 방대한 데이터를 기반으로 광범위한 지식과 우수한 문장 생성 능력을 갖추고 있습니다. 그러나 이러한 모델들은 학습 시점 이후의 최신 정보나 특정 주제에 대한 심층 지식을 반영하는 데 한계가 있으며, 때때로 환각(hallucination) 문제로 답변의 정확성을 떨어뜨리기도 합니다. 이러한 문제를 해결하기 위해, RAG(Retrieval Augmented Generation) 프레임워크가 등장했습니다. RAG는 필요한 정보를 자체 데이터베이스에 저장하고 검색해, 답변 생성의 사전 정보로 제공함으로써 학습 시점 이후의 데이터 및 프라이빗 정보를 답변에 반영할 수 있도록 합니다.

일반적인 RAG 프레임워크에서는 벡터 저장소를 데이터베이스로 활용해, 관련 정보를 저장하고 검색합니다. 이 접근법은 비구조화 된 정보를 문맥적 유사도에 근거하여 신속하게 찾아내기에 효과적입니다. 그러나, 하나의 정보를 여러 개의 chunk로 나누어 저장하면서 하나의 문맥적 의미가 여러 chunk에 분리되되거나, 서로 다른 chunk에 포함된 개념 간의 관계를 효과적으로 반영하지 못한다는 한계가 있습니다.

그래프 RAG는 핵심 개념 간의 복잡한 관계를 유연하게 정의하고, 데이터 간 연결 관계를 정보 탐색에 함께 반영함으로써 위 한계에 대응할 수 있습니다. 이 글에서는 그래프 RAG의 개념을 설명하고, 실습을 통해 Amazon Bedrock 기반 그래프 RAG 구현 방법을 소개합니다. 이 과정에서 ‘벡터 검색(vector search)’와 ‘그래프 순회(graph traversal)’ 두 가지 접근법을 모두 활용하면서, 그래프 RAG의 특징에 대하여 알아봅니다.

사전 준비

여기부터는 그래프 RAG에 대한 소개와 실습을 병행합니다. 실습은 아래 아키텍처를 활용하며, 실습을 모두 수행하기까지 1시간 이내로 소요됩니다.

1) 그래프 데이터베이스 구성

이 섹션에서는 Neo4j 그래프 데이터베이스를 실습 환경으로 구성합니다. 실습 환경 구성에는 CloudFormation을 사용합니다.

위 버튼을 클릭하여, EC2 인스턴스에 Neo4j 데이터베이스 및 관련 라이브러리를 자동 배포합니다 (약 10분 소요). 웹 브라우저로 데이터베이스에 접근하기 위해, 퍼블릭 서브넷에 EC2 인스턴스를 기본 구성합니다. 배포 이후, 필요에 따라 네트워크 및 방화벽 구성을 변경하여 프라이빗 인스턴스로 설정할 수 있습니다. 이 실습에서는 EC2 인스턴스에 Neo4j 커뮤니티 버전을 설치해 사용합니다. Neo4j는 AWS 파트너로 등록되어 있어, 마켓 플레이스에 다양한 엔터프라이즈 배포 옵션을 제공하고 있습니다.

이밖에, Amazon Bedrock 및 Neo4j 데이터베이스의 API 호출 목적으로 Amazon SageMaker의 Jupyter Lab 환경을 사용합니다. 위에 제공된 CloudFormation 템플릿에는 SageMaker 구성이 포함되지 않습니다. 만약, 실습에 사용할 SageMaker 환경을 새롭게 구축해야 한다면, 다음 링크를 참고해주세요

스택 생성이 완료되면 CloudFormation의 [Outputs] 탭에서 Neo4j가 설치된 EC2 인스턴스의 퍼블릭 IP 주소를 확인할 수 있습니다.

퍼블릭 IP 주소를 사용해 웹 브라우저에서 {IP 주소}:7474로 접속하면, Neo4j 사용자 인터페이스가 나타납니다. 초기 로그인 정보는 Username: neo4j / Password: neo4j 입니다.

로그인 후, 기본 비밀번호를 사용자 지정 암호로 변경해야 합니다. 비밀번호 변경 후 성공적으로 로그인하면 Neo4j를 실습에 사용할 준비가 완료된 것입니다.

2) 그래프 RAG의 데이터 준비

아래 그림은 그래프 RAG의 준비 및 활용 영역을 도식화 한 것입니다. 그래프 RAG에서 준비(아래 그림의 “Ingesting Data into KG”) 영역은 구조화 되지 않은 정보를 ‘엔티티(entity)’와 이들 간의 ‘연결 관계(relation)’로 변환하고, 그래프에 로드하는 과정을 포함합니다. 최근에는 Amazon Bedrock이 지원하는 대규모 언어모델과 프롬프트 엔지니어링 기법을 활용하여, 자연어 텍스트를 그래프 데이터로 손쉽게 변환할 수 있게 되었습니다.

이어지는 실습 과정에서는 그래프 RAG 활용 영역( “Consuming Data from KG”) 이해에 중점을 두고 있으며, 데이터 준비과정에는 임의로 그래프를 정의하여 사용할 예정입니다. 기존에 보유한 자체 데이터를 그래프 RAG에 활용하고자 하는 경우, 그래프 데이터 준비과정을 먼저 거쳐야 합니다. 이 블로그에서는 이 과정을 상세하게 다루지 않지만, re:invent 2023 발표 영상 또는 Github의 샘플코드를 참고하여, Amazon Bedrock을 그래프 데이터 준비 과정에 활용하는 방법을 확인할 수 있습니다.

그래프 RAG 활용 방법 소개 및 실습

이 섹션에서는 그래프 RAG에서 정보를 획득하기 위한 두 가지 활용 방법을 소개할 예정입니다.

• 벡터 검색: 자연어를 벡터로 임베딩하고, 벡터 간 유사도 계산을 통해 문맥적 의미가 유사한 정보 추출
• 그래프 순회: 키워드를 기준으로 그래프 속 엔티티 간 연결관계를 순회하여, 도달한 엔티티에서 필요한 정보 추출

두 방법은 데이터 접근 로직 및 결과에 확연한 차이를 갖습니다. 따라서, 데이터 특성에 따라 적합한 방법을 선택하거나, 두 방법을 동시에 사용함으로써 사용자에 더욱 풍부한 답변을 제공할 수 있습니다. 이제 그래프 RAG에서 두 방법을 어떻게 활용하는지 소개하겠습니다.

1) 벡터 검색으로 정보 추출

벡터 검색은 구조화되지 않은 자연어를 원형 그대로 저장/검색하는 데 유용하여, 기존 RAG 시스템에 널리 사용되고 있습니다. 그래프 RAG에서도 그래프 데이터베이스에 벡터 인덱스를 정의하고, 유사한 정보를 신속하게 알아내는 것이 가능합니다. 아래 실습에서는 그래프 RAG에서 1) 벡터 검색의 사용방법과 2) 단순 벡터 검색에서 발생하는 문제 상황, 3) 그래프 RAG에서 이 문제를 해결하는 방법에 대해 알아봅니다. 구체적 구문 설명은 코드 내 주석을 참고하시길 바랍니다.

# 필요한 라이브러리 설치
%pip install langchain wikipedia neo4j transformers py2neo graphdatascience

# 라이브러리 불러오기
import re, os, json
import pandas as pd
import boto3
import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="wikipedia")

from langchain.document_loaders import WikipediaLoader
from langchain.embeddings import BedrockEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from transformers import AutoTokenizer
from langchain.chains import GraphCypherQAChain
from langchain.chat_models import BedrockChat
from langchain.vectorstores.neo4j_vector import Neo4jVector
from langchain.graphs import Neo4jGraph
from langchain.prompts.prompt import PromptTemplate
from langchain.llms import Bedrock
from py2neo import Graph
from graphdatascience import GraphDataScience

# 클라이언트에서 Neo4j 원격 호출환경 구성
uri = "bolt://{Neo4j가 설치된 EC2 인스턴스의 IP 주소}:7687"
username = "neo4j"
password = "{신규 설정된 비밀번호}"
graph = Graph(uri, auth=(username, password))
gds = GraphDataScience(
    uri,
    auth=(username, password),
    aura_ds=False
)

# boto3의 Amazon Bedrock 클라이언트 및 Embedding 모델 설정
REGION = "{Bedrock 모델 접근이 허용된 리전}" # (예)'us-east-1'
boto3_bedrock = boto3.client('bedrock-runtime', region_name=REGION)
bedrock_embeddings = BedrockEmbeddings(model_id="amazon.titan-embed-text-v1", client=boto3_bedrock, region_name=REGION)

#------- 각종 헬퍼 모듈 정의 -------#
# Neo4j 데이터 초기화
def init_graph_data():
    graph.run("MATCH (n) DETACH DELETE n")

def bert_len(text):
    tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
    tokens = tokenizer.encode(text, max_length=512, truncation=True)
    return len(tokens)

# 하나의 문서를 여러 개의 chunk로 분리 -> (langchain) Document로 리턴
def chunk_document(query, chunk_size, chunk_overlap):
    raw_documents = WikipediaLoader(query=query, doc_content_chars_max=20000, load_max_docs=1).load()
    text_splitter = RecursiveCharacterTextSplitter(
              chunk_size = chunk_size,
              chunk_overlap  = chunk_overlap,
              length_function = bert_len,
              separators=['\n\n', '\n', ' ', ''],
          )
          
    documents = text_splitter.create_documents([raw_documents[0].page_content])
    return documents

# 하나의 문서를 여러 개의 chunk로 분리 -> Text로 리턴
def chunk_text(query, chunk_size, chunk_overlap):
    raw_documents = WikipediaLoader(query=query, doc_content_chars_max=20000, load_max_docs=1).load()
    text_splitter = RecursiveCharacterTextSplitter(
              chunk_size = chunk_size,
              chunk_overlap  = chunk_overlap,
              length_function = bert_len,
              separators=['\n\n', '\n', ' ', ''],
          )
    documents = text_splitter.split_text(raw_documents[0].page_content)
    return documents


# 벡터 임베딩을 그래프에 로드
def vector_load(text):
    neo4j_vector = Neo4jVector.from_documents(
        text,
        bedrock_embeddings,
        url=uri,
        username=username,
        password=password
    )
    return neo4j_vector

# 벡터 검색 결과 확인
def search_context(neo4j_vector, question):
    vector_results = neo4j_vector.similarity_search(question, k=1)
    vector_result = vector_results[0].page_content
    return vector_result

위 코드 구문에서는 Neo4j 데이터베이스 연결 환경을 구성하고, 텍스트를 벡터로 변환한 후 그래프로 임베딩하는 모듈들을 정의합니다. 위에 정의된 vector_load() 함수는 Amazon Bedrock의 임베딩 모델을 활용하기 때문에, 이 모듈을 실행하기에 앞서 Amazon Bedrock의 Model Access에서 모델 별 접근 권한을 확보해야 합니다. 이 실습 예제는 아래 2개의 모델을 활용합니다.

1. Amazon – Titan Embeddings G1 – Text
2. Anthropic – Claude v2.1
   • 여기서는 Anthropic의 Claude 2.1 모델을 언어 모델로 사용하며, 이밖에 AI21 Labs, Cohere, Meta, Stability AI 등 다른 언어 모델도 선택할 수 있습니다.

테스트 시나리오에서는 AWS의 스토리지 서비스인 Amazon S3에 대해 설명하는 위키피디아 문서를 벡터 형태로 Neo4j DB에 저장한 후, 벡터 검색으로 원하는 컨텍스트를 추출합니다. 이 때, chunk 크기가 답변의 정확도에 미치는 영향을 알아볼 예정입니다.

첫 번째는 데이터 chunk 크기를 2000으로 설정하여, 전체 문서(<2000)가 하나의 벡터에 임베딩 되도록 합니다.

# 그래프 데이터 초기화
init_graph_data()

query = "Amazon S3"
chunk_size = 2000
chunk_overlap = 30

# 문서를 chunk_size = 2000 으로 분리
docs = chunk_document(query, chunk_size, chunk_overlap)

# 분리된 문서를 그래프에 로드
long_vector = vector_load(docs)

question = "What are the different storage classes offered by Amazon S3? Please provide a list of all available storage classes."
print(search_context(long_vector, question))

# 출력 결과
...
=== Amazon S3 storage classes ===
Amazon S3 offers nine different storage classes with different levels of durability, availability, and performance requirements.
Amazon S3 Standard is the default. It is general purpose storage for frequently accessed data.
Amazon S3 Express One Zone is a single-digit millisecond latency storage for frequently accessed data and latency-sensitive applications. It stores data only in one availability zone.
Amazon S3 Standard-Infrequent Access (Standard-IA) is designed for less frequently accessed data, such as backups and disaster recovery data.
Amazon S3 One Zone-Infrequent Access (One Zone-IA) performs like the Standard-IA, but stores data only in one availability zone.
Amazon S3 Intelligent-Tiering moves objects automatically to a more cost-efficient storage class.
Amazon S3 on Outposts brings storage to installations not hosted by Amazon.
Amazon S3 Glacier Instant Retrieval is a low-cost storage for rarely accessed data, but which still requires rapid retrieval.
Amazon S3 Glacier Flexible Retrieval is also a low-cost option for long-lived data; it offers 3 retrieval speeds, ranging from minutes to hours.
Amazon S3 Glacier Deep Archive is the lowest cost storage for long-lived archive data that is accessed less than once per year and is retrieved asynchronously.The Amazon S3 Glacier storage classes above are distinct from Amazon Glacier, which is a separate product with its own APIs.
...

위와 같이, 전체 문서를 하나의 chunk에 임베딩 한다면, “Amazon S3의 스토리지 클래스 종류”에 대해 질문하거나 다른 어떤 질문을 하더라도 전체 문서를 컨텍스트로 제공할 것입니다. 이렇게 추출된 컨텍스트에는 원하는 정보가 누락되지는 않습니다. 하지만, 너무 긴 문서를 하나의 chunk에 임베딩 할 경우 1) 필요한 문맥적 의미가 희석되어 적합한 chunk가 추출되지 않거나, 2) 연관성이 낮은 문장들까지 컨텍스트에 너무 많이 포함하면서 답변 품질에 부정적 영향을 미치기도 합니다. 대부분의 RAG 기법에서는 이 문제를 해결하기 위해 데이터 chunk를 작게 분리하는 방법을 채택합니다.

이번에는 chunk 크기를 200으로 줄여서, Amazon S3에 대한 위키피디아 문서가 10개의 chunk에 나눠 저장되는 상황을 가정해보겠습니다. create_text_embedding_entries() 모듈은 데이터를 원하는 크기의 chunk로 나눈 다음, Amazon Bedrock을 통해 벡터 임베딩 합니다. 각 chunk의 속성에는 seqid를 할당하여 문서 속에서 각 chunk의 순서 및 위치 정보를 나타내도록 합니다. update_embeddings() 모듈은 ‘Amazon S3’라는 핵심 개념과 각각의 chunk가 CHUNKED 라는 그래프 속 연결관계를 맺도록 정의합니다.

def chunks(xs, n=3):
    n = max(1, n)
    return [xs[i:i + n] for i in range(0, len(xs), n)]

# chunk 단위의 임베딩을 생성하며, chunk 식별 정보(sequence) 생성
def create_text_embedding_entries(query, chunk_size, chunk_overlap):
    docs = chunk_text(query, chunk_size, chunk_overlap)
    service_name = query
    res = []    
    seq_id = -1
    
    for d in chunks(docs):
        embeddings = bedrock_embeddings.embed_documents(d)
        for i in range(len(d)):
            seq_id += 1
            res.append({'name': service_name,
                        'seqId': seq_id,
                        'contextId': service_name + str(seq_id),  # unique 
                        'textEmbedding': embeddings[i],  # chunked
                        'text': d[i]  })
    return res

# 제공된 벡터 임베딩을 그래프에 업데이트
def update_embeddings(emb):
    total = len(emb)
    count = 0
    for d in chunks(emb, 100):
        gds.run_cypher('''
        UNWIND $records AS record
        MERGE(s:Service {name:record.name})
        CREATE(c:Chunk {chunkid:record.contextId, seqid:record.seqId, text:record.text})
        MERGE(s)-[:CHUNKED]->(c)
        with c, record
        CALL db.create.setVectorProperty(c, 'embedding', record.textEmbedding)
        YIELD node
        RETURN distinct 'done'
        ''', params = {'records':d})
        count += len(d)

# 그래프 데이터 초기화
init_graph_data()

chunk_size = 200
query = 'Amazon S3'

# 소규모 chunk를 벡터로 생성하고, 그래프에 업데이트
vector = create_text_embedding_entries(query, chunk_size, chunk_overlap)
update_embeddings(vector)

위 구문을 수행한 이후 데이터 구조는 아래와 같습니다.

이제 벡터 검색을 사용하여, ‘Amazon S3의 스토리지 클래스’라는 질문에 해당하는 chunk를 컨텍스트로 추출하겠습니다.

# 벡터 인덱스를 활용해 높은 유사도의 chunk를 추출
vector_search = """
WITH $embedding AS e
CALL db.index.vector.queryNodes('vector', $k, e) yield node, score
RETURN node.text AS result
"""

graph_instance = Neo4jGraph(url=uri, username=username, password=password)

# 질문을 임베딩으로 변환
question = "What are the different storage classes offered by Amazon S3? Please provide a list of all available storage classes."
embedding = bedrock_embeddings.embed_query(question)

# 컨텍스트 확보
context = graph_instance.query(
    vector_search, {'embedding': embedding, 'k': 3})
   
context = [el['result'] for el in context]
print(context[0])

# 출력 결과
=== Amazon S3 storage classes ===
Amazon S3 offers nine different storage classes with different levels of durability, availability, and performance requirements.
Amazon S3 Standard is the default. It is general purpose storage for frequently accessed data.
Amazon S3 Express One Zone is a single-digit millisecond latency storage for frequently accessed data and latency-sensitive applications. It stores data only in one availability zone.
Amazon S3 Standard-Infrequent Access (Standard-IA) is designed for less frequently accessed data, such as backups and disaster recovery data.
Amazon S3 One Zone-Infrequent Access (One Zone-IA) performs like the Standard-IA, but stores data only in one availability zone

# 프롬프트 템플릿 정의
prompt_template = """
Human: You are a helpful, respectful, and honest assistant, dedicated to providing valuable and accurate information.
Guidance for answers below
    Answer the question only using the in the context given below, and not with the prior knowledge.
    If you don't see answer in the context just Reply "Sorry , the answer is not in the context so I don't know".

Now read this context and answer the question. 
{context}

Based on the provided context above and information from the retriever source, provide a detailed answer to the below question
{question}

If the information is not available in the context , respond with "don't know."

Assistant: """

prompt = PromptTemplate.from_template(prompt_template)

# Bedrock의 언어모델로 Claude 선택하고, 컨텍스트 기반 답변 생성
llm = Bedrock(model_id="anthropic.claude-v2:1", client=boto3_bedrock, model_kwargs={'max_tokens_to_sample':10240, "temperature": 0})
answer = llm(prompt.format(context=context, question=question))
print(answer)

# 출력 결과
Based on the context, Amazon S3 offers the following storage classes:
- Amazon S3 Standard - general purpose storage for frequently accessed data
- Amazon S3 Express One Zone - single-digit millisecond latency storage for frequently accessed data and latency-sensitive applications. Stores data only in one availability zone.  
- Amazon S3 Standard-Infrequent Access (Standard-IA) - for less frequently accessed data like backups and disaster recovery
- Amazon S3 One Zone-Infrequent Access (One Zone-IA) - similar to Standard-IA but stores data only in one availability zone
The context states that "Amazon S3 offers nine different storage classes with different levels of durability, availability, and performance requirements."
However, it only lists out the four storage classes mentioned above.
Since the full list of nine storage classes is not provided in the context, I do not have enough information to provide the complete list.

위와 같이 chunking에 의해 그래프 정보가 소실 되었을 때, 그래프 RAG에서 이에 대응하는 방법을 알아보겠습니다. 그래프 RAG에서는 전체 문서에서 각 chunk의 문맥적 위치를 그래프 속 관계 및 순서정보(seqid)로 표현할 수 있고, 이를 활용하면 연관된 컨텍스트를 손쉽게 확보할 수 있습니다. 예를 들어, Amazon S3 외 수많은 AWS 서비스 문서가 RAG 데이터베이스에 로드되어 있다고 가정했을 때, 벡터 검색으로 계산된 유사한 chunk를 ‘Amazon S3’ 하위 chunk 중 하나에서 발견하면, Amazon S3에 연결된 다른 문서들을 순회할 수 있는 셈입니다.

그래프를 추가 순회하는 작업은 vector_search 구문에서 엔티티 간 관계 및 속성으로 사전 정의될 수 있습니다. 아래 구문은 문서 전체(#2) 또는 인접한 소수의 chunk(#3)를 추출하도록 정의하는 예시입니다.

# 1. 기본 : 발견된 chunk의 텍스트만 추출
vector_search = """
WITH $embedding AS e
CALL db.index.vector.queryNodes('vector', $k, e) yield node, score
RETURN node.text AS result
"""

# 2. 문서 전체 : 발견된 chunk와 연계된 전체 문서를 리턴
vector_search_full_documents = """
WITH $embedding AS e
CALL db.index.vector.queryNodes('vector', $k, e) yield node, score
MATCH (s:Service)-[:CHUNKED]->(node)
WITH s
MATCH (s)-[:CHUNKED]->(docs:Chunk)
WITH s, docs
ORDER BY docs.seqid ASC
RETURN s AS service, COLLECT(docs.text) AS result
"""

# 3. 인접 chunk : 발견된 chunk와 인접한 chunk의 텍스트를 리턴
vector_search_larger_chunk = """
WITH $embedding AS e
CALL db.index.vector.queryNodes('vector', $k, e) yield node, score
MATCH (s:Service)-[:CHUNKED]->(node)
WITH s, node.seqid AS seqid
MATCH (s)-[:CHUNKED]->(docs:Chunk)
WHERE docs.seqid >= seqid - 1 AND docs.seqid <= seqid + 1
WITH s, docs
ORDER BY docs.seqid ASC
RETURN s AS service, COLLECT(docs.text) AS result
"""

graph_instance = Neo4jGraph(url=uri, username=username, password=password)

question = "What are the different storage classes offered by Amazon S3? Please provide a list of all available storage classes."
embedding = bedrock_embeddings.embed_query(question)
context = graph_instance.query(
    vector_search_full_documents, {'embedding': embedding, 'k': 1})
context = [el['result'] for el in context]

이런 방식으로 컨텍스트를 수집해 활용하면, 작은 chunk 크기를 유지하면서도 완전한 답변을 제공할 수 있습니다.

answer = llm(prompt.format(context=context, question=question))
print(answer)

# 출력 결과
 Based on the context, Amazon S3 offers the following storage classes:
- Amazon S3 Standard - general purpose storage for frequently accessed data
- Amazon S3 Express One Zone - single-digit millisecond latency storage for frequently accessed data and latency-sensitive applications. Stores data only in one availability zone.  
- Amazon S3 Standard-Infrequent Access (Standard-IA) - for less frequently accessed data like backups and disaster recovery
- Amazon S3 One Zone-Infrequent Access (One Zone-IA) - similar to Standard-IA but stores data only in one availability zone
- Amazon S3 Intelligent-Tiering - moves objects automatically to a more cost-efficient storage class
- Amazon S3 on Outposts - brings storage to installations not hosted by Amazon 
- Amazon S3 Glacier Instant Retrieval - low-cost storage for rarely accessed data that still requires rapid retrieval
- Amazon S3 Glacier Flexible Retrieval - low-cost option for long-lived data with 3 retrieval speeds
- Amazon S3 Glacier Deep Archive - lowest cost for long-lived archive data accessed less than once per year and retrieved asynchronously

2) 그래프 순회를 활용한 정보 추출

그래프 순회는 개념 사이의 연결을 반영한 심도 있는 정보를 추출합니다. 벡터 기반 검색이 유사도에 따라 상위 K개의 chunk를 찾아내고 상향식으로 인접한 chunk에 접근할 수 있다면, 그래프 순회는 사전 정의된 그래프 스키마를 기준으로 주요 개념을 식별하고 하향식으로 구체적 정보에 접근합니다.

그래프 구조 속에서 필요한 정보를 찾기 위해서는 그래프 데이터베이스의 쿼리 언어, 스키마, 연결 구조 등을 활용해야 합니다. 많은 그래프 데이터베이스가 대규모 언어 모델과 통합 가능한 Graph QA Chain을 제공하고 있어서, 이를 Amazon Bedrock과 함께 활용하면 사용자의 자연어 질문을 기반으로 그래프 속 관계를 순회할 수 있습니다.

이어지는 실습에서는 Amazon Bedrock을 활용한 그래프 순회 과정을 소개할 예정입니다. 이에 앞서, Amazon S3 외에 Amazon EC2 및 Amazon EBS를 서비스 유형으로 추가하고, Amazon EC2가 Amazon S3 및 Amazon EBS를 ‘StorageType’으로 사용한다는 hasStorageType 관계를 정의합니다. 이를 위해, 다음 쿼리를 Neo4j 브라우저에서 실행합니다.

MATCH (s1:Service {name: 'Amazon S3'})
MERGE (s2:Service {name: 'Amazon EBS'})
MERGE (ec2:Service {name: 'Amazon EC2'})
MERGE (ec2)-[:hasStorageType]->(s1)
MERGE (ec2)-[:hasStorageType]->(s2)
# 위 다섯 줄이 하나의 쿼리 구문이며, Neo4j 쿼리 편집기에 붙여넣은 후 파란색 화살표를 누르면 실행됩니다.

아래 과정은 Amazon S3의 경우와 마찬가지로, Amazon EC2 및 Amazon EBS 대한 1) 위키피디아 문서를 로드하고, 2) 이를 여러 개의 chunk로 나눈 뒤, 3) 해당 chunk들을 벡터로 임베딩하여 그래프에 업데이트합니다. 아래 구문으로 작업을 수행하며, 작업이 완료 되기까지 5분 이내로 소요됩니다.

queries = ['Amazon EC2', 'Amazon EBS']

for query in queries:
    vector = create_text_embedding_entries(query, chunk_size, chunk_overlap)
    update_embeddings(vector)

작업이 완료되면, 위와 같은 그래프의 준비가 끝났습니다. 이어지는 실습에서는 “Amazon EC2에서 사용 가능한 스토리지 유형과 특징, 고성능에 좋은 스토리지 유형”에 대해 질문하면서, 앞서 알아봤던 벡터 검색과 그래프 순회 방법을 비교하겠습니다.

쿼리 실행 중에 “Failed to read from defunct connection”과 같은 메시지가 나타날 수 있습니다. 이는 Neo4j와의 연결이 타임아웃으로 인해 일시적으로 끊어지면서 발생하는 현상입니다.
– 경고 메시지: 이는 일반적으로 큰 문제가 아니므로, 걱정하지 않으셔도 됩니다. 쿼리는 정상적으로 실행되고 있을 가능성이 높습니다.
– 오류 메시지: 만약 오류 메시지가 나타난다면, 연결이 완전히 끊어진 것일 수 있습니다. 이 경우, 쿼리를 재실행하면 새로운 연결이 생성되어 문제가 해결됩니다.

벡터 검색을 활용한 답변

# 벡터 검색 후 인접 chunk를 컨텍스트로 추출
vector_search = """
WITH $embedding AS e
CALL db.index.vector.queryNodes('vector', $k, e) yield node, score
MATCH (s:Service)-[:CHUNKED]->(node)
WITH s, node.seqid AS seqid
MATCH (s)-[:CHUNKED]->(docs:Chunk)
WHERE docs.seqid >= seqid - 3 AND docs.seqid <= seqid + 3
WITH s, docs
ORDER BY docs.seqid ASC
RETURN s AS service, COLLECT(docs.text) AS result
"""

question = """
Tell me about the different storage types available for Amazon EC2. 
I need detailed information on the 'characteristics' for each 'storage'. 
Additionally, I want to know which storage type is best suited for 'performance'.
"""

# 질문을 벡터 임베딩으로 변환
embedding = bedrock_embeddings.embed_query(question)
context = graph_instance.query(
    vector_search, {'embedding': embedding, 'k': 3})
context = [el['result'] for el in context]
print(context[0])

위 벡터 검색으로 출력된 컨텍스트는 Amazon EC2 위키피디아 문서의 chunk들을 반환했습니다. 위 예제 코드에서는 Top-3의 유사한 chunk를 활용하도록 했으며, 소규모 chunk 크기의 영향을 줄이고자 seqid를 기준으로 인접한 chunk까지 함께 추출하도록 했습니다. 추출된 컨텍스트를 언어 모델에 제공하여 답변을 구성했을 때, 제시된 키워드에 맞춰 답변했지만 Amazon S3 및 Amazon EBS 문서에서 얻을 수 있는 구체적 정보들 충분히 반영하지는 못했습니다. (검색 결과는 여러 요인에 따라 변동적 입니다)

vector_answer = llm(prompt.format(context=context, question=question))
print(vector_answer)

# 출력 결과
Based on the context, there are two main types of storage available for Amazon EC2:

1. Instance-store volumes:
- Temporary block level storage volumes
- Survive rebooting but are lost when the EC2 instance is stopped or terminated
- Good for temporary data or buffers

2. EBS (Elastic Block Store) volumes:  
- Provide persistent block level storage
- Act as hard drives that can be formatted with a file system and mounted
- Support advanced features like snapshots and cloning
- Range from 1GB to 16TB in size
- Built on replicated storage for data redundancy 
- Best suited for performance as they can be provisioned with high IOPS for low latency

For performance, EBS volumes are better as they allow provisioning with high IOPS (input/output operations per second) which provides low latency. The context does not provide comparative details on the IOPS capabilities of instance-store vs EBS volumes.

So in summary, EBS volumes are best suited for performance given their ability to specify IOPS rates when provisioning the volumes

이제 그래프 순회를 통해 답변을 구성하여, 그 차이점을 비교해보겠습니다.

그래프 순회를 통한 답변

그래프 순회로 답변을 제공하기 위해서는 두 단계를 거쳐야 합니다. 첫째, 자연어로 된 질문을 그래프 쿼리 언어(Neo4j에서는 Cypher)로 변환하고 이를 통해 컨텍스트를 수집합니다. 둘째, 수집된 컨텍스트를 바탕으로 질문에 대한 답변을 구성합니다. 두 과정에서 모두 Amazon Bedrock의 대규모 언어 모델을 활용할 수 있습니다.

첫 번째 단계에서는 사용자의 질문을 Cypher 쿼리 언어로 변환하기 위한 프롬프트를 작성합니다. 질문에 맞는 컨텍스트를 효과적으로 취합하기 위해서는 잘 구성된 프롬프트가 필수적이며, 여기에는 사용자 질문, 쿼리 변환을 위한 지침, 예시, 그리고 그래프 스키마에 대한 구체적인 정보를 포함하는 것이 좋습니다. 아래 구문에서는 쿼리 변환 프롬프트 템플릿을 정의하고, BedrockChat 클래스의 인스턴스를 생성합니다.

CYPHER_GENERATION_TEMPLATE = """

You are an excellent Assistant for generating Cypher Query Language for graph searches. 
You will be exploring a graph that represents the connections between AWS services.

<instruction>
Guidance for answers below
Each Service has relationships with other Services.
 - (s1:Service)-[:]->(s2:Service)
Each Service node is connected to Chunks containing detailed information about the service:
 - (s1:Service)-[:CHUNKED]->(c:Chunk)
Each Chunk has detailed information about the service in its text.
 - (c:Chunk {{c.text}})
 - RETURN c.text
</instruction>

<example>
Here are a few examples of generated Cypher statements for particular questions:
# Question :
Explain instance types Amazon EC2 has.
# Generated Cypher :
MATCH (ec2:Service {{name:"Amazon EC2"}})-[:hasInstanceType]->(s:Service)
WITH s
MATCH (s:Service)-[:CHUNKED]->(c:Chunk)
RETURN COLLECT(DISTINCT c.text) AS Text

# Question :
Explain the cost of each instance type of Amazon EC2.
# Generated Cypher :
MATCH (ec2:Service {{name:"Amazon EC2"}})-[:hasInstanceType]->(s:Service)
WITH s
MATCH (s:Service)-[:CHUNKED]->(c:Chunk)
WHERE c.text CONTAINS "cost"
RETURN COLLECT(DISTINCT c.text) AS Text
</example>

Use only the provided relationship types and properties in the schema.
Do not use any other relationship types or properties that are not provided.
<schema>
{schema}
</schema>

The question is: 
{question} 
"""

CYPHER_GENERATION_PROMPT = PromptTemplate(
    input_variables=['schema', 'question'], validate_template=True, template=CYPHER_GENERATION_TEMPLATE
)

이어서, Langchain의 GraphCypherQAChain 라이브러리를 이용합니다. 이 라이브러리는 1) 자연어 질문을 입력받고, 2) 제시된 언어모델을 활용해 Cypher 쿼리 언어로 변환한 다음, 3) 쿼리 수행 결과 그래프에서 추출된 정보를 답변으로 제공합니다.

graph_instance = Neo4jGraph(url=uri, username=username, password=password)
chat_llm = BedrockChat(model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.1}, region_name=REGION)

chain = GraphCypherQAChain.from_llm(
    chat_llm,
    graph=graph_instance, 
    cypher_prompt=CYPHER_GENERATION_PROMPT,
    verbose=True,
    validate_cypher=True,
    return_direct=True
)

다음 단계에서는 GraphCypherQAChain을 통해 얻은 결과를 ‘사실(Fact)’로 제공하고, 질문에 맞게 응답을 재구성합니다.

def chat(question):
    r = chain(question)
    
    summary_prompt_tpl = f"""Human: 
    Fact: {json.dumps(r['result'])}

    * Describe the above fact as if you are answering this question "{r['query']}"
    * Provide the response according to the keywords provided within the quotation marks.
    * When the fact is not empty, assume the question is valid and the answer is true
    * Do not return helpful or extra text or apologies
    * List the results in rich text format if there are more than one results
    Assistant:
    """
    return llm(summary_prompt_tpl)

def chat_response(input_text):
    try:
        return chat(input_text)
    except:
        return "I'm sorry, there was an error retrieving the information you requested."
        
graph_answer = chat(question) 
print(graph_answer)

위 구문을 실행하면, ‘Amazon EC2에서 활용 가능한 스토리지 타입’ 이라는 질문과 연관된 엔티티 및 관계를 그래프에서 순회하고, ‘characteristics’ , ‘performance’ 등의 키워드가 포함된 컨텍스트를 수집합니다. (검색 결과는 여러 요인에 따라 변동적입니다)

> Entering new GraphCypherQAChain chain...
Generated Cypher:
MATCH (ec2:Service {name:"Amazon EC2"})-[:hasStorageType]->(storage:Service)
WITH storage 
MATCH (storage)-[:CHUNKED]->(chunk:Chunk) 
WHERE chunk.text CONTAINS "characteristics" OR chunk.text CONTAINS "performance"
RETURN storage.name AS StorageType, 
       COLLECT(DISTINCT chunk.text) AS Details
ORDER BY 
  CASE storage.name 
    WHEN "Amazon EBS" THEN 0
    WHEN "Amazon EC2 Instance Store" THEN 1
    ELSE 2
  END
> Finished chain.
 Based on the fact provided, there are two main storage types available for Amazon EC2:

1. Amazon Elastic Block Store (EBS)

Characteristics:
- Provides raw block-level storage that can be attached to EC2 instances
- Used by Amazon RDS
- Offers a range of options for storage performance and cost
- Has SSD-backed storage for transactional workloads like databases 
- Has disk-backed storage for throughput intensive workloads like MapReduce

Best for performance: 
- io1 and io2 volume types offer highest performance in IOPS
- st1 and sc1 offer highest throughput in MB/s

2. Amazon S3

Characteristics: 
- Object storage service offered by AWS
- Offers 9 different storage classes with different durability, availability and performance
- Has storage classes like Standard, Standard-IA, One Zone-IA etc.

Best for performance:
- Amazon S3 Standard provides low latency general purpose storage
- Amazon S3 Express One Zone provides single-digit millisecond latency

출력 결과를 보면, Amazon EC2에 대한 질문으로 출발하여 그래프 관계를 통해 Amazon EBS와 Amazon S3 문서에 접근하고, 해당 하위 문서에서 구체적 근거(특징 및 고성능 스토리지 타입)를 컨텍스트로 수집하고 있습니다. 아래 표에서 두 방법으로 발생하는 답변의 차이를 확인할 수 있습니다. (검색 결과는 여러 요인에 따라 변동적 입니다)

그래프 순회 방법은 그래프 쿼리를 통해 엔티티 간 관계를 탐색하고, 최종적으로 도달한 엔티티에서 필요한 정보를 추출합니다. 그렇기 때문에, Amazon EC2에서 hasStorageType 연결 관계로 Amazon S3 및 Amazon EBS 정보에 접근했던 것처럼, 그래프 순회 과정에서 최대한 정확한 정보에 도달하는 것이 중요합니다. 만약, 사용자의 질문 내용이 그래프 스키마에 잘 반영되어 있지 않으면 GraphQAChain의 쿼리 변환 정확도가 떨어질 수 있습니다.

필요한 컨텍스트의 범위에 따라 벡터 검색 및 그래프 순회 방법을 함께 사용하거나, 그래프 순회 방법을 우선적으로 활용하더라도 안정성을 높이기 위해 벡터 검색을 Fallback 옵션으로 제공하는 것이 좋습니다. 아래는 벡터 검색 및 그래프 순회를 함께 사용하는 시나리오를 도식화 한 것입니다.

결론

그래프 RAG는 복잡하게 연결된 관계를 유연하게 정의하고, 벡터 검색 및 그래프 순회를 정보 탐색 옵션으로 함께 제공하여, 사내 지식검색 시스템, 추천 시스템, 뉴스 기사 검색 등 다양한 시나리오에 접목 가능합니다. 이 블로그에서는 그래프 RAG의 적용 방법에 대한 소개 및 실습을 아래 순서로 진행했습니다.

• 첫 번째, 구조화 되지 않은 텍스트를 그래프 형식으로 변환하는 데이터 준비 과정에 대해 설명했습니다. 이 과정에서도 Amazon Bedrock의 대규모 언어모델을 활용함으로써, 자연어 기반 데이터를 원하는 구조의 지식 그래프로 표현할 수 있습니다.
• 두 번째, 그래프 RAG에서 벡터 검색 방법을 간단한 질의응답에 활용하면서, 컨텍스트 추출 시 chunking으로 인해 발생할 수 있는 정보 누락 문제를 확인했습니다. 이후, 그래프 RAG에서 인접 노드를 탐색하여 추가 컨텍스트를 확보하는 대응 방안을 소개했습니다.
• 세 번째, 그래프를 순회하여 정보를 얻어내는 그래프 RAG 활용 방법을 소개했습니다. 특히, 벡터 검색과 비교하여 그래프 순회 방법의 장단점에 대해 알아봤습니다.

Amazon Bedrock과 Neo4j를 활용한 그래프 RAG 구축에 대한 추가 설명은 다음 링크에서 확인하실 수 있습니다.