Feat: contents writing agent 구현

.env.example

@@ -14,4 +14,6 @@ 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...
 
-# Others...
+# Others...
+NOTION_PARENT_ID=...
+NOTION_API_KEY=ntn_...

agents/marketing/modules/chains.py

@@ -4,15 +4,19 @@
 기본적으로 modules.prompt 템플릿과 modules.models 모듈을 사용하여 LangChain 체인을 생성합니다.
 """
 
-# from langchain.schema.runnable import RunnablePassthrough, RunnableSerializable
-# from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
+from langchain.chains.summarize import load_summarize_chain
+from langchain.schema.runnable import RunnableSerializable  # , RunnablePassthrough
+from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
+
+from agents.marketing.modules.models import get_openai_model
+from agents.marketing.modules.prompts import (
+    get_write_contents_prompt,
+    notion_page_creation_prompt,
+    summary_doc_prompt,
+)
+
 # from langchain_core.pydantic_v1 import BaseModel, Field
 
-# from agents.marketing.modules.models import get_openai_model
-# from agents.marketing.modules.prompts import (
-#     get_campaign_generation_prompt,
-#     get_content_creation_prompt,
-# )
 
 # 예시 함수들입니다. 참고용으로 남겨둡니다.
 
@@ -95,3 +99,43 @@
 #         | model  # LLM 모델 호출
 #         | StrOutputParser()  # 결과를 문자열로 변환
 #     )
+
+
+def map_reduce_summary_chain() -> RunnableSerializable:
+    """
+    Map Reduce 방식으로 문서를 나누어서 각각을 요약한 뒤 모두 합쳐서 한번 더 요약합니다.
+    """
+    model = get_openai_model()
+    prompt = summary_doc_prompt()
+    chain = load_summarize_chain(
+        llm=model, chain_type="map_reduce", map_prompt=prompt, combine_prompt=prompt
+    )
+    return chain
+
+
+def stuff_summary_chain() -> RunnableSerializable:
+    """
+    문서를 한번에 보고 요약합니다.
+    """
+    model = get_openai_model()
+    prompt = summary_doc_prompt()
+    chain = load_summarize_chain(
+        llm=model,
+        chain_type="stuff",
+        prompt=prompt,
+    )
+    return chain
+
+
+def write_blog_content_chain() -> RunnableSerializable:
+    model = get_openai_model()
+    prompt = get_write_contents_prompt()
+
+    return prompt | model | StrOutputParser()
+
+
+def create_notion_page_chain() -> RunnableSerializable:
+    model = get_openai_model()
+    prompt = notion_page_creation_prompt()
+
+    return prompt | model | JsonOutputParser()

agents/marketing/modules/models.py

@@ -3,23 +3,26 @@
 이 모듈은 LangChain에서 사용할 LLM 모델을 설정하고 반환하는 함수들을 포함합니다.
 """
 
-# from langchain_openai import ChatOpenAI
+import os
 
-# 예시 함수들입니다. 참고용으로 남겨둡니다.
+from langchain_openai import ChatOpenAI
 
-# def get_openai_model() -> ChatOpenAI:
-#     """
-#     OpenAI 모델 인스턴스를 생성하고 반환합니다.
-#
-#     이 함수는 교육 마케팅 관련 작업에 적합한 설정으로
-#     OpenAI의 ChatGPT 모델을 초기화합니다.
-#
-#     Returns:
-#         ChatOpenAI: 초기화된 OpenAI 모델 인스턴스
-#     """
-#     # OpenAI 모델 초기화 및 반환
-#     return ChatOpenAI(
-#         model_name="gpt-4",  # 모델 이름 (gpt-4 사용)
-#         temperature=0.7,  # 창의성 조절 (0.7은 균형 잡힌 창의성)
-#         max_tokens=1500,  # 최대 토큰 수
-#     )
+
+def get_openai_model(
+    model_name: str = "gpt-3.5-turbo", temperature: float = 0.7, max_tokens: int = 4000
+) -> ChatOpenAI:
+    """
+    OpenAI 모델 인스턴스를 생성하고 반환합니다.
+
+    이 함수는 교육 마케팅 관련 작업에 적합한 설정으로
+    OpenAI의 ChatGPT 모델을 초기화합니다.
+
+    Returns:
+        ChatOpenAI: 초기화된 OpenAI 모델 인스턴스
+    """
+    return ChatOpenAI(
+        api_key=os.environ["OPENAI_API_KEY"],
+        model_name=model_name,  # 모델 이름 (gpt-4 사용)
+        temperature=temperature,  # 창의성 조절 (0.7은 균형 잡힌 창의성)
+        max_tokens=max_tokens,  # 최대 토큰 수
+    )

agents/marketing/modules/nodes.py

