Ontology & Knowledge Graph

RAG 앱을 위한 지식 그래프 구축: YouTube Transcript를 Neo4j로!

hsystems 2026. 8. 8. 10:30
728x90
반응형

소개

이번 블로그에서는 YouTube 비디오 대본을 Knowledge Graph로 스크랩하는 방법을 알아볼 거예요. Retrieval-Augmented Generation (RAG) 애플리케이션을 위한 거죠. 초기 대본을 저장하기 위해 Google Cloud Platform을 사용하고, 대본에서 문서를 만들기 위해 Langchain을, 결과 문서를 저장하는 Graph Database로 를 사용할 거예요. 이번 예시에서는 Knowledge Graph가 객관적인 음악적 사실을 포함하고 있어요. 앤서니 판타노가 엄선한 몇 가지 음악 장르를 직접 다루고 있죠.

Watermelons
멜론 그래프 | pixlr.com 제공

요구사항

  • Google Cloud Platform 계정
  • GCP YouTube Data API v3에 대한 액세스
  • 무료 Neo4j AuraDB 인스턴스

서비스 설정

따라오시려면 먼저 몇 가지 설정을 해야 해요. 그렇지 않다면 이 부분을 건너뛰셔도 괜찮아요. 이 프로젝트의 GitHub는 여기서 찾을 수 있어요:

GCP

이 프로젝트에서는 최소한 다음 사항에 대해 잘 알고 있어야 해요: Google Cloud Platform (GCP). 제공된 Storage Bucket과 YouTube Data API를 사용할 거예요. 스토리지 옵션을 사용하는 경우 결제를 계정에 연결해야 해요. 이 프로젝트는 쉽게 무료 등급 가격 내에서 유지될 거지만, 진행한 후에는 프로젝트에서 생성했을 수 있는 Storage Bucket과 해당 콘텐츠 및 기타 리소스를 삭제해야 해요.

YouTube API 단계:

  • 새 프로젝트 만들기 GCP.
  • YouTube Data API v3를 검색하고 API를 활성화하세요.
  • 자격 증명을 생성하라는 메시지가 표시될 거예요. — 링크를 따라가세요.
  • 공개 데이터를 선택합니다.
  • 다음을 선택합니다.
  • 새 API 키를 안전한 위치에 복사하고 완료를 선택합니다.

Cloud Storage 단계:

  • 동일한 GCP 프로젝트에서…
  • 사이드바에서 Cloud Storage → 버킷으로 이동합니다.
  • 새 버킷을 생성하고 버킷 이름을 안전한 위치에 저장합니다.

서비스 계정 단계 (Cloud Storage에 필요):

  • 사이드바에서 IAM 및 관리자 → 서비스 계정으로 이동합니다.
  • 서비스 계정 만들기를 선택합니다.
  • 서비스 계정에 적절한 이름을 제공합니다.
  • 만들기 및 계속을 선택합니다.
  • 역할에서 선택 .
  • 완료를 선택합니다.
  • 새 서비스 계정 아래에서 작업 탭을 선택하고 키 관리를 선택합니다.
  • 키 추가 → 새 키 만들기를 선택합니다.
  • 선택하다 JSON.
  • 키를 생성하여 안전한 위치에 다운로드하세요.

Neo4j

Neo4j는 최근에 Vector Index를 추가한 기본 Graph Database에요. Vector Index 검색 기능이 추가되었죠. 이를 통해 Vector Database의 장점과 Graph Database의 장점을 결합할 수 있어요.

Neo4j는 여기 사용 사례에 충분한 능력을 갖춘 모든 신규 사용자에게 무료 AuraDB 인스턴스를 제공하고 있어요.

오라 단계:

  • 이 링크를 따라 Aura 콘솔로 이동하세요.
  • 무료 AuraDB 인스턴스를 생성하세요 (AuraDS가 아니에요).
  • 연결 정보가 담긴 파일을 안전한 곳에 잘 보관하세요.

참고: Aura 인스턴스는 3일 동안 활동이 없으면 일시 중단되고, 액세스하지 않으면 결국 자동으로 삭제될 거예요.

환경 설정

