그날, 메모하기가 귀찮은 나는 카카오톡에 URL만 쌓아두었는데, Langchain을 만나 슬랙과 노션으로 자동 정리하는 마법을 터득했다!

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
import json
from slack_sdk import WebClient
import os
from langchain_openai import ChatOpenAI

import sys
sys.path.append('../')
from url_agent import URLAgent

slack_token = os.getenv('SLACK_BOT_TOKEN')
client = WebClient(token=slack_token)

# 전역 변수로 처리된 이벤트 ID 저장
processed_events = set()

@csrf_exempt
@require_POST
def slack_events(request):
    json_data = json.loads(request.body)
    print(json_data)

    # 'challenge' 요청 처리
    if 'challenge' in json_data:
        return JsonResponse({'challenge': json_data['challenge']})
    
    event_id = json_data.get('event_id', '')
    # 중복된 이벤트인지 확인
    if event_id in processed_events:
        return JsonResponse({'status': 'Already processed'})

    processed_events.add(event_id)
    
    text = json_data.get('event', {}).get('text', '')

    # ChatOpenAI 인스턴스 생성 및 URLAgent 처리
    key = os.getenv('OPENAI_API_KEY')
    llm = ChatOpenAI(api_key=key, temperature=0)
    URLAgent(llm, text)

    return JsonResponse({'status': 'OK'})

GPT의 challenge 설명
Slack API에서 "challenge"는 주로 이벤트 구독 시 사용되는 보안 메커니즘의 일부입니다. 이는 Slack이 앱의 이벤트 URL을 검증하기 위해 사용합니다. 여기서의 목적은 악의적인 사용자나 의도치 않은 소스로부터 오는 요청을 차단하여, 오직 실제 Slack 서버로부터의 요청만이 처리되도록 하는 것입니다.

    # 'challenge' 요청 처리
    if 'challenge' in json_data:
        return JsonResponse({'challenge': json_data['challenge']})

# 전역 변수로 처리된 이벤트 ID 저장
processed_events = set()

@csrf_exempt
@require_POST
def slack_events(request):
...
    event_id = json_data.get('event_id', '')
    # 중복된 이벤트인지 확인
    if event_id in processed_events:
        return JsonResponse({'status': 'Already processed'})
 processed_events.add(event_id)
...

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages 
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser

from langchain.agents import tool
from langchain.schema.agent import AgentFinish
from langchain.agents import AgentExecutor

from image_caption import load_image_caption
from youtube_loader import load_youtube
from web_loader import load_web
from summarize_documents import summarize
from save_to_notion import save_to_notion

from langchain_openai import ChatOpenAI
import os
import dotenv
dotenv.load_dotenv(verbose=True)

key = os.getenv('OPENAI_API_KEY')

llm = ChatOpenAI(api_key=key, temperature=0)

def summarize_to_notion(url, documents, type):
    summaries = summarize(llm, documents)
    save_to_notion(url, summaries, type)

@tool
def summarize_web(url):
    """Describes a web page"""
    documents = load_web(url)
    summarize_to_notion(url, documents, 'Web')
    return AgentFinish(log="summarize_web", return_values={"output": "작업 완료!"})
    
@tool
def summarize_youtube(url):
    """Describes a youtube video"""
    documents = load_youtube(url)
    summarize_to_notion(url, documents, 'Youtube')
    return AgentFinish(log="summarize_youtube", return_values={"output": "작업 완료!"})
    
@tool
def summarize_image(url):
    """Describes an image"""
    documents = load_image_caption(url)
    summarize_to_notion(url, documents, 'Image')
    return AgentFinish(log="summarize_image", return_values={"output": "작업 완료!"})

tools = [summarize_web, summarize_youtube, summarize_image]

def URLAgent(llm, input):
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You need to look at the URL and invoke the appropriate tool.",
            ),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="URL_judgment"),
        ]
    )

    llm_with_tools = llm.bind_tools(tools)

    agent = (
        {
            "input": lambda x: x["input"],
            "URL_judgment": lambda x: format_to_openai_tool_messages(
                x["intermediate_steps"]
            ),
        }
        | prompt
        | llm_with_tools
        | OpenAIToolsAgentOutputParser()
    )

    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    list(agent_executor.stream({"input": input}))

prompt = ChatPromptTemplate.from_messages(...)

tools = [summarize_web, summarize_youtube, summarize_image]
...
llm_with_tools = llm.bind_tools(tools)

    agent = (
        {
            "input": lambda x: x["input"],
            "URL_judgment": lambda x: format_to_openai_tool_messages(
                x["intermediate_steps"]
            ),
        }
        | prompt
        | llm_with_tools
        | OpenAIToolsAgentOutputParser()
    )
...
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
list(agent_executor.stream({"input": input}))