@@ -6,9 +6,20 @@
 각 노드는 execute 메서드를 구현하여 상태(state)를 입력받아 처리하고, 처리 결과를 새로운 상태 업데이트로 반환합니다.
 """
 
-# from agents.base_node import BaseNode
+import os
 
-# from agents.marketing.modules.chains import set_campaign_generation_chain, set_content_creation_chain
+import requests
+from langchain_core.documents import Document
+
+from agents.base_node import BaseNode
+from agents.marketing.modules.chains import (
+    create_notion_page_chain,
+    map_reduce_summary_chain,
+    stuff_summary_chain,
+    write_blog_content_chain,
+)
+from agents.marketing.modules.state import ContentState
+from agents.marketing.modules.utils import load_pdf_documents
 
 # 아래는 구현 예정인 노드 클래스들입니다. 실제 구현 시 주석을 해제하고 사용하면 됩니다.
 
@@ -86,3 +97,99 @@
 #
 #         # 생성된 마케팅 콘텐츠를 새로운 상태 업데이트로 반환
 #         return {"messages": marketing_content}
+
+
+class DocSummarizationNode(BaseNode):
+    """
+    문서를 요약해주는 노드
+
+    필요에따라 map reduce 방식, stuff 방식 혹은 둘 모두를 사용하여 문서를 요약하여 state에 추가합니다.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.map_reduce_summary_chain = map_reduce_summary_chain()
+        self.stuff_summary_chain = stuff_summary_chain()
+
+    def load_pdf(self, input_file: str) -> list[Document]:
+        return load_pdf_documents(input_file)
+
+    def execute(self, state: ContentState) -> dict:
+        """
+        주어진 상태(state)에 포함된 문서명(str)을 가지고 문서를 열람한 뒤 요약합니다.
+
+        llm의 버전에 따라 다르지만 gpt기준 토큰 수 제한이 있어 map reduce를 먼저 수행한 뒤 stuff로 한번 더 요약합니다.
+
+        Args:
+            state (ContentState): Workflow의 현재 상태. input_file 정보를 포함.
+
+        Returns:
+            dict: 요약된 문서를 포함한 상태 업데이트
+        """
+        # TODO: pdf 외의 다른 입력도 고려하여 분기를 만들지?
+
+        interview_doc = self.load_pdf(state.get("input_file"))
+
+        document_summary = self.map_reduce_summary_chain.invoke(interview_doc)
+        document_summary = self.stuff_summary_chain.invoke(document_summary)
+        return {"document_summary": document_summary}
+
+
+class NotionWritingNode(BaseNode):
+    """
+    노션 컨텐츠 작성 및 업로드 노드
+
+    제공받은 문서(또는 요약)을 바탕으로 Notion에 게시할 블로그 컨텐츠를 작성합니다.
+    작성한 컨텐츠는 Notion API를 사용해 게시합니다.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.write_content_chain = write_blog_content_chain()
+        self.create_page_chain = create_notion_page_chain()
+        self.parent_id = os.environ["NOTION_PARENT_ID"]
+        self.api_key = os.environ["NOTION_API_KEY"]
+
+    def create_page(self, payload: dict) -> str:
+        # TODO: retry 코드 추가 필요
+
+        payload["parent"] = {"page_id": self.parent_id}
+
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+            "Notion-Version": "2022-06-28",
+        }
+
+        res = requests.post(
+            "https://api.notion.com/v1/pages", headers=headers, json=payload
+        )
+
+        if res.status_code == 200:
+            return "success ✅"
+        else:
+            return f"failed ❌: {res.status_code}, {res.text}"
+
+    def execute(self, state: ContentState) -> dict:
+        """
+        주어진 상태(state)의 문서 요약을 바탕으로 블로그 컨텐츠를 작성합니다.
+        이후 작성한 컨텐츠를 json형태로 바꾸어 notion api를 통해 notion에 게시합니다.
+
+        Args:
+            state (ContentState): Workflow의 현재 상태. input_file 정보를 포함.
+
+        Returns:
+            dict: 생성된 블로그 콘텐츠를 포함한 상태 업데이트
+        """
+        page_content = self.write_content_chain.invoke(
+            {"document_summary": state["document_summary"]}
+        )
+        output_json = self.create_page_chain.invoke({"page_content": page_content})
+        # output_str = re.sub(r"```json|```", "", output_str).strip() # TODO: 필요한지 검증
+
+        results = self.create_page(output_json)
+
+        return {
+            "page_content": page_content,
+            "results": results,
+        }

agents/marketing/modules/prompts.py

@@ -4,7 +4,7 @@
 각 함수는 특정 작업에 맞는 프롬프트 템플릿을 생성합니다.
 """
 
-# from langchain_core.prompts import PromptTemplate
+from langchain_core.prompts import PromptTemplate
 
 # 예시 함수들입니다. 참고용으로 남겨둡니다.
 
@@ -101,3 +101,70 @@
 #     ...
 #     """
 #     return PromptTemplate.from_template(template)