프로젝트 루트 디렉터리의 명령줄에서 다음 명령어를 실행하세요.

python3 -m venv venv

IDE에서 venv/bin/activate 파일을 열고 파일 맨 아래에 다음 줄을 추가하세요. 이 정보는 위 단계에서 저장한 파일에 있는 정보로 채워야 해요.

이 방법은 Linux/MacOS에서 작동해요. Windows에서는 해당 쉘에 맞는 activate.* 파일을 편집하세요.

export NEO4J_USERNAME=”DB USERNAME”
export NEO4J_PASSWORD="DB PASSWORD"
export NEO4J_DATABASE="DB DATABASE"
export NEO4J_URI="DB_URI"
export YOUTUBE_API_KEY="key"
export GCP_SERVICE_ACCOUNT_KEY_PATH="/path/to/service_account_file/creds.json"
export GCP_BUCKET_NAME="your-video-name"

변수들을 설정했다면 파일을 저장하세요.

데이터 모델

데이터 모델은 전체 기록이 512자 청크(상위)로 분할되고, 그 청크가 다시 140자 크기(하위) 청크로 더 분할되는 상위-하위 문서 구조를 따르고 있어요. 512는 표준 청크 크기이고, 140은 일반적인 질문 크기랍니다 (제가 RAG 애플리케이션을 다뤄본 경험을 바탕으로요).

검색 전략

하위 노드에만 텍스트 Vector Embedding이 포함되고, 벡터 검색 알고리즘의 대상이 돼요. 더 작은 하위 청크만 임베딩함으로써 청크와 질문의 임베딩이 유사한 정보 세분성을 갖도록 신경 쓰고 있죠. 이렇게 하면 유용한 응답을 만들어내기에 충분한 컨텍스트를 제공하면서도 검색 정확도를 높일 수 있어요. 유사도 점수가 가장 높은 항목을 찾으면 그래프 관계를 탐색해서 하위 노드의 상위 항목을 LLM으로 반환할 수 있답니다.

스크랩 기록

채널 ID 가져오기

스크립트를 가져오려면 먼저 The Needle Drop YouTube 채널의 채널 ID를 검색해야 해요. YouTube API로 검색하고 나타나는 첫 번째 채널을 선택하면 돼요.

검색어 형식을 다음과 같이 지정하세요: the+needle+drop

자신의 YouTube 채널에서 스크립트를 스크랩하고 싶다면 계정 → 고급 설정 → 채널 ID에서 채널 ID를 찾을 수 있어요.

def _get_channel_id() -> str:

    address = f"https://www.googleapis.com/youtube/v3/search?q=the+needle+drop&key={os.environ.get('YOUTUBE_API_KEY')}&part=snippet"
    req = requests.get(address)
    data = req.json()
    
    return data['items'][0]['snippet']['channelId']

원하는 재생목록 ID 받기

몇몇 재생목록 ID가 포함된 JSON 파일을 준비해뒀어요. 하지만 다음 방법으로 다른 재생목록 ID를 찾을 수도 있어요.

  • 관심 있는 YouTube 채널 페이지로 이동하세요.

예:

  • 채널 재생목록 페이지로 이동하세요.
  • 원하는 재생목록을 클릭하세요.
  • 주소 표시줄에서 "list=" 다음에 나오는 ID를 복사하세요.

예:

  • 다음과 같이 resources/playlist_ids.json 파일에 재생목록 ID와 제목을 추가하세요.
"electronic": "PLP4CSgl7K7ormBIO138tYonB949PHnNcP"

동영상 정보 얻기

이제 YouTube 동영상 정보 가져오기를 시작할 수 있어요. 각 API 호출은 최대 50개의 동영상이 포함된 페이지를 반환하므로, 모든 동영상을 검색할 때까지 "nextPageToken" 값을 사용해서 각 페이지를 계속 찾아야 해요. 다음 재귀적 방법을 사용하면 이걸 할 수 있죠.

