[주얼리 브랜드에 대한 소비자 반응 분석] LLM RAG(검색, 증강, 생성) 코드로 이해하기 ②

지난 시간 코드로 RAG 과정을 이해하기 포스팅을 업로드했다.

 

[주얼리 브랜드에 대한 소비자 반응 분석] LLM RAG(검색, 증강, 생성) 코드로 이해하기 ①

 

다음 단계 중 1~3단계를 1편에 업로드했는데 연이어 4~6번도 진행해 마무리 짓고자 한다.

 

[코드 블록별 구조화] 관련 라이브러리 import 초기 설정 및 데이터 로드 RAG 관련 함수 -----여기서부터 진행됨------ 구조화된 답변을 출력하기 위한 pydantic 모델 정의 비동기 분석 및 검증 로직 메인 실행 블록

 

---

4. 구조화된 답변을 출력하기 위한 Pydantic 모델 정의

이 부분은 LLM에 답안지 양식을 주는 단계로 LLM이 원하는 형태로만 딱 맞춰서 답안을 출력하도록 한다.

Pydantic이란 파이썬 타입 힌트를 사용하여 데이터의 유효성을 검증하고 파싱하는 라이브러리이다.

쉽게 말해 “너 내가 정의해놓은 브랜드 목록과 라인 목록중에서만 출력해!” 라고 강제하는 것이라 할 수 있겠다.

# --- 1. LLM의 구조화된 답변을 위한 모델 정의 ---
class ExtractionResult(BaseModel):
    """LLM이 추출한 브랜드/라인 정보와 그 근거를 담는 데이터 모델"""
    brand: List[str] = Field(description="추출된 '브랜드' 형식의 문자열 리스트.", default=[])
    line: List[str] = Field(description="추출된 '라인' 형식의 문자열 리스트.", default=[])
    brand_lines: List[str] = Field(description="추출된 '브랜드_라인' 형식의 문자열 리스트.", default=[])
    reason: str = Field(description="게시글의 어떤 단어나 문맥을 근거로 추출했는지에 대한 간결한 요약.")

• class ExtractionResult(BaseModel): 클래스를 만드는데, BaseModel을 상속 받음.
  • Basemodel은 Pydantic 라이브러리에서 제공하는 기능으로 지정한 타입의 데이터가 입력되었는지 검사하고 AI가 이해할 수 있는 JSON형식으로 변화해주는 도구이다.
• brand: List[str] = Field(description="추출된 '브랜드' 형식의 문자열 리스트.", default=[])
  • brand는 문자열로 된 리스트로 출력하라는 의미이다.

 

5. 비동기 분석 및 검증 로직

1. 엑셀의 행 하나를 가져와서 텍스트(제목+본문)을 합친다.
2. RAG 검색 : 게시글 내용을 바탕으로 참고할만한 문서들을 Vector_store에서 찾아온다. (”이 글에서 언급된 제품 정보 찾아”)
3. LLM 분석 : 게시글과 참고 문서를 LLM에게 던져주면 (”이 글을 읽고 브랜드랑 라인명좀 뽑아줘”)
4. 검증 및 후처리 : LLM이 내놓은 답이 진짜 정확한 브랜드/라인 명칭인지 확인한다. (”brand_line” 형식으로 예쁘게 다듬어서 출력해”) 

