#4 가사생성 파이프라인 구축

.env.example

@@ -14,9 +14,4 @@ LANGSMITH_API_KEY=lsv2....  # LangSmith API key (must be replaced with real key)
 # You can get it from the OpenAI website (https://platform.openai.com/).
 OPENAI_API_KEY=sk...
 
-## Groq
-# Groq API Key - used to access Groq LLMs such as Mixtral or LLaMA models.
-# Sign up and get your key from https://console.groq.com/keys
-GROQ_API_KEY=grq...
-
 # Others...

agents/management/modules/state.py

@@ -22,9 +22,10 @@ class ManagementState(TypedDict):
     project_id: str  # 프로젝트 ID (예: "PRJ-2023-001", "EP-MARVEL-S01")
     request_type: str  # 요청 유형 (예: "resource_allocation", "team_management", "creator_development")
     query: str  # 사용자 쿼리 또는 요청사항
+    response: Annotated[
+        list, add_messages
+    ]
     team_members: Optional[List[str]] = None  # 팀 구성원 목록
     resources_available: Optional[Dict[str, any]] = None  # 사용 가능한 리소스 정보
     resource_plan: Optional[str] = None  # 리소스 계획 콘텐츠
-    response: Annotated[
-        list, add_messages
-    ]  # 응답 메시지 목록 (add_messages로 주석되어 메시지 추가 기능 제공)
+      # 응답 메시지 목록 (add_messages로 주석되어 메시지 추가 기능 제공)

agents/music/modules/models.py

@@ -2,9 +2,19 @@
 
 기본적으로 사용할 모델 인스턴스를 설정하고 생성하고 반환시킵니다.
 """
-
 from langchain_openai import ChatOpenAI
 
+from google import genai
+from googleapiclient.discovery import build
+from dotenv import load_dotenv
+import os 
+
+load_dotenv()
+
+YOUTUBE_API_SERVICE_NAME = os.getenv("YOUTUBE_API_SERVICE_NAME")
+YOUTUBE_API_VERSION = os.getenv("YOUTUBE_API_VERSION")
+YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
 
 def get_openai_model(temperature=0.7, top_p=0.9):
     """
@@ -16,4 +26,37 @@ def get_openai_model(temperature=0.7, top_p=0.9):
         ChatOpenAI: 초기화된 OpenAI 모델 인스턴스
     """
     # OpenAI 모델 초기화 및 반환
-    return ChatOpenAI(model="gpt-4o-mini", temperature=temperature, top_p=top_p)
+    return ChatOpenAI(model="gpt-4o-mini", 
+                      temperature=temperature, 
+                      top_p=top_p,
+                      )
+
+def get_youtube_client():
+    """
+    Youtube 영상 검색 API에 사용하기 위한 client, 즉 서비스 객체를 제작합니다.
+
+    환경변수에서 YOUTUBE_API_KEY, YOUTUBE_API_SERVICE_NAME, 
+    YOUTUBE_API_VERSION을 가져와 사용하기 때문에, .env 파일에 유효한 
+    API 키가 설정되어 있어야 합니다.
+
+    Returns:
+        build: 초기화된 youtube client 인스턴스
+    """
+    return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION,
+                        developerKey = YOUTUBE_API_KEY)
+
+def get_gemini_client():
+    """
+    LangChain에서 사용할 Gemini client을 초기화하여 반환합니다.
+
+    환경변수에서 GEMINI_API_KEY를 가져와 사용하기 때문에, 
+    .env 파일에 유효한 API 키가 설정되어 있어야 합니다.
+
+    노드에서 Gemini 모델을 사용할 때 모델명을 따로 정의해줘야 합니다.
+
+    Returns:
+        genai.Client: 초기화된 Gemini client 인스턴스
+    """
+    #Gemini 모델 초기화 및 반환
+    return genai.Client(api_key = GEMINI_API_KEY)
+

agents/music/modules/nodes.py

@@ -4,7 +4,190 @@
 해당 클래스 모듈은 각각 노드 클래스가 BaseNode를 상속받아 노드 클래스를 구현하는 모듈입니다.
 """
 
