Merge pull request #6 from Amore-Innovation/feature/rag-db-v4

feat: 랭그래프 에이전트 상태 관리 및 브랜드 톤 노드 추가

Author: Kwon-hayoung

app/agent/graph.py

@@ -1,9 +1,11 @@
+import operator
 from typing import TypedDict, Optional, Any, List
 from langgraph.graph import StateGraph, END
 from app.agent.nodes.retrieval import retrieve
 from app.agent.nodes.query_transform import transform_query
 from app.agent.nodes.generation import generate
 from app.agent.nodes.grading import grade_message, grade_documents
+from app.agent.nodes.brand_mapping import brand_mapping
 from langchain_core.documents import Document
 
 class AgentState(TypedDict):
@@ -14,6 +16,9 @@ class AgentState(TypedDict):
     trigger_context: Optional[dict]  
     trigger_data: Optional[dict]  # 트리거 데이터 청크
     status: Optional[str]
+    product_contexts: List[dict]
+    review_contexts: List[dict]
+    grade_style_loop_count: int
 
     question: str
     generation: str # 생성된 메시지
@@ -25,61 +30,60 @@ class AgentState(TypedDict):
     grade_style_loop_count: int
     feedback: str
 
-def build_graph():
 
-    # Conditional Edge for Document Grading
-    def decide_document(state):
-        """문서 검색 로직
-        1. 통과 -> 메시지 생성
-        2. 탈락 -> 쿼리 수정 및 문서 재검색
-        3. 탈락 2회 이상 -> 메시지 생성 (무한루프 방지)"""
-
-        grade = state.get("relevance_grade")
-        grade_document_loop_count = state.get("grade_document_loop_count", 0)
+# --- 1. 문서 검색 조건부 로직 (최대 2회) ---
+def decide_document(state):
+    """문서 검색 로직
+    1. 통과 -> 메시지 생성
+    2. 탈락 -> 쿼리 수정 및 문서 재검색
+    3. 탈락 2회 이상 -> 메시지 생성 (무한루프 방지)"""
+
+    grade = state.get("relevance_grade")
+    grade_document_loop_count = state.get("grade_document_loop_count", 0)
+
+    if grade == 'yes':
+        print("검수 통과 -> 메시지 생성으로 넘어갑니다.")
+        return "generate"
 
-        if grade == 'yes':
-            print("검수 통과 -> 메시지 생성으로 넘어갑니다.")
-            return "generate"
+    if grade_document_loop_count >= 2:
+        print("재탐색 한계 도달 -> 문서 재검색 없이 메시지 생성으로 넘어갑니다.")
+        return "generate"
 
-        if grade_document_loop_count >= 2:
-            print("재탐색 한계 도달 -> 문서 재검색 없이 메시지 생성으로 넘어갑니다.")
-            return "generate"
+    print("검수 탈락 -> 쿼리 재작성 및 문서 재검색")
+    return "transform_query"
 
-        print("검수 탈락 -> 쿼리 재작성 및 문서 재검색")
-        return "modify_query"
+# --- 2. 메시지 생성 조건부 로직 (최대 2회) ---
+def decide_generation(state):
+    """메시지 생성 로직
+    1. 통과 -> 최종 메시지 출력
+    2. 탈락 -> 피드백 반영해 메시지 재생성
+    3. 탈락 2회 이상 -> 최종 메시지 출력 (무한루프 방지)"""
 
-    
-    
-    # Conditional Edge for Message Grading
-    def decide_generation(state):
-        """메시지 생성 로직
-        1. 통과 -> 최종 메시지 출력
-        2. 탈락 -> 피드백 반영해 메시지 재생성
-        3. 탈락 2회 이상 -> 최종 메시지 출력 (무한루프 방지)"""
+    grade = state.get("style_grade")
+    grade_style_loop_count = state.get("grade_style_loop_count", 0)
 
-        grade = state.get("style_grade")
-        grade_style_loop_count = state.get("grade_style_loop_count", 0)
+    if grade == 'yes':
+        print("검수 통과 -> 메시지 생성 완료.")
+        return "brand_mapping"
 
-        if grade == 'yes':
-            print("검수 통과 -> 메시지 생성 완료.")
-            return END
+    if grade_style_loop_count >= 2:
+        print("재생성 한계 도달 -> 메시지 재생성 없이 출력.")
+        return "brand_mapping"
 
-        if grade_style_loop_count >= 2:
-            print("재생성 한계 도달 -> 메시지 재생성 없이 출력.")
-            return END
-
-        print("검수 탈락 -> 피드백 반영하여 메시지 재생성.")
-        return "generate"
+    print("검수 탈락 -> 피드백 반영하여 메시지 재생성.")
+    return "generate"
 
 
-
+# --- 3. 그래프 빌더 ---
+def build_graph():
     workflow = StateGraph(AgentState)
 
     workflow.add_node("retrieve", retrieve)
     workflow.add_node("grade_documents", grade_documents) 
     workflow.add_node("transform_query", transform_query)  
     workflow.add_node("generate", generate)
     workflow.add_node("grade_message", grade_message)
+    workflow.add_node("brand_mapping", brand_mapping)
 
     # Set entry point
     workflow.set_entry_point("retrieve")