# --- 비동기 처리를 위한 개별 행 분석 함수  ---
async def process_row(row, text_chain, vector_store, keyword_searcher, semaphore,
                      valid_brands_set, valid_lines_set, valid_brand_lines_set):
    """텍스트로 브랜드/라인을 추출합니다."""
    async with semaphore:
        new_row_data = row.to_dict()
        test_post = f"{row.get('title', '')} {row.get('cleaned_content', '')}"
        
        final_brands = set()
        final_lines = set()
        final_brand_lines = set()
        final_reason = "내용 없음"

        try:
            # --- 1단계: RAG 검색 ---
            retrieved_docs = hybrid_search(query=test_post, vector_store=vector_store, keyword_searcher=keyword_searcher, k=10)
            reference_list_str = "\\n".join([f"- {doc.page_content}" for doc in retrieved_docs])

            # --- 2단계: 텍스트 기반 분석 ---
            if pd.notna(test_post) and test_post.strip():
                text_output = await text_chain.ainvoke({"post_content": test_post, "reference_list": reference_list_str})
                final_reason = text_output.get('reason', 'N/A')
            else:
                text_output = {}

            # --- 공통 후처리 함수 ---
            def process_and_validate(llm_output, brands_set, lines_set, brand_lines_set):
                # brand_lines 처리
                for item in llm_output.get('brand_lines', []):
                    brand, line = None, None
                    if isinstance(item, str):
                        parts = item.split('_', 1)
                        if len(parts) == 2: brand, line = parts
                    elif isinstance(item, dict):
                        brand, line = item.get('brand'), item.get('line')
                    
                    if brand and line:
                        brand_line_str = f"{brand}_{line}"
                        if (brand in valid_brands_set and line in valid_lines_set and brand_line_str in valid_brand_lines_set):
                            brands_set.add(brand)
                            lines_set.add(line)
                            brand_lines_set.add(brand_line_str)
                # 단독 brand 처리
                for brand in llm_output.get('brand', []):
                    if isinstance(brand, str) and brand in valid_brands_set:
                        brands_set.add(brand)
            
            # 텍스트 분석 결과 검증
            process_and_validate(text_output, final_brands, final_lines, final_brand_lines)

            new_row_data.update({
                'brand': sorted(list(final_brands)),
                'line': sorted(list(final_lines)),
                'brand_line': sorted(list(final_brand_lines)),
                'reason': final_reason
            })

        except Exception as e:
            new_row_data.update({'brand': ['ERROR'], 'line': [], 'brand_line': [], 'reason': str(e)})
        return new_row_data

• semaphore : API 요청이 많을 시 오류를 방지하기 위한 코드 , 동시 실행수를 물리적으로 제한하여 과부하를 막는 장치

 

6. 메인 실행 블록

1. 초기화 : 구글 api 확인 후, RAG 시스템(검색 도구) LLM 모델을 준비하고
2. 데이터 로딩 : 분석할 파일 불러오고, 정답지 만들어 달라 한다.
3. 프롬프트 설정: LLM에게 줄 지시사항을 아주 상세히 설정했다. (자꾸 예외상황이 생겨 규칙이 하나 둘 씩 추가되어 다음과 같이 길어져버렸따.)
4. 비동기 작업 지시: 병렬 처리로 속도가 매우 빨라진다.
5. 파일저장 : 작업이 끝나면 파일 저장한다. 

