app.py

import gradio as gr
import requests
import os
import json
from collections import deque

# 환경 변수에서 API 토큰 가져오기
TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")

# API 토큰이 설정되어 있는지 확인
if not TOKEN:
    raise ValueError("API token is not set. Please set the HUGGINGFACE_API_TOKEN environment variable.")

# 대화 기록을 관리하는 큐 (최대 10개의 대화 기록을 유지)
memory = deque(maxlen=10)

def respond(
    message,
    history: list[tuple[str, str]],
    system_message="AI Assistant Role",
    max_tokens=512,
    temperature=0.7,
    top_p=0.95,
):
    # 시스템 메시지에 접두사 추가
    system_prefix = "System: 입력어의 언어(영어, 한국어, 중국어, 일본어 등)에 따라 동일한 언어로 답변하라."
    full_system_message = f"{system_prefix}{system_message}"

    # 현재 대화 내용을 메모리에 추가
    memory.append((message, None))

    messages = [{"role": "system", "content": full_system_message}]

    # 메모리에서 대화 기록을 가져와 메시지 목록에 추가
    for val in memory:
        if val[0]:
            messages.append({"role": "user", "content": val[0]})
        if val[1]:
            messages.append({"role": "assistant", "content": val[1]})

    headers = {
        "Authorization": f"Bearer {TOKEN}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": "meta-llama/Meta-Llama-3.1-405B-Instruct",
        "max_tokens": max_tokens,
        "temperature": temperature,
        "top_p": top_p,
        "messages": messages
    }

    response = requests.post("https://api-inference.huggingface.co/v1/chat/completions", headers=headers, json=payload, stream=True)
    
    # Stream 방식으로 데이터를 출력
    response_text = ""
    for chunk in response.iter_content(chunk_size=None):
        if chunk:
            chunk_data = chunk.decode('utf-8')
            try:
                response_json = json.loads(chunk_data)
                # content 영역만 출력
                if "choices" in response_json:
                    content = response_json["choices"][0]["message"]["content"]
                    response_text += content
                    yield response_text  # 누적된 응답을 스트림 방식으로 반환
            except json.JSONDecodeError:
                continue  # 유효하지 않은 JSON이 있을 경우 무시하고 다음 청크로 넘어감

# Gradio Blocks API 사용
with gr.Blocks() as demo:
    with gr.Row():
        chatbot = gr.Chatbot()
        with gr.Column():
            message = gr.Textbox(label="Your message:")
            system_message = gr.Textbox(value="AI Assistant Role", label="System message")
            max_tokens = gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens")
            temperature = gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature")
            top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p (nucleus sampling)")
            send_button = gr.Button("Send")

    def handle_response(message, history, system_message, max_tokens, temperature, top_p):
        bot_response = respond(message, history, system_message, max_tokens, temperature, top_p)
        for response in bot_response:
            history.append((message, response))
            yield history, history

    send_button.click(
        handle_response,
        inputs=[message, chatbot, system_message, max_tokens, temperature, top_p],
        outputs=[chatbot, chatbot],
        queue=True
    )

if __name__ == "__main__":
    demo.queue().launch(max_threads=20)