[주얼리 브랜드에 대한 소비자 반응 분석] LLM RAG(검색, 증강, 생성) 코드로 이해하기 ①
baektree2025. 10. 20. 22:26
LLM RAG 로직을 생성한 코드를 가져왔다.
코드 블록으로 단계별 이해를 해보려 한다.
단, 코드가 하도 길어 다음 흐름도 기준 1번부터 3번까지만 현재 포스팅으로 담아보았다.
4 ~ 6번은 다음시간에!
[코드 블록별 구조화]
관련 라이브러리 import
초기 설정 및 데이터 로드
RAG 관련 함수 ------ 여기까지만 ----------
구조화된 답변을 출력하기 위한 pydantic 모델 정의
비동기 분석 및 검증 로직
메인 실행 블록
1. 관련 라이브러리 import
import os
import asyncio
from dotenv import load_dotenv
import pandas as pd
from typing import List
# LangChain 및 Pydantic 관련 라이브러리
from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import InMemoryVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
# 기타 라이브러리
import numpy as np
from tqdm.asyncio import tqdm
pydantic: 데이터 유효성 검사 및 설정 관리를 위한 라이브러리, 모델(BaseModel)을 정의하여 LLM 출력이나 입력 데이터가정해진 구조(ExtractionResult)와 타입(Field)을 따르도록 강제하여, 프로그램이 안정적으로 데이터를 파싱할 수 있게 함.
langchain_google_genai: LangChain 프레임 워크에서 google의 생성형 ai 모델을 사용하기 위한 통합 모듈
ChatGoogleGenerativeAI: 대화(채팅)형 LLM 모델을 호출하여 텍스트 생성 및 분석을 요청
GoogleGenerativeAIEmbeddings: 텍스트를벡터로 변환하여 RAG 시스템의 의미 기반 검색에사용
langchain_community : LangChain에서 제공하는 다양한 외부 도구(벡터 저장소, 로더 등)를 포함하는 모듈
InMemoryVectorStore: 데이터를 메모리에 임시로 저장하는 벡터 데이터베이스(Vector Store).RAG 검색을 위해 임베딩된 문서를 저장하고 검색 쿼리와의 유사도를 비교.
Document: LangChain 내에서 처리되는표준 데이터 단위입니다. 원본 내용(page_content)과 메타데이터(metadata)를 담는 구조체.
JsonOutputParser: LLM의 출력(보통 문자열 형태)을 정의된JSON 구조(Pydantic 모델)에 맞춰 파싱하고, 파싱 실패 시 오류를 처리.
2. 초기 설정 및 데이터 로드
# --- 초기 설정 ---
load_dotenv()
api_key = os.getenv("GOOGLE_API_KEY")
# 파일 경로
product_name_csv_path =
output_path =
CONCURRENT_REQUESTS = 10
# --- 데이터 로드 ---
try:
product_name_df = pd.read_csv(product_name_csv_path, dtype=str).fillna('')
except FileNotFoundError:
print(f"경고: '{product_name_csv_path}' 에서 제품 파일을 찾을 수 없습니다.")
exit()
load_dotenv() : .env 파일에서 GOOGLE_API_KEY를 로드
CONCURRENT_REQUESTS : 동시 요청수 즉 병렬 처리의 최대 한도를 설정하여 API 요청이 서버나 네트워크에 과부하 주지 않도록 제어
except FileNotFoundError: 제품 파일을 찾지 못한 경우, 프로그램이 정상적으로 종료되도록 예외 처리
3. RAG(검색 증강 생성) 관련 함수
LLM이 게시글 분석 시 참고할 수 있는 브랜드/라인 찾기
def make_data_to_documents(df_data: pd.DataFrame) -> List[Document]:
return [Document(page_content=f"브랜드: {item.get('brand_k', '')}, 라인: {item.get('product_k', '')}, 브랜드_별칭: {item.get('brand_slang', '')}, 라인_별칭: {item.get('product_slang', '')}", metadata={"brand_k": item.get('brand_k', ''), "product_k": item.get('product_k', '')}) for item in df_data.to_dict('records')]
class KeywordSearch:
def __init__(self, documents: List[Document]):
self.documents = documents
self.doc_contents = [doc.page_content.lower() for doc in documents]
def search(self, query: str, k: int = 10):
query_keywords = set(query.lower().split())
scores = [(doc, float(sum(1 for kw in query_keywords if kw in content))) for doc, content in zip(self.documents, self.doc_contents)]
scores = [s for s in scores if s[1] > 0]
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:k]
def create_vector_store(documents: List[Document], embeddings: GoogleGenerativeAIEmbeddings) -> InMemoryVectorStore:
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=documents)
return vector_store
def hybrid_search(query: str, vector_store: InMemoryVectorStore, keyword_searcher: KeywordSearch, k: int = 10):
semantic_results = vector_store.similarity_search_with_score(query, k=k*2)
keyword_results = keyword_searcher.search(query, k=k*2)
def normalize_scores(results):
if not results: return {}
scores = np.array([score for _, score in results])
min_s, max_s = scores.min(), scores.max()
if max_s == min_s: return {doc.page_content: 0.5 for doc, _ in results}
return {doc.page_content: (score - min_s) / (max_s - min_s) for doc, score in results}
semantic_norm = normalize_scores(semantic_results)
keyword_norm = normalize_scores(keyword_results)
all_docs = {doc.page_content: doc for doc, _ in semantic_results + keyword_results}
hybrid_scores = {content: (semantic_norm.get(content, 0) * 0.2) + (keyword_norm.get(content, 0) * 0.8) for content in all_docs}
sorted_content = sorted(hybrid_scores.items(), key=lambda item: item[1], reverse=True)
return [all_docs[content] for content, _ in sorted_content[:k]]
class KeywordSearch:
def __init__(self, documents: List[Document]):
self.documents = documents
self.doc_contents = [doc.page_content.lower() for doc in documents]
def search(self, query: str, k: int = 10):
query_keywords = set(query.lower().split())
scores = [(doc, float(sum(1 for kw in query_keywords if kw in content))) for doc, content in zip(self.documents, self.doc_contents)]
scores = [s for s in scores if s[1] > 0]
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:k]
2-1) 인스턴스를 땅하고 찍으면 __ init __ (생성자 함수) 실행
인스턴스 속성 documents 를 생성 → 문서 객체 리스트, 구분용 [ doc1…. doc3 ] , 실제 저장은 document 그대로 저장됨.
인스턴스 속성 contents 를 생성 → page_content를 소문자로 만든 리스트 ['...라인: 러브...별칭: 러브링, 러브팔찌...']
2-2) 인스턴스 메서드(기능)를 통해 키워드 기반 점수화 → 상위 k개만 출력
쿼리를 소문자화, 공백으로 구분해서 쪼개기 ex {’까르띠에는', '러브', '팔찌'} ← 검색 쿼리: “까르띠에는 러브 팔찌..”