# --- 비동기 메인 함수 ---
async def main():
    """전체 분석 프로세스를 비동기적으로 실행합니다."""
    if not api_key:
        print("경고: GOOGLE_API_KEY 환경 변수가 설정되지 않았거나 비어 있습니다.")
        exit()

    print("RAG 시스템 초기화 중...")
    documents_for_rag = make_data_to_documents(product_name_df)
    embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
    vector_store = create_vector_store(documents_for_rag, embeddings)
    keyword_searcher = KeywordSearch(documents_for_rag)
    
    print("LLM 초기화 중...")
    
    llm_model = ChatGoogleGenerativeAI(model="gemini-1.5-flash-latest", temperature=0.0) 
    print("초기화 완료.")

    valid_brands_set = set(product_name_df['brand_k'].unique())
    valid_lines_set = set(product_name_df['product_k'].unique())
    valid_brand_lines_set = set((product_name_df['brand_k'] + '_' + product_name_df['product_k']).dropna())

    parser = JsonOutputParser(pydantic_object=ExtractionResult)
    
    # 텍스트 분석용 프롬프트 
    text_prompt_template = """
    너는 '게시글' 본문을 분석하여, 글 안에서 '명시적으로 언급된' 브랜드와 라인 이름만 추출하는 AI이다.

    **[가장 중요한 규칙]**
    1. 너의 유일한 정보 소스는 '게시글' 본문이다. '참고 목록'은 보조 자료일 뿐이다.
    2. '게시글'에 특정 단어가 없으면, '참고 목록'에 그 내용이 존재하더라도 절대 추출해서는 안 된다.
    3. 예를 들어, 게시글에 '롤렉스'만 있고 '오메가'는 없다면, 참고 목록에 '오메가'가 있더라도 절대 '오메가'를 추출하면 안 된다. 롤렉스만 추출한다. 이것이 너의 최우선 규칙이다.
    4. '게시글'에 특정 단어가 있다면, '참고 목록'에 그 내용을 찾아 추출해야한다.
    5. 예를 들어, 게시물에 '씨드'가 있다면 '참고 목록'에서 찾아 '롤렉스_씨-드웰러를 추출해야한다.

    [추출 절차]
    1. '게시글'을 처음부터 끝까지 읽고 브랜드 또는 라인으로 보이는 모든 단어와 문맥을 찾는다.
    2. '게시글'에서 찾은 단어들을 '참고 목록'과 비교하여, 정확한 브랜드/라인 명칭을 확인하고 `brand_line` 형식으로 구성한다.
    3. '게시글'의 어떤 단어나 문맥을 근거로 추출했는지 `reason`에 명확하게 요약한다.
    4. 게시글에서 어떤 브랜드나 라인도 찾을 수 없다면, 모든 목록을 빈 리스트 `[]`로, `reason`은 '언급된 브랜드/라인 없음'으로 답변한다.

    [출력 지침]
    - 브랜드와 라인이 모두 언급되어 `brand_line`를 구성할 수 있으면 `brand`, `line`, `brand_line` 필드를 모두 채운다.
    - 만약 브랜드만 언급되고 특정 라인이 언급되지 않았다면, `brand` 리스트에만 해당 브랜드 이름을 추가한다. (예: '롤렉스 시계' -> `brand`: ['롤렉스'])
    - 반드시 다음 JSON 형식에 맞춰서 답변해야 한다: {format_instructions}

    [특별 규칙]
    1. 롤렉스 라인 판별:
    - '롤렉스 관련된 게시글'에서 '레이디-데이트저스트'를 지칭하는 키워드('28', '26', '플루쥬빌', '여자', '여자 데이져스트')가 발견되면, '레이디-데이트저스트'를 최종 결과로 출력해야 한다.
    - '롤렉스 관련된 게시글'에서 '데이트저스트'를 지칭하는 키워드 ('31')이 나오면 '데이트저스트'를 최종 결과로 출력해야 한다.
    - '롤렉스 관련되 게시글'에서 '티파니', '오이스터', '펩시', '롤', '시계', '콤비', '텐포인트', '데잇져스트' 중 2개 이상 같이 나오면 '티파니앤코'를 절대 출력하지마.

    2. 반클리프 아펠 라인 판별:
    - '반클리프 아펠' 브랜드 라인을 지칭하는 단어('기요세', '무당벌레', '칼세도니', '오닉스', '텐모티브', '모팁', '모티브', '마더오브펄', '마오펄', '빈티지') 중에 단어 1개라도 포함되어 있다면 '후보 목록'에서 찾은 '브랜드_라인'을 그대로 출력한다.

    3. 모호성 해결: '노트' 키워드 판별법:
    - '노트'라는 키워드를 분석할 때는 다음 규칙을 최우선으로 적용하여, '티파니앤코_노트' 라인인지 일반 '노트북(컴퓨터)'인지 반드시 구별해야 한다.
    - '노트북(컴퓨터)'으로 판단하여 제외하는 경우: '노트' 키워드 주변에 아래와 같은 [전자기기 관련 단서](삼성, LG, 애플, 맥북, 그램, 갤럭시북, 스펙, 사양, 피씨방, 핸드폰, 휴대폰)가 하나라도 나타나면, 절대 '티파니앤코_노트'로 분석해서는 안 된다.
    - 그 외는 '티파니앤코_노트'로 출력한다.

    [실제 작업]
    게시글:{post_content}
    참고 목록:\\n{reference_list}
    추출 결과:
    """
    text_prompt = ChatPromptTemplate.from_template(
        text_prompt_template,
        partial_variables={"format_instructions": parser.get_format_instructions()}
    )
    text_chain = text_prompt | llm_model | parser
    
    try:
        if input_path.endswith('.csv'):
            df = pd.read_csv(input_path)
        elif input_path.endswith('.xlsx'):
            df = pd.read_excel(input_path)
        else:
            raise ValueError("지원하지 않는 파일 형식입니다.")
    except FileNotFoundError:
        print(f"경고: '{input_path}' 에서 입력 파일을 찾을 수 없습니다.")
        exit()
        
    semaphore = asyncio.Semaphore(CONCURRENT_REQUESTS)
    
    # 비동기 작업 생성 시, 'vision_model' 인자 제거
    tasks = [process_row(row, text_chain, vector_store, keyword_searcher, semaphore, 
                         valid_brands_set, valid_lines_set, valid_brand_lines_set) 
             for _, row in df.iterrows()]
    
    # 분석 설명도 텍스트 분석에 맞춰 수정
    output_rows = await tqdm.gather(*tasks, desc="게시글 텍스트 분석 중") 
    
    output_df = pd.DataFrame(output_rows)

    print("\\n--- 분석 결과가 추가된 최종 DataFrame ---")
    pd.set_option('display.max_columns', None) 
    print(output_df[['cleaned_content', 'image_url', 'brand', 'line', 'reason']].head())
    
    output_df.to_csv(output_path, index=False, encoding='utf-8-sig')
    print(f"\\n✅ 최종 결과가 '{output_path}' 파일로 저장되었습니다.")