+
+
+def summary_doc_prompt() -> PromptTemplate:
+    template = """
+        다음 문서를 확인하고 정리해줘.
+        문서의 핵심 요약과, 주요 질문사항을 포함하여 정리해줘.
+        {text}
+    """
+    return PromptTemplate(template=template, input_variables=["text"])
+
+
+def get_write_contents_prompt() -> PromptTemplate:
+    template = """
+        너는 블로그 글을 쓰는 컨텐츠팀에서 근무하고 있어.
+        아래 내용을 확인하고 ux research에 관련된 블로그 컨텐츠를 작성해줘
+        입력: {document_summary}
+
+        중요 조건:
+        - 다양한 사람들이 이해할 수 있도록 쉽게 써야 해.
+        - 노션에 글을 쓸거야. 마크다운 문법과 아이콘을 활용해서 눈에 잘 띄는 컨텐츠를 만들어줘.
+    """
+    return PromptTemplate.from_template(template)
+
+
+def notion_page_creation_prompt() -> PromptTemplate:
+    template = """
+        너는 노션 페이지를 만들어야 해. 주어진 입력에 대해서 JSON body를 만들어줘.
+        입력: {page_content}
+
+        중요 조건:
+        - 절대 간단한 요약 JSON을 출력하지 마. 반드시 Notion API의 완전한 JSON 구조를 따라야 해.
+        - 결과는 JSON 형태로만 출력해줘. 대신 제일 처음과 끝에 코드 블록(```json)은 절대 붙이지 마.**
+        - 임의로 내용을 요약하지 마. 있는 내용 그대로를 포함하는 JSON을 만들어야 해.
+        - children의 paragraph 속성 아래에 'rich_text'를 반드시 넣어야 해.
+        - 글의 내용은 children 아래에 여러 블록으로 적절하게 나뉘어야 해.
+        - 아래는 JSON 예시야 사용자 입력을 바탕으로 해당 JSON 내용을 채워줘. JSON의 형태는 엄격하게 검증해줘.
+        ```
+        "parent": {{ "page_id": "부모 페이지 ID" }},
+        "properties": {{
+            "title": [
+                {{
+                    "type": "text",
+                    "text": {{
+                    "content": "페이지 제목"
+                    }}
+                }}
+            ]
+        }},
+        "children": [
+            {{
+                "object": "block",
+                "type": "paragraph",
+                "paragraph": {{
+                    "rich_text": [
+                        {{
+                            "type": "text",
+                            "text": {{
+                                "content": "문단 내용"
+                            }}
+                        }}
+                    ]
+                }}
+            }}
+        ]
+        ```
+    """
+    return PromptTemplate.from_template(template)

agents/marketing/modules/state.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Annotated, List, TypedDict
+from typing import Annotated, List, Optional, TypedDict
 
 from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.graph.message import add_messages
@@ -28,3 +28,11 @@ class MarketingState(TypedDict):
     messages: Annotated[List[HumanMessage | AIMessage], add_messages]  # 메시지 목록
     # Annotated는 Python 타입 어노테이션으로, add_messages는 LangGraph에서 사용하는 특별한 함수입니다.
     # add_messages는 메시지 목록에 새로운 메시지를 추가할 때 이전 메시지를 유지하면서 추가하는 기능을 자동으로 처리합니다.
+
+
+@dataclass
+class ContentState(TypedDict):
+    input_file: str
+    document_summary: Optional[str]
+    content: Optional[str]
+    result: Optional[str]

agents/marketing/modules/utils.py

@@ -13,6 +13,37 @@
 # import json
 # from datetime import datetime
 
+from langchain.document_loaders import PyPDFLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_core.documents import Document
+
+
+def load_pdf_documents(
+    file_path: str,
+    chunk_size: int = 5000,
+    chunk_overlap: int = 100,
+) -> list[Document]:
+    """
+    PDF 파일을 로드하고, 지정된 크기로 텍스트를 분할한 후 Document 객체 리스트로 반환합니다.
+
+    RecursiveCharacterTextSplitter를 통해 긴 텍스트를 여러 chunk로 나눕니다.
+    각 chunk는 downstream LLM 처리에 적합한 길이로 유지됩니다.
+
+    Args:
+        file_path (str): 분할할 PDF 파일의 경로.
+        chunk_size (int, optional): 각 chunk의 최대 문자 수.
+        chunk_overlap (int, optional): 인접한 chunk 간 중복 문자 수.
+
+    Returns:
+        List[Document]: 분할된 문서 조각들의 리스트. 각 Document는 page_content와 metadata를 포함합니다.
+    """
+    loader = PyPDFLoader(file_path)
+    documents = loader.load()
+    splitter = RecursiveCharacterTextSplitter(
+        chunk_size=chunk_size, chunk_overlap=chunk_overlap
+    )
+    return splitter.split_documents(documents)
+
 
 # def get_message_text(msg: BaseMessage) -> str:
 #     """메시지의 텍스트 내용을 가져옵니다."""

agents/marketing/pyproject.toml

@@ -9,4 +9,7 @@ version = "0.1.0"
 description = "교육 콘텐츠 마케팅 전략 및 콘텐츠 생성을 위한 LangGraph Workflow 모듈"
 readme = "README.md"
 requires-python = ">=3.13"
-dependencies = []
+dependencies = [
+    "langchain-openai>=0.3.18",
+    "pypdf>=5.5.0",
+]