def scrape_video_info(self, next_page_token: str = None, total_results: int = -1, videos: List[str] = []) -> List[str]:
                
    address = f"https://www.googleapis.com/youtube/v3/playlistItems?playlistId={self.playlist_id}&key={os.environ.get('YOUTUBE_API_KEY')}&part=snippet&maxResults=50"
    
    if not next_page_token:
        vid_req = requests.get(address)
    
    else:
        vid_req = requests.get(address+f'&pageToken={next_page_token}')
        
    vids = vid_req.json()
    
    if total_results == -1:
        total_results = vids['pageInfo']['totalResults']
        print('total results set: ', total_results)
    
    videos += [{"id": x['snippet']['resourceId']['videoId'], 
                "title": x['snippet']['title'],
                "publish_date": x['snippet']['publishedAt'][:10]} for x in vids['items']]
    
    
    print("total results: ", total_results)
    print("ids retrieved: ", len(videos), "n")
    
    if "nextPageToken" not in vids.keys():
        print("complete")
        self._scraped_video_info = videos
        return videos
    
    self.scrape_video_info(next_page_token=vids['nextPageToken'], total_results=total_results, videos=videos)

모든 동영상을 검색한 후에는 안전하게 보관하기 위해 ID, 제목, 게시 날짜를 GCP 저장소 버킷에 CSV 파일로 저장할 수 있어요. 이는 채널 재생목록에 업로드된 새 동영상을 다시 확인하고 검색하기로 결정했을 때 비교하는 데에도 유용하죠.

Table of the top few csv rows
CSV 파일 헤드

비디오 대본 받기

이제 스크립트를 수집하기 위해 youtube_transcript_api 라이브러리를 사용해 볼게요. 먼저 모든 YouTube 동영상 정보가 포함된 최근 생성된 CSV 파일을 검색합니다. 그런 다음 행을 반복해서 각 동영상의 스크립트 작성을 시도할 수 있어요.

def _create_transcript(video_id: str) -> str:

    raw_transcript = YouTubeTranscriptApi.get_transcript(video_id)

    # instantiate the text formatter
    formatter = TextFormatter()

    # format the video into a string without timestamps, etc...
    transcript_formatted = formatter.format_transcript(raw_transcript)

    # replace newlines with a space 
    return transcript_formatted.replace("n", " ")

가끔 녹취록이 실패하는 경우가 있는데요... 제가 본 바로는 라이브 영상이 예정되어 있지만 아직 녹화되지 않았거나 자막이 활성화되지 않은 경우에 이런 일이 발생할 수 있어요. 업로드된 각 스크립트는 아래 예와 같이 JSON 형식이에요.

{
"video_id": "-80PEYVeI4E", 
"title": "Godspeed You! Black Emperor- Lift Yr. Skinny Fists Like...", 
"publish_date": "2014-07-23", 
"transcript": "[Laughter] hi everyone Anthony fantano here the internet's..."
}

청킹 기록

청킹은 텍스트를 보다 관리하기 쉬운 조각으로 나누어 LLM에 공급하는 프로세스예요. 우리는 LangChain을 사용할 건데요, TokenTextSplitter를 사용해서 각 성적표 상위 청크를 처리할 거예요. 표준 청크 크기는 512이고 중복되는 토큰은 64개죠. 그런 다음 RecursiveCharacterTextSplitter를 사용해서 하위 청크를 생성할 건데, 크기는 140이에요. 또한 성적표에서 "um" 및 "ah"와 같은 단어를 제거해서 텍스트를 정리할 거예요.

기록 로드 중

청크된 후에는 Neo4j Python 드라이버와 일부 Cypher를 사용해서 그래프에 기록을 로드할 수 있어요. 기록 청크를 로드하기 전에 `Source` 및 `Document` Node에 고유성 제약 조건(Uniqueness Constraint)을 만들어야 해요. 이렇게 하면 실수로 중복 데이터를 로드하지 않도록 할 수 있죠.

CREATE CONSTRAINT source_url FOR (s:Source) REQUIRE s.url IS UNIQUE;
CREATE CONSTRAINT document_id FOR (d:Document) REQUIRE d.index IS UNIQUE;

또한 Vector Index 검색을 사용하려면 `Child` Node의 임베딩 속성에 대한 Index를 생성해야 해요. 이는 다음 Query를 통해 수행되는데, 여기서 96은 SpaCy 임베딩의 임베딩 차원이에요.