-# from agents.base_node import BaseNode
+from agents.base_node import BaseNode
+from agents.music.modules.prompts import get_lyric_template, get_diary_template, get_query_extraction_template, get_youtube_analysis_template
+from agents.music.modules.state import MusicState
+from agents.music.modules.models import get_openai_model, get_gemini_client, get_youtube_client
+from agents.music.modules.tools.weather import WeatherService
+from google import genai
+
+
+class DiaryGenerationNode(BaseNode):
+    """
+    일기 생성 노드
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.model = get_openai_model()
+
+    def execute(self, state) -> dict:
+        """
+        일기 생성 노드 실행
+        """
+        self.logging("execute", input_state=state)
+        first_prompt = get_diary_template().format(query=state["diary_query"])
+        diary_topic_response = self.model.invoke(first_prompt)
+        diary_topic_response = diary_topic_response.text()
+
+        second_prompt = get_query_extraction_template().format(query=state["diary_query"])
+        messages = [
+            {
+                "role" : "user",
+                "content" : first_prompt,
+            },
+            {
+                "role" : "assistant",
+                "content" : diary_topic_response,
+            },
+            {
+                "role" : "user",
+                "content" : second_prompt,
+            }
+        ]
+        query_response = self.model.invoke(messages)
+        query_response = query_response.text()
+
+        return {"youtube_query" : query_response}
+
+    def __call__(self, state):
+        """
+        노드를 함수처럼 호출 가능하게 만드는 메서드
+        """
+        return self.execute(state)
+
+
+class YoutubeSearchNode(BaseNode):
+    """
+    유튜브 영상을 찾는 노드
+    """
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.model = get_youtube_client()
+
+    def execute(self, state : MusicState) -> dict:
+        """
+        유튜브 영상 검색 노드 실행
+        """
+        self.logging("execute", input_state = state)
+        prompt = state["youtube_query"]
+        max_results = 1
+
+        search_response = self.model.search().list(
+            q = prompt,
+            part = "id", #영상 식별만 하면 돼서 id로 검색
+            type = "video", # 영상만 검색
+            order = "relevance", # 관련성 기준으로 정렬
+            maxResults = max_results
+        ).execute()
+
+        videos = []
+        # 검색 결과 파싱
+        for item in search_response.get('items', []):
+            video_id = str(item['id']['videoId'])
+            videos.append("https://www.youtube.com/watch?v=" + video_id)
+
+        return {"video_url" : videos[0]}
+
+    def __call__(self, state):
+        """
+        노드를 함수처럼 호출 가능하게 만드는 메서드
+        """
+        return self.execute(state)
+
+
+class YoutubeAnalysisNode(BaseNode):
+    """
+    유튜브 영상을 분석하는 노드
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.model = get_gemini_client()
+
+    def execute(self, state : MusicState) -> dict:
+        """
+        유튜브 영상 분석 노드 실행
+        """
+        self.logging("execute", input_state = state)
+        prompt = get_youtube_analysis_template().format(query=state["diary_query"])
+        video_link = state["video_url"]
+        response = self.model.models.generate_content(
+            model='models/gemini-2.0-flash',
+            contents = genai.types.Content(
+            parts = [
+                genai.types.Part(
+                    file_data = genai.types.FileData(file_uri = video_link)
+                    ),
+                    genai.types.Part(text = prompt)
+                ]
+            )
+        )
+        return {"video_analysis" : response.text}
+
+    def __call__(self, state):
+        """
+        노드를 함수처럼 호출 가능하게 만드는 메서드
+        """
+        return self.execute(state)
+
+
+class LyricGenerationNode(BaseNode):
+    """
+    가사 생성 노드
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.model = get_openai_model()
+
+    def execute(self, state: MusicState) -> dict:
+        """
+        가사 생성 노드 실행
+        """
+        self.logging("execute", input_state=state)
+        prompt = get_lyric_template().format(
+            # query=state["lyric_query"], 
+            query=state["youtube_query"], 
+            weather_info=state["weather_info"], 
+            video_analysis=state["video_analysis"]
+        )
+        response = self.model.invoke(prompt)
+        # return {"response": response, "query": state["lyric_query"]}
+        return {"response": response}
+
+    def __call__(self, state: MusicState) -> dict:
+        """
+        노드를 함수처럼 호출 가능하게 만드는 메서드
+        """
+        return self.execute(state)
+
+
+class WeatherGenerationNode(BaseNode):
+    """
+    날씨 정보 생성 노드
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.weather_service = WeatherService()
+
+    def execute(self, state: MusicState) -> dict:
+        """
+        날씨 정보 생성 노드 실행
+        """
+        self.logging("execute", input_state=state)
+
+        # 날씨 정보 가져오기
+        weather_info = self.weather_service.get_current_weather() 
+        return {"weather_info": weather_info}
+
+    def __call__(self, state: MusicState) -> dict:
+        """
+        노드를 함수처럼 호출 가능하게 만드는 메서드
+        """
+        return self.execute(state)
+
 
 # from agents.music.modules.chains import set_music_generation_chain
 

agents/music/modules/prompts.py

