langgraph_example.py

import uuid
import discord
from discord.ext import commands
from langgraph_sdk import get_client
from dotenv import load_dotenv
import os

# 환경 변수 로드
load_dotenv()

# 환경 변수 설정
DISCORD_BOT_TOKEN = os.getenv('DISCORD_BOT_TOKEN')  # 디스코드 봇 토큰
LANGGRAPH_ENDPOINT = os.getenv('LANGGRAPH_ENDPOINT')  # LangGraph API 엔드포인트
LANGGRAPH_API_KEY = os.getenv('LANGGRAPH_API_KEY')  # LangGraph API 키
LANGGRAPH_ASSISTANT_ID = os.getenv(
    'LANGGRAPH_ASSISTANT_ID')  # LangGraph 어시스턴트 ID

# 디스코드 봇 인텐트 설정
intents = discord.Intents.default()
intents.message_content = True  # 메시지 내용 읽기 권한 활성화

# 디스코드 봇 초기화
bot = commands.Bot(command_prefix='!', intents=intents)

# LangGraph 클라이언트 초기화
client = get_client(api_key=LANGGRAPH_API_KEY, url=LANGGRAPH_ENDPOINT)

# 사용자별 대화 스레드 저장소
threads_db = {}


@bot.event
async def on_message(message: discord.Message):
    # 봇이 보낸 메시지는 무시
    if message.author == bot.user:
        return

    # 봇이 멘션되었는지 확인
    if bot.user.mentioned_in(message):
        # 멘션을 제거하고 메시지 내용만 추출
        query = message.content.replace(f'<@{bot.user.id}>', '').strip()

        # 메시지 보내는 동안 타이핑 표시
        async with message.channel.typing():
            try:
                # 사용자의 대화 스레드 ID 확인
                thread_id = threads_db.get(message.author.id, None)

                # 새로운 대화 스레드 생성 (첫 메시지인 경우)
                if thread_id is None:
                    thread = await client.threads.create(
                        metadata={
                            "user_id": message.author.id,
                        }
                    )
                    thread_id = thread["thread_id"]
                    threads_db[message.author.id] = thread_id

                # 첨부된 이미지 파일 처리
                files = [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": attachment.url
                        }
                    }
                    for attachment in message.attachments
                    if attachment.content_type.startswith('image/')
                ]

                # 사용자 메시지 생성
                new_human_message = {
                    "id": str(uuid.uuid4()),
                    "type": "human",
                    "content": [
                        {
                            "type": "text",
                            "text": query
                        },
                        *files
                    ],
                }

                # LangGraph API를 통해 응답 생성
                response = await client.runs.wait(
                    thread_id=thread_id,
                    assistant_id=LANGGRAPH_ASSISTANT_ID,
                    input={
                        "messages": [new_human_message],
                    },
                    metadata={
                        "user_id": message.author.id,
                    },
                    config={
                        "configurable": {
                            "model": "openai/gpt-4o"  # 사용할 AI 모델 지정
                        }
                    }
                )
                answer = response["messages"][-1]["content"]

                # 디스코드 메시지 길이 제한(2000자)에 맞춰 답변을 분할하여 전송
                chunks = [answer[i:i+2000]
                          for i in range(0, len(answer), 2000)]
                for chunk in chunks:
                    await message.channel.send(chunk)

            except Exception as e:
                await message.channel.send(f"Error: {e}")

    # 다른 명령어들을 처리하기 위해 이벤트를 계속 진행
    await bot.process_commands(message)


@bot.command()
async def clear(ctx: commands.Context):
    """!clear 명령어를 사용하면 대화 기록이 초기화됩니다."""
    # 사용자의 대화 기록 초기화
    threads_db.pop(ctx.author.id, None)
    await ctx.send("대화 기록이 초기화되었습니다.")


# 디스코드 봇 실행
bot.run(DISCORD_BOT_TOKEN)