from langchain_community.document_loaders import WebBaseLoader

def load_web(url):
    loader = WebBaseLoader(url, autoset_encoding=True)
    documents = loader.load()
    return documents

from langchain_community.document_loaders import YoutubeLoader

def load_youtube(url):
    urls = url.split("=")[-1]
    loader = YoutubeLoader(urls, language=["ko", "en"])  # 'ko'가 우선, 없을 경우 'en' 사용
    documents = loader.load()
...

rom langchain_community.document_loaders.generic import GenericLoader
from langchain_community.document_loaders.parsers import OpenAIWhisperParser

def load_youtube_audio(url):
    urls = [url]
    save_dir = "~/data/youtube_audio"
    loader = GenericLoader(YoutubeAudioLoader(urls, save_dir), OpenAIWhisperParser())
    documents = loader.load()
    return documents

from langchain_community.document_loaders import ImageCaptionLoader

import requests
import datetime

def download_image(image_url):
    ...

def load_image_caption(image_url):
    loader = ImageCaptionLoader(image_url)
    documents = loader.load()
    return documents

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_core.documents.base import Document

import os
os.getenv('OPENAI_API_KEY')

import requests
import datetime
import base64

def download_image(image_url):
   ...

def encode_image(image_path):
    """Getting the base64 string"""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode("utf-8")

def load_image_caption(url):
    """Make image summary"""
    file_name = download_image(url)
    
    base64_string = encode_image(file_name)
    chat = ChatOpenAI(model="gpt-4-vision-preview", max_tokens=1024)

    try:
        chat_completion = chat.invoke(
            [
                HumanMessage(
                    content=[
                        {"type": "text", "text": "Describe the image:"},
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/jpg;base64,{base64_string}"},
                        },
                    ]
                )
            ]
        )

        # Create a Document object with the image description
        document = Document(page_content=chat_completion.content)

        # Return a list containing the single Document
        return [document]
    except Exception as e:
        print("An error occurred:", e)
        return []

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains.summarize import load_summarize_chain
from langchain.prompts import PromptTemplate

def summarize(llm, documents):
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=3000)
    docs = text_splitter.split_documents(documents)
    combine_template = """{text}
    You summarize this document in the following format.
    Be sure to write it in Korean.
    Write the results of your summary in the following format:
    제목: Title
    내용: Write your key takeaways in bullet point format
    """
    combine_prompt = PromptTemplate(template=combine_template, input_variables=['text'])
    chain = load_summarize_chain(
        llm,
        chain_type="map_reduce",
        combine_prompt=combine_prompt,
        verbose=True
        )
    summaries = chain.invoke(docs)
    return summaries  # Directly return the summaries if they are already strings

    text_splitter = RecursiveCharacterTextSplitter(chunk_size=3000)
    docs = text_splitter.split_documents(documents)

    combine_template = """{text}
    You summarize this document in the following format.
    Be sure to write it in Korean.
    Write the results of your summary in the following format:
    제목: Title
    내용: Write your key takeaways in bullet point format
    """
    combine_prompt = PromptTemplate(template=combine_template, input_variables=['text'])
    chain = load_summarize_chain(
        llm,
        chain_type="map_reduce",
        combine_prompt=combine_prompt,
        verbose=True
        )
    summaries = chain.invoke(docs)
    return summaries  # Directly return the summaries if they are already strings

rom notion_client import Client

import re
import datetime
import os
import dotenv
dotenv.load_dotenv(verbose=True)

def extract_title(summary):
    # 정규 표현식을 사용하여 제목 추출
    match = re.search(r'제목: (.+?)\n', summary)
    return match.group(1) if match else 'No Title Found'

def save_to_notion(url, summaries, type):
    notion_api_key = os.getenv('NOTION_API_KEY')
    database_id = os.getenv('NOTION_DATABASE_ID')
    current_datetime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
    notion = Client(auth=notion_api_key)
    
    title = extract_title(summaries.get('output_text', ''))

    # Create a new page
    new_page = notion.pages.create(parent={"database_id": database_id}, properties={
        "title": {
            "title": [
                {
                    "text": {
                        "content": title
                    }
                }
            ]
        },
        "Type": {
            "select": {
                "name": type
            }
        },
        "Created": {
            "date": {
                "start": current_datetime
            }
        },
        "URL": {
            "url": url
        }
    })

    # Add summary to the page
    notion.blocks.children.append(block_id=new_page["id"], children=[
        {
            "object": "block",
            "type": "paragraph",
            "paragraph": {
                "rich_text": [
                    {
                        "type": "text",
                        "text": {
                            "content": summaries.get('output_text', '')
                        }
                    }
                ]
            }
        },
        {
            "object": "block",
            "type": "embed",
            "embed": {
                "url": url
            }
        }
    ])