if __name__ == "__main__":
    asyncio.run(main())

• llm_model = ChatGoogleGenerativeAI(model="gemini-1.5-flash-latest", temperature=0.0) : LLM 로드해서 gemini-1.5-flash 모델을 사용. temperature = 0.0으로 설정해서 창의성을 죽이고 정확도에 집중

---

(추가) 파싱이란?

• 일련의 문자열 데이터를 문법적 규칙에 따라 분석하여, 컴퓨터가 실행 및 활용 가능한 자료구조로 변환하는 과정이다.
• 본 프로젝트에서는 LLM이 생성한 JSON 형식의 문자열을 파이썬이 인식할 수 있는 딕셔너리 객체로 변환하는 작업을 의미한다.
• LLM은 근본적으로 비정형 텍스트만을 출력할 수 있기 때문에, 데이터 분석을 위해서는 연산이 가능한 구조화된 데이터 객체로 변환해 주는 파싱 과정이 반드시 필요하다.
• 예시를 통해 자세히 알아보자.
  • 파싱 전
    LLM에게 JSON 형태로 출력해달라고 요청한 상황

    • # LLM이 뱉은 날것의 데이터 (Type: str)
      raw_text = '{"brand": "Rolex", "line": "Submariner", "price": 1500}'
      
      # ❌ 불가능한 작업 (에러 발생)
      # 글자에서 'brand'를 찾으라니 컴퓨터가 이해 못 함
      print(raw_text["brand"])  
      # -> TypeError: string indices must be integers
      

  • 파싱 후
    JSON 문자열 구조를 해석해 파이썬 딕셔너리 형태로 변환

    • # 파싱을 거쳐 다시 태어난 데이터 (Type: dict)
      parsed_data = {"brand": "Rolex", "line": "Submariner", "price": 1500}
      
      # ✅ 가능한 작업 (성공)
      # "브랜드 이름만 가져와!"라고 명령 가능
      print(parsed_data["brand"]) 
      # -> 출력 결과: Rolex
      

  • 프로젝트 당시 파싱이란 개념이 생소해서 단순한 변환 작업 정도로만 이해하고 넘어갔는데, 오늘에서야 그 기술적 의미를 정립할 수 있었다.
    1. LLM의 한계: LLM은 기본적으로 모든 아웃풋을 ‘문자열(String)’형태로만 출력한다.
    2. 따라서 데이터 분석에 활용가능한 자료구조로 바꾸기 위해선 LLM에게 JSON 모양으로만 대답하게 강제하는 과정이 필요하다.
    3. 굳이 JSON형태로 강제하는 이유는 JSON이 파싱하기 적합한 형태이기 때문이다.
    4. JSON은 파이썬의 딕셔너리와 생김새가 비슷하지만, 실제로는 자료구조가 아닌 경량의 텍스트 기반 데이터 교환 형식일 뿐이다.
    5. 따라서 LLM이 출력할 수 있는 형태인 JSON으로 값을 받아 이를 다시 분석에 활용가능한 형태의 딕셔너리로 변환하는 것이 파싱이라고 할 수 있겠다.