@@ -88,22 +92,23 @@ def decide_generation(state):
         "grade_documents",
         decide_document,
         {
-            "modify_query": "transform_query",
+            "transform_query": "transform_query",
             "generate": "generate"
         }
     )
     workflow.add_edge("transform_query", "retrieve")
     workflow.add_edge("generate", "grade_message")
+
     workflow.add_conditional_edges(
         "grade_message",
         decide_generation,
         {
             "generate": "generate",
-            END: END
+            "brand_mapping": "brand_mapping"
         }
     )
 
+    workflow.add_edge("brand_mapping", END)
     return workflow.compile()
 
-
-agent_executor = build_graph()
+agent_executor = build_graph()

app/agent/nodes/brand_mapping.py

@@ -0,0 +1,78 @@
+import json
+from typing import Dict, Any
+from langchain_openai import ChatOpenAI
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.output_parsers import StrOutputParser
+from app.core.config import settings
+from app.agent.prompts import (
+    SULWHASOO_PROMPT, ETUDE_PROMPT, AESTURA_PROMPT, 
+    INNISFREE_PROMPT, BREADY_PROMPT
+)
+
+llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
+
+async def brand_mapping(state: Dict[str, Any]):
+    print("===== [Final Step] 브랜드 톤 매핑 =====")
+    
+    # 1. 이전 노드에서 넘어온 데이터 확보
+    current_msg = state.get("generation", "")
+    product_contexts = state.get("product_contexts", [])
+
+    print(f"DEBUG - 넘어온 상품 컨텍스트 개수: {len(product_contexts)}")
+    if product_contexts:
+        print(f"DEBUG - 첫 번째 상품 데이터: {product_contexts[0]}")
+    
+    # 2. Context 분석하여 브랜드 식별
+    sample_product = product_contexts[0] if product_contexts else {}
+    target_text = str(sample_product.get("브랜드", "")) + str(sample_product.get("제품명", ""))
+    
+    # 3. 적절한 프롬프트 매칭
+    selected_prompt = INNISFREE_PROMPT # 기본값
+    brand_key = "COMMON"
+
+    if "설화수" in target_text:
+        selected_prompt = SULWHASOO_PROMPT
+        brand_key = "SULWHASOO"
+    elif "에뛰드" in target_text:
+        selected_prompt = ETUDE_PROMPT
+        brand_key = "ETUDE"
+    elif "에스트라" in target_text:
+        selected_prompt = AESTURA_PROMPT
+        brand_key = "AESTURA"
+    elif "비레디" in target_text:
+        selected_prompt = BREADY_PROMPT
+        brand_key = "B.READY"
+    elif "이니스프리" in target_text:
+        selected_prompt = INNISFREE_PROMPT
+        brand_key = "INNISFREE"
+
+    print(f"CRM 메시지 브랜드: {brand_key}")
+
+    # 4. 브랜드 전용 프롬프트로 최종 보정
+    refine_template = """
+    {brand_guideline}
+    
+    위 브랜드 가이드라인에 맞춰 아래 [초안 메시지]를 최종 수정해줘.
+    메시지의 핵심 내용은 유지하되, 브랜드의 목소리와 규칙(이모지, 어미 등)을 완벽하게 적용해야 해.
+
+    [초안 메시지]:
+    {draft_message}
+    """
+    
+    prompt = ChatPromptTemplate.from_template(refine_template)
+    chain = prompt | llm | StrOutputParser()
+    
+    final_msg = await chain.ainvoke({
+        "brand_guideline": selected_prompt,
+        "draft_message": current_msg
+    })
+
+    print(f"브랜드 톤 적용 완료 ({brand_key})")
+
+    return {
+        "generation": final_msg,
+        "product_contexts": state.get("product_contexts", []),
+        "review_contexts": state.get("review_contexts", []),
+        "persona_data": state.get("persona_data"),
+        "persona_id": state.get("persona_id")
+    }

app/agent/nodes/generation.py

@@ -20,6 +20,8 @@ async def generate(state: Dict[str, Any]):
 
     # retrieval 노드에서 넘겨준 documents 리스트
     documents = state.get("documents", [])
+    p_contexts = state.get("product_contexts", [])
+    r_contexts = state.get("review_contexts", [])
 
     feedback = state.get("feedback", None)
     context = "\n".join([d.page_content for d in documents])
@@ -35,13 +37,17 @@ async def generate(state: Dict[str, Any]):
     chain = prompt | llm | StrOutputParser()
 
     # chain.invoke는 동기 함수지만, 전체 노드는 async로 구성하는 것이 좋습니다.
-    # invoke()가 블로킹되지 않도록 주의하거나 ainvoke 사용 권장
     msg = await chain.ainvoke({
         "user_json": json.dumps(user, ensure_ascii=False, default=str),
         "context": context,
         "feedback_note": feedback_instruction
     })
 
     print(f"\n\n[생성된 메시지]\n{msg}\n=====")
+    print(f"DEBUG [GradeDocs] 최종 필터링된 상품 수: {len(p_contexts)}")
 
-    return {"generation": msg}
+    return {"generation": msg,
+            "product_contexts": state.get("product_contexts", []),
+            "review_contexts": state.get("review_contexts", []),
+            "documents": state.get("documents", [])
+            }