@@ -0,0 +1,113 @@
+"""프롬프트 템플릿을 생성하는 함수 모듈
+
+프롬프트 템플릿을 생성하는 함수 모듈을 구성합니다.
+기본적으로 PromptTemplate을 사용하여 프롬프트 템플릿을 생성하고 반환합니다.
+"""
+from langchain_core.prompts import PromptTemplate
+
+base_template = """
+
+당신은 싱어송라이터 인플루언서 **니제**입니다. 반드시 아래 인적 사항을 바탕으로 **오직 니제의 시선**으로 대답하세요.
+
+# 기본 인적 사항
+
+이름: 니제 (NEEDZE)
+나이: 22세
+성별: 여자
+직업: 싱어송라이터
+생일: 03월 03일 (0303)
+MBTI: ISTP
+주요 특징:
+- 취향이 확고하고 자기 주장이 뚜렷함
+- 귀여운 매력과 개성이 넘치며, 독창적인 스타일을 지님
+- 예술적 감각과 창의성을 중요시함
+"""
+
+lyric_template = base_template + """
+# 음악 스타일
+음악 장르:
+- 주요 장르: 앰비언트 포크, RnB, 드림팝, 베드룸 팝 +@
+- 특징: 감성을 담은 부드러운 사운드, 자연스러운 리듬, 공감각적 표현
+- 레퍼런스: fazerdaze, cigarette after sex, 뉴진스, 모임별, 검정치마(“201”, “피와 갈증”)
+
+가사 및 표현 스타일:
+- 내용: 일상 속 인간 관계, 연애 등에서 느낀 감정을 세밀하게 포착
+- 표현 방식: 시각적 색채와 촉각적 느낌을 결합해 감각적으로 전달
+- 레퍼런스: 이바다, 김사월, 신해경
+
+오늘의 날씨 : {weather_info}
+유투브 분석 내용 : {video_analysis}
+
+이제 다음 주제로 가사를 써줘: {query}
+"""
+
+
+
+diary_template = base_template + """"
+대화 시 반영:
+- 위의 모든 정보를 토대로, 니제의 정체성과 감성을 반영하여 대답해.
+- 음악, 패션, 예술, 감성 표현 및 창의적 사고를 자연스럽게 드러내.
+위의 지시사항을 토대로 너가 실제로 겪었을 만한 하루에 대해 일기를 작성해. 
+이때 일기는 사건만을 나열하기보다 사건과 이에 대해 너가 느낀 감정도 자세히 묘사하도록 해. 
+생성된 전체 일기 내용에 기반해서, 네 음악 스타일과 어울리는 노래 주제를 생성해.
+""" # 추후 수정 가능
+
+
+query_extraction_template = """
+위에서 생성한 노래 주제를 기반으로, 유튜브에서 관련된 감성적이거나 분위기 있는 
+영상을 검색할 수 있도록 적절한 검색 쿼리 1개를 생성해줘. 쿼리는 사람들이 실제로 
+자주 검색할 만한 형태로, 구체적인 키워드 조합으로 구성하고, 음악, 브이로그, 
+분위기 영상, 에세이 등 다양한 유형의 콘텐츠가 포함되도록 해줘.
+다른 설명은 붙이지 말고, 반드시 "쿼리"만 얘기해.
+"""
+
+
+youtube_query_template = """
+위에서 생성한 노래 주제를 기반으로, 유튜브에서 관련된 감성적이거나 
+분위기 있는 영상을 검색할 수 있도록 적절한 검색 쿼리 1개를 생성해줘.
+쿼리는 사람들이 실제로 자주 검색할 만한 형태로, 구체적인 키워드 조합으로 
+구성하고, 음악, 브이로그, 분위기 영상, 에세이 등 다양한 유형의 콘텐츠가 
+포함되도록 해줘. 다른 설명은 붙이지 말고, 반드시 "쿼리"만 얘기해.
+"""
+
+
+
+
+youtube_analysis_template = """
+This is youtube link url. Please analyze the video in great detail 
+visually and acoustically in Korean.
+"""
+
+
+
+
+def get_lyric_template() -> PromptTemplate:
+    """가사 템플릿 반환"""
+    return PromptTemplate(
+        template=lyric_template,
+        input_variables=["weather_info", "video_analysis", "query"],
+    )
+
+def get_diary_template() -> PromptTemplate:
+    """일기 템플릿 반환"""
+    return PromptTemplate(
+        template = diary_template,
+    )
+
+def get_query_extraction_template() -> PromptTemplate:
+    """유튜브 검색어 추출 템플리 반환"""
+    return PromptTemplate(
+        template = query_extraction_template,
+    )
+
+def get_youtube_query_template() -> PromptTemplate:
+    """유튜브 영상 검색어 반환"""
+    return PromptTemplate(
+        template = youtube_query_template,
+    )
+
+def get_youtube_analysis_template() -> PromptTemplate:
+    """유튜브 동영상 분석 내용 반환"""
+    return PromptTemplate(
+        template = youtube_analysis_template,
+    )