CREATE VECTOR INDEX `text-embeddings`
FOR (n: Child) ON (n.embedding)
OPTIONS {indexConfig: {
`vector.dimensions`: 96,
`vector.similarity_function`: 'cosine'
}
}

아래 코드 조각은 하위 문서를 반복하고 그래프에 로드할 사전 형식을 지정하는 부분이에요. 하위 문서에는 상위 및 소스 `Node`에 대한 메타데이터가 포함되어 있죠. 이곳은 하위 `Node`에 대한 `Vector Embedding`이 생성되는 곳이기도 해요. 여기서는 SpaCy를 사용하고 있는데, LangChain에 의해 조정된 무료 임베딩 옵션이에요. 그런 다음 사전 목록을 그래프에 로드하게 됩니다.

result = []
for idx, bat in enumerate(batch_method(chunker.chunks_as_list, 20)):
    new_nodes = prepare_new_nodes(data=bat, playlist_id=playlist_id, embedding_service=embed)
    result+=new_nodes
    print("total percent: ", str(round(((20*idx)+1) / playlist_total, 4)*100)[:4], "%", " batch", idx+1, end="r")

writer.load_nodes(data=result)

다음은 데이터를 로드하는 데 사용되는 Cypher `Query`입니다.

UNWIND $data AS param

MERGE (child:Document {index: param.child_index})
MERGE (parent:Document {index: param.parent_index})
MERGE (s:Source {url: param.url})
SET
  child:Child,
  child.create_time = datetime(),
  child.text = param.transcript,
  child.embedding = param.embedding,
  
  parent:Parent,
  parent.text = param.parent_transcript,
  
  s.title = param.title,
  s.playlist_id = param.playlist_id,
  s.video_id = param.video_id,
  s.publish_date = param.publish_date
  
MERGE (parent)-[:HAS_SOURCE]->(s)
MERGE (child)-[:HAS_PARENT]->(parent)

임베딩에 대한 참고 사항

이 프로젝트에서는 OpenAI 또는 VertexAI와 같은 더 정교한 옵션에 대한 무료 대안으로 SpaCy 임베딩을 사용하고 있어요. 저희 임베딩은 96차원인데, OpenAI와 VertexAI는 각각 1536차원과 768차원 임베딩을 제공하죠. 이로 인해 임베딩의 의미와 컨텍스트 보존이 크게 저하될 수 있어요. 이 프로젝트에서 다른 임베딩 서비스를 사용하려면 `tools/embedding.py`에 있는 `EmbeddingService` 클래스의 `__init__` 메서드와 `Vector` `Index`의 `'Vector.dimensions'` 매개변수를 적절하게 변경하면 됩니다.

def __init__(self) -> None:
    """
    SpaCy embedding service.
    """
    self.embedding_service = SpacyEmbeddings()

그래프

아래 그래프에서 예시 경로를 볼 수 있어요. 보라색 `Node`는 소스 동영상이며 동영상 제목, 게시 날짜 등의 정보를 포함하고, 파란색 `Node`는 `Vector` 검색 시 반환될 상위 텍스트 청크이며, 주황색 `Node`는 일치시킬 임베딩이 포함된 하위 텍스트 청크입니다.

Example path from graph
(:소스)<-[:HAS_SOURCE]-(:부모)<-[:HAS_PARENT]-(:자식)

결론 및 다음 단계

이제 `RAG` 애플리케이션에 사용할 수 있는 간단한 `Knowledge Graph`를 구성했어요. 다음 블로그에서는 이 그래프를 사용하여 기본 `RAG` 애플리케이션을 구축하는 방법을 살펴볼게요. 이 프로젝트의 Github는 여기에서 찾을 수 있습니다: 이 프로젝트는 계속 업데이트될 예정이에요.

  • 웹스크래핑

에이치시스템즈의 LogTree는 Neo4j 기반 GraphRAG 플랫폼으로, 데이터를 자동으로 지식그래프화하고 자연어 질의로 즉시 답을 제공합니다.

👉 에이치시스템즈 홈페이지

728x90
반응형
저작자표시 비영리 변경금지 (새창열림)