SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day41_LLM의 주요 기능과 활용 사례

LLM

자연어를 이해하고 생성하는 AI모델

https://standout.tistory.com/1879

LLM의 정의와 LLM 종류

 

 

openai api는 텍스트 생성, 자연어 처리, 이미지 이해등 다양한 작업을 api로 사용할 수 있다. 

https://standout.tistory.com/1891

openai api란? api_key, 예시코드, web_search, File Search, Code Interpreter, Function Calling, Remote MCP, Streaming, Agent

https://standout.tistory.com/1892

Chatbot이란? (feat.openai chatgpt)

 

 

 

예제 프로젝트 llm_usage_fastapi_app 를 분석해보자. 

 

 

 

llm_schema.py

import pydantic basemodel, field 요청과 응답데이터 구조를 정의하기위한 pydantic basemodel 데이터 구조와 타입을 정의하는 부모클래스로 요청 데이터를 basemodel을 이용해 검증하고 응답데이터도 관리한다 . 자동으로 제디터를 검증하고 json으로 반환하며 형을 확인한다, api문서인 swaggerui에서도 이에 따른 형태가 자동생성되며 field()는 각 필드에 대한추가정보와 검증 조건을 설정하는 함수로, 기본값, 설명, 제약조건, 별칭등을 정의하는데 사용한다. description을 추가하거나 example로 예시데이터 제공, default기본값, gt/ge 숫자제한, lt/le 로 숫자범위지정, min_lenght/max_length 문자열 길이 제한, alias json이름 변경 등으로 basemodel과 field함께 사용한다.

LLMRequest() 

prompt, system_instruction, task_type, model, temperature, top_p, max_output_tokens를 설정한다. 이때 temperature는 ge0~le2, 기본값 7이고 top p 는 0, 1, 0.95, max_output_tokens는 1, 8192, 800이다.

LLMResponse()

provicer, model, task_type, result 결과이 모두 str로 저장된 객체

# 요청과 응답 데이터 구조를 정의하기 위해 Pydantic BaseModel을 가져옵니다.
from pydantic import BaseModel, Field


# LLM 호출 요청에서 공통으로 사용할 입력 데이터 구조입니다.
class LLMRequest(BaseModel):
    # LLM에게 전달할 사용자 입력 문장입니다.
    prompt: str = Field(..., description="LLM에 전달할 사용자 입력 문장")

    # 모델의 역할과 응답 방식을 지정하는 시스템 지시문입니다.
    system_instruction: str = Field("친절하고 정확한 AI 강사처럼 답변하시오.", description="시스템 지시문")

    # 기능 유형입니다. sentence, qa, summary, translation, chat, usecase 중 하나로 사용합니다.
    task_type: str = Field("sentence", description="실행할 LLM 기능 유형")

    # 사용할 모델명을 요청마다 덮어쓸 수 있습니다. 비워두면 .env 기본 모델을 사용합니다.
    model: str | None = Field(None, description="요청별 모델명")

    # 생성 다양성 설정입니다. 낮으면 안정적이고 높으면 창의적입니다.
    temperature: float = Field(0.7, ge=0.0, le=2.0, description="출력 다양성")

    # 확률 누적 범위를 제한하는 설정입니다.
    top_p: float = Field(0.95, ge=0.0, le=1.0, description="Top-p 샘플링")

    # 최대 출력 토큰 수입니다.
    max_output_tokens: int = Field(800, ge=1, le=8192, description="최대 출력 토큰 수")


# LLM 호출 결과를 클라이언트에 반환하기 위한 응답 데이터 구조입니다.
class LLMResponse(BaseModel):
    # 어떤 공급자 API를 사용했는지 표시합니다.
    provider: str

    # 실제 호출에 사용된 모델명을 표시합니다.
    model: str

    # 실행한 기능 유형을 표시합니다.
    task_type: str

    # LLM이 생성한 최종 텍스트입니다.
    result: str

 

 

 

config.py

import os, detrnv, 

pydantic_settings BaseSettings 애플리케이션 설정을 관리하기 위한 Pydantic클래스. 코드에 직접 적지않고 환경변수 .env 파일또는 운영체제 환경변수에 저장된 설정값을 자동으로 읽어오는 기능. api키, 데이터베이스 주소, 비밀번호등을 코드에 하드코딩하면 보산상 좋지않으니 설정을 안전하고 일관되게 관리할수있도록 도와준다. BaseModel과 차이는 요청과 응답데이터 json데이터를 검증한다면 basesettings는 설정을 읽고 검증한다는것. 

load_env()

settings()

env를 읽어 각 키를 변수에 할당함

settings() 객체생성

# 운영체제 환경변수를 읽기 위해 os 모듈을 가져옵니다.
import os

# .env 파일의 값을 환경변수로 등록하기 위해 load_dotenv 함수를 가져옵니다.
from dotenv import load_dotenv

# Pydantic 기반 설정 클래스를 만들기 위해 BaseSettings를 가져옵니다.
from pydantic_settings import BaseSettings

# .env 파일을 현재 프로세스의 환경변수로 로드합니다.
load_dotenv()


# 프로젝트 전체에서 사용할 설정값을 한 곳에서 관리하는 클래스입니다.
class Settings(BaseSettings):
    # OpenAI API 호출에 사용할 API Key입니다.
    openai_api_key: str | None = os.getenv("OPENAI_API_KEY")

    # Gemini API 호출에 사용할 API Key입니다.
    gemini_api_key: str | None = os.getenv("GEMINI_API_KEY")

    # OpenAI 호출에 사용할 기본 모델명입니다.
    openai_model: str = os.getenv("OPENAI_MODEL", "gpt-5.5")

    # Gemini 호출에 사용할 기본 모델명입니다.
    gemini_model: str = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")

    # 생성 결과의 다양성을 조절하는 기본 temperature 값입니다.
    default_temperature: float = float(os.getenv("DEFAULT_TEMPERATURE", "0.7"))

    # 후보 토큰 누적 확률 범위를 조절하는 기본 top_p 값입니다.
    default_top_p: float = float(os.getenv("DEFAULT_TOP_P", "0.95"))

    # 모델이 생성할 수 있는 최대 출력 토큰 수입니다.
    default_max_output_tokens: int = int(os.getenv("DEFAULT_MAX_OUTPUT_TOKENS", "800"))

    # API 요청이 너무 오래 걸릴 때 중단하기 위한 기본 제한 시간입니다.
    default_timeout_seconds: int = int(os.getenv("DEFAULT_TIMEOUT_SECONDS", "60"))


# 앱 어디서나 import 하여 사용할 설정 객체를 생성합니다.
settings = Settings()

 

 

 

prompt_factory.py

build_prompt()

프롬프트 만들기, task_type을 lower() 소문자로 통일해 strip() 공백 제거

task가 sentence일때, qa일때, summary일때, translation일때, chat일때, usecase일때 적절한 prompt안에 userinput들을 할당해 return함. 

이때 f**** 알수없는 타입을 위해 기본 질의응답 방식도 정의.

# 기능 유형별 프롬프트를 생성하는 함수입니다.
def build_prompt(task_type: str, user_prompt: str) -> str:
    # task_type 값을 소문자로 통일하여 비교 오류를 줄입니다.
    task = task_type.lower().strip()

    # 문장 생성 기능 프롬프트입니다.
    if task == "sentence":
        return f"""
다음 주제에 맞는 자연스러운 문장을 생성하시오.

주제:
{user_prompt}
"""

    # 질의 응답 기능 프롬프트입니다.
    if task == "qa":
        return f"""
다음 질문에 대해 핵심 개념, 이유, 예시를 포함하여 답변하시오.

질문:
{user_prompt}
"""

    # 요약 기능 프롬프트입니다.
    if task == "summary":
        return f"""
다음 내용을 핵심 중심으로 요약하시오.
가능하면 3~5개의 문장으로 정리하시오.

내용:
{user_prompt}
"""

    # 번역 기능 프롬프트입니다.
    if task == "translation":
        return f"""
다음 문장을 자연스럽게 번역하시오.
한국어이면 영어로, 영어이면 한국어로 번역하시오.

문장:
{user_prompt}
"""

    # 채팅 기능 프롬프트입니다.
    if task == "chat":
        return f"""
사용자와 자연스럽게 대화하시오.
이전 대화 맥락이 있다면 참고하여 답변하시오.

사용자 입력:
{user_prompt}
"""

    # 다양한 Use Case 기능 프롬프트입니다.
    if task == "usecase":
        return f"""
다음 요구사항을 실제 업무 활용 사례처럼 처리하시오.
필요하면 결과물 형식까지 갖추어 작성하시오.

요구사항:
{user_prompt}
"""

    # 알 수 없는 task_type이면 기본 질의응답 방식으로 처리합니다.
    return f"""
다음 요청을 분석하여 적절한 답변을 작성하시오.

요청:
{user_prompt}
"""

 

 

 

openai_service.py

import openai openai

app.core의 config를 가져온다. 

app.service의 prompt_factory를 가져온다.

# OpenAI 공식 Python SDK에서 OpenAI 클라이언트를 가져옵니다.
from openai import OpenAI

# 앱 설정값을 가져옵니다.
from app.core.config import settings

# 기능별 프롬프트 생성 함수를 가져옵니다.
from app.services.prompt_factory import build_prompt

 

OpenAIService 클래스.

init() api key 없으면 에러발생. openai() client 생성

generate() 각 prompt, task_type, sysstem_instruction, model, temperature, top_p, max_output_tokens를 받는다. 

선택된 model이있으면 사용, 없으면 settings.openai_model 기본설정을 사용한다. 

prompt buildpromp 앞선 prompt_factory에서 적절한 프롬프트를 생성

self.client.response.create() 매개변수를 활용해 openai response api 호출

생성된 테기스트와 model을 return.

# OpenAI 공식 Python SDK에서 OpenAI 클라이언트를 가져옵니다.
from openai import OpenAI

# 앱 설정값을 가져옵니다.
from app.core.config import settings

# 기능별 프롬프트 생성 함수를 가져옵니다.
from app.services.prompt_factory import build_prompt


# OpenAI API 호출을 담당하는 서비스 클래스입니다.
class OpenAIService:
    # 서비스 객체가 생성될 때 API 클라이언트를 준비합니다.
    def __init__(self):
        # API Key가 없으면 명확한 오류 메시지를 발생시킵니다.
        if not settings.openai_api_key:
            raise ValueError("OPENAI_API_KEY가 설정되지 않았습니다. .env 파일을 확인하십시오.")

        # OpenAI 클라이언트를 생성합니다.
        self.client = OpenAI(api_key=settings.openai_api_key, timeout=settings.default_timeout_seconds)

    # OpenAI Responses API를 사용하여 텍스트를 생성합니다.
    def generate(
        self,
        prompt: str,
        task_type: str,
        system_instruction: str,
        model: str | None,
        temperature: float,
        top_p: float,
        max_output_tokens: int,
    ) -> tuple[str, str]:
        # 요청별 모델명이 있으면 사용하고, 없으면 .env 기본 모델을 사용합니다.
        selected_model = model or settings.openai_model

        # 기능 유형에 맞는 최종 프롬프트를 생성합니다.
        final_prompt = build_prompt(task_type, prompt)

        # OpenAI Responses API를 호출합니다.
        response = self.client.responses.create(
            # 사용할 OpenAI 모델명을 지정합니다.
            model=selected_model,
            # 시스템 지시문으로 모델의 응답 태도와 역할을 지정합니다.
            instructions=system_instruction,
            # 실제 사용자 요청 프롬프트를 전달합니다.
            input=final_prompt,
            # 출력의 창의성과 무작위성을 조절합니다.
            temperature=temperature,
            # 누적 확률 기반 샘플링 범위를 조절합니다.
            top_p=top_p,
            # 최대 출력 길이를 제한합니다.
            max_output_tokens=max_output_tokens,
        )

        # 생성된 텍스트와 실제 사용 모델명을 반환합니다.
        return response.output_text, selected_model

 

 

 

gemini_service.py

google에서 genai 불러오기 공식 ai SDK 클라이언트

google.genai types 생성설정 타입을 사용하기 위한 types 모듈

app.core config가져오기. 

app.service prompt_factory가져오기

Geminiservice 클래스

init() api 키가 없으면 오류를 발생시키고 genai.client() 클라이언트 생성

generate() 마찬가지로 매개변수를 받는다. 

모델선택된것 혹은 기본으로 설정하고 프롬프트 생성해 types.generatecontenconfig() 생성옵션을 구성해 client.models.generate_content()해 똑같이 결과 text와 model return.

# Google Gemini 공식 Gen AI SDK 클라이언트를 가져옵니다.
from google import genai

# Gemini 생성 설정 타입을 사용하기 위해 types 모듈을 가져옵니다.
from google.genai import types

# 앱 설정값을 가져옵니다.
from app.core.config import settings

# 기능별 프롬프트 생성 함수를 가져옵니다.
from app.services.prompt_factory import build_prompt


# Gemini API 호출을 담당하는 서비스 클래스입니다.
class GeminiService:
    # 서비스 객체가 생성될 때 Gemini 클라이언트를 준비합니다.
    def __init__(self):
        # API Key가 없으면 명확한 오류 메시지를 발생시킵니다.
        if not settings.gemini_api_key:
            raise ValueError("GEMINI_API_KEY가 설정되지 않았습니다. .env 파일을 확인하십시오.")

        # Gemini Developer API용 클라이언트를 생성합니다.
        self.client = genai.Client(api_key=settings.gemini_api_key)

    # Gemini generate_content API를 사용하여 텍스트를 생성합니다.
    def generate(
        self,
        prompt: str,
        task_type: str,
        system_instruction: str,
        model: str | None,
        temperature: float,
        top_p: float,
        max_output_tokens: int,
    ) -> tuple[str, str]:
        # 요청별 모델명이 있으면 사용하고, 없으면 .env 기본 모델을 사용합니다.
        selected_model = model or settings.gemini_model

        # 기능 유형에 맞는 최종 프롬프트를 생성합니다.
        final_prompt = build_prompt(task_type, prompt)

        # Gemini 생성 옵션을 구성합니다.
        config = types.GenerateContentConfig(
            # 시스템 지시문으로 모델의 역할과 응답 방식을 지정합니다.
            system_instruction=system_instruction,
            # 출력의 창의성과 무작위성을 조절합니다.
            temperature=temperature,
            # 누적 확률 기반 샘플링 범위를 조절합니다.
            top_p=top_p,
            # 최대 출력 길이를 제한합니다.
            max_output_tokens=max_output_tokens,
        )

        # Gemini API에 프롬프트와 설정을 전달하여 결과를 생성합니다.
        response = self.client.models.generate_content(
            # 사용할 Gemini 모델명을 지정합니다.
            model=selected_model,
            # 사용자 요청 프롬프트를 전달합니다.
            contents=final_prompt,
            # 앞에서 구성한 생성 설정을 전달합니다.
            config=config,
        )

        # Gemini 응답 텍스트와 실제 사용 모델명을 반환합니다.
        return response.text or "", selected_model

 

 

 

llm_router.py

import

fastapi apirouter, httpexception fastapi라우터와 http 예외처리를 사용하기 위해 importm.

app.schemas에서의 llm_schema, 

app.services에서의 openai_service

app.services의 gemini_service iport

라우터 설정

# FastAPI 라우터와 HTTP 예외 처리를 사용하기 위해 가져옵니다.
from fastapi import APIRouter, HTTPException

# 요청과 응답 데이터 모델을 가져옵니다.
from app.schemas.llm_schema import LLMRequest, LLMResponse

# OpenAI API 호출 서비스 클래스를 가져옵니다.
from app.services.openai_service import OpenAIService

# Gemini API 호출 서비스 클래스를 가져옵니다.
from app.services.gemini_service import GeminiService

# /api/llm 하위 URL을 담당하는 라우터를 생성합니다.
router = APIRouter(prefix="/api/llm", tags=["LLM"])

 

@router.post() /openai일때 

run_openai() 

openai 객체를 생성해 요청값과 config값을 함께 전달해 호출, 각 호출, 결과값들이 잘 str형으로 정리된 LLMResponse() 클래스 객체로 return. ㅇ에러발생시 HTTPException()객체 raise

@router.post() /gemini일때도 마찬가지로

genimiservice() 서비스 객체를 만들어 service.generate() api를 호출한뒤에 LLMResponse()로 return, 에러처리

@router.post() /openai/sentence type이 sentence일때 type을 바꾸어 return run_openai 

외 각각 qa일때, summart일때, translateion일때, chat일때 , usecase일때각각 return. 

@router.post() /gemini/sentence일때도 마찬가지. return run_gemini(), 

외 qa, sumary, translation, chat, usecase 도 각각 return.

# FastAPI 라우터와 HTTP 예외 처리를 사용하기 위해 가져옵니다.
from fastapi import APIRouter, HTTPException

# 요청과 응답 데이터 모델을 가져옵니다.
from app.schemas.llm_schema import LLMRequest, LLMResponse

# OpenAI API 호출 서비스 클래스를 가져옵니다.
from app.services.openai_service import OpenAIService

# Gemini API 호출 서비스 클래스를 가져옵니다.
from app.services.gemini_service import GeminiService

# /api/llm 하위 URL을 담당하는 라우터를 생성합니다.
router = APIRouter(prefix="/api/llm", tags=["LLM"])


# OpenAI API를 사용하는 공통 실행 엔드포인트입니다.
@router.post("/openai", response_model=LLMResponse)
def run_openai(request: LLMRequest):
    # API 호출 중 오류가 발생할 수 있으므로 예외 처리를 적용합니다.
    try:
        # OpenAI 서비스 객체를 생성합니다.
        service = OpenAIService()

        # 요청값과 config 값을 함께 전달하여 OpenAI API를 호출합니다.
        result, model = service.generate(
            prompt=request.prompt,
            task_type=request.task_type,
            system_instruction=request.system_instruction,
            model=request.model,
            temperature=request.temperature,
            top_p=request.top_p,
            max_output_tokens=request.max_output_tokens,
        )

        # FastAPI 응답 모델에 맞추어 결과를 반환합니다.
        return LLMResponse(provider="openai", model=model, task_type=request.task_type, result=result)

    # 모든 예외를 HTTP 500 응답으로 변환하여 클라이언트에 전달합니다.
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# Gemini API를 사용하는 공통 실행 엔드포인트입니다.
@router.post("/gemini", response_model=LLMResponse)
def run_gemini(request: LLMRequest):
    # API 호출 중 오류가 발생할 수 있으므로 예외 처리를 적용합니다.
    try:
        # Gemini 서비스 객체를 생성합니다.
        service = GeminiService()

        # 요청값과 config 값을 함께 전달하여 Gemini API를 호출합니다.
        result, model = service.generate(
            prompt=request.prompt,
            task_type=request.task_type,
            system_instruction=request.system_instruction,
            model=request.model,
            temperature=request.temperature,
            top_p=request.top_p,
            max_output_tokens=request.max_output_tokens,
        )

        # FastAPI 응답 모델에 맞추어 결과를 반환합니다.
        return LLMResponse(provider="gemini", model=model, task_type=request.task_type, result=result)

    # 모든 예외를 HTTP 500 응답으로 변환하여 클라이언트에 전달합니다.
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# 문장 생성 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/sentence", response_model=LLMResponse)
def openai_sentence(request: LLMRequest):
    # 요청의 기능 유형을 문장 생성으로 고정합니다.
    request.task_type = "sentence"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 질의응답 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/qa", response_model=LLMResponse)
def openai_qa(request: LLMRequest):
    # 요청의 기능 유형을 질의응답으로 고정합니다.
    request.task_type = "qa"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 요약 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/summary", response_model=LLMResponse)
def openai_summary(request: LLMRequest):
    # 요청의 기능 유형을 요약으로 고정합니다.
    request.task_type = "summary"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 번역 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/translation", response_model=LLMResponse)
def openai_translation(request: LLMRequest):
    # 요청의 기능 유형을 번역으로 고정합니다.
    request.task_type = "translation"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 채팅 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/chat", response_model=LLMResponse)
def openai_chat(request: LLMRequest):
    # 요청의 기능 유형을 채팅으로 고정합니다.
    request.task_type = "chat"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 다양한 Use Case 전용 OpenAI 엔드포인트입니다.
@router.post("/openai/usecase", response_model=LLMResponse)
def openai_usecase(request: LLMRequest):
    # 요청의 기능 유형을 Use Case로 고정합니다.
    request.task_type = "usecase"
    # 공통 OpenAI 실행 함수를 재사용합니다.
    return run_openai(request)


# 문장 생성 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/sentence", response_model=LLMResponse)
def gemini_sentence(request: LLMRequest):
    # 요청의 기능 유형을 문장 생성으로 고정합니다.
    request.task_type = "sentence"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)


# 질의응답 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/qa", response_model=LLMResponse)
def gemini_qa(request: LLMRequest):
    # 요청의 기능 유형을 질의응답으로 고정합니다.
    request.task_type = "qa"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)


# 요약 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/summary", response_model=LLMResponse)
def gemini_summary(request: LLMRequest):
    # 요청의 기능 유형을 요약으로 고정합니다.
    request.task_type = "summary"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)


# 번역 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/translation", response_model=LLMResponse)
def gemini_translation(request: LLMRequest):
    # 요청의 기능 유형을 번역으로 고정합니다.
    request.task_type = "translation"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)


# 채팅 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/chat", response_model=LLMResponse)
def gemini_chat(request: LLMRequest):
    # 요청의 기능 유형을 채팅으로 고정합니다.
    request.task_type = "chat"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)


# 다양한 Use Case 전용 Gemini 엔드포인트입니다.
@router.post("/gemini/usecase", response_model=LLMResponse)
def gemini_usecase(request: LLMRequest):
    # 요청의 기능 유형을 Use Case로 고정합니다.
    request.task_type = "usecase"
    # 공통 Gemini 실행 함수를 재사용합니다.
    return run_gemini(request)

 

 

main.py

Import FastAPI fastapi 애플리케이션 생성을 위한. 

fastapi.staticifiles staticfiles 정적파일 서비스

fastapi.responses fileresponse html 파일 응다비을 반환하기 위한 fileresponse

app.api에 명시한 llm_router import

# FastAPI 애플리케이션 생성을 위해 FastAPI 클래스를 가져옵니다.
from fastapi import FastAPI

# 정적 파일을 서비스하기 위해 StaticFiles 클래스를 가져옵니다.
from fastapi.staticfiles import StaticFiles

# HTML 파일 응답을 반환하기 위해 FileResponse 클래스를 가져옵니다.
from fastapi.responses import FileResponse

# LLM API 라우터를 가져옵니다.
from app.api.llm_router import router as llm_router

 

 

fastapi 객체 생성

app = FastAPI(
    # Swagger 문서에 표시될 앱 이름입니다.
    title="OpenAI + Gemini LLM FastAPI App",
    # Swagger 문서에 표시될 앱 설명입니다.
    description="문장 생성, 질의응답, 요약, 번역, 채팅, Use Case 예제를 OpenAI API와 Gemini API로 실행하는 앱입니다.",
    # 앱 버전입니다.
    version="1.0.0",
)

 

 

app.include_roupter llm_router llm 경로의 llm라우터를 앱에 등록한다. include

static 폴더를 staticFiles() url로 서비스한다. app.mount() 

# /api/llm 경로의 LLM 라우터를 앱에 등록합니다.
app.include_router(llm_router)

# app/static 폴더를 /static URL로 서비스합니다.
app.mount("/static", StaticFiles(directory="app/static"), name="static")

 

 

@app.get() / 일때 , 즉 사용자가 접속히 index.html파일을 반환한다. 

@app.get() /api/health일때 서버가 정상실행중임을 문자열로 알려주자

# 루트 URL에서 HTML UI를 반환합니다.
@app.get("/")
def index():
    # 사용자가 브라우저로 접속하면 index.html 파일을 반환합니다.
    return FileResponse("app/static/index.html")


# 서버 상태 확인용 엔드포인트입니다.
@app.get("/api/health")
def health():
    # 서버가 정상 실행 중임을 JSON으로 반환합니다.
    return {"status": "ok", "message": "LLM FastAPI server is running"}

 

 

 

 

 

 

 

이번에는 음악추천 chatbot 서비스 예제프로젝트를 분석해보자. 

 

 

사용할 데이터들은 아래와 같은 형식이다 .

[
  {
    "title": "Noel(노을) - I Miss You(그리워 그리워)",
    "artist": "Noel",
    "mood": "그리움",
    "tags": [
      "그리움",
      "보고싶다",
      "이별",
      "발라드",
      "추억",
      "슬픔"
    ],
    "reason": "그리운 사람을 떠올릴 때 어울리는 감성 발라드입니다.",
    "url": "https://www.youtube.com/watch?v=rr7arlNxLro"
  },

 

 

 

 

music_recommender.py

import path, torch

class MusicRecomender

init() path, songs, vocabulary, song_vectors 객체 만들어 할당

_load_dataset() 매개변수로 받은 dataset_path위치의 json파일을 열어 retun.

_build_vocabulary() set()ㅇ로 중복제거, songs를 for문으로 반복해 돌려가면 tags 리스트를 반복해 전처리. strip(), lower()해 return

_text_to_vector() 벡터 크기만큼 torch.zero 0으로 채워진 Tensor생성, text를 소문자로 변환해 for문으로 사전의 각 단어가 입력 문장에 들어있는지 확인. 키워드가 문장에 포함되어있다면 해당위치의 값을 1로 설정.

extra_tags가 전달된경우 for문으로 돌려 strip().lower() 문자열을 정리하고 단어사전에 있으면 해당 위치를 1로 설정. 

synonym_map 사용자가 태그 단어를 쓰지않은 경우를 위해 간단한 동의어 규칙 정의, for문으로 반복하며  만일 normalized_text 입력문장에 포함되어있을때 for문을 돌려 lower(), 태그들을 반영, 해당 위치에 number 1를 반영해 해당 감성 태그가 존재함을 표시해 return한다. 

_build_song_vectors 

torch.tensor형 list로 음악별 벡터를 저장할 리스트를 만든다 .

songs를 for문으로 반복해 title, artist, mood, tags를 하나의 텍스트로 합쳐 태그와 함께 tensor 벡터로 변환해 return. 

벡터 리스트들을 하나의 2차원 tensor로 묶었다. 

recommend

사용자가 입력한 문장을 tensor 벡터로 변환한다. 없을경우에는 작은 값을 더한다. 

사용자 벡터와 음악벡터 사이의  torch.nn.functional.cosine_similarity()계산.  두벡터가 얼마나 비슷한 방향을 가지고 있는지 유사도를 계산하는 함수. 

추천개수가 음악개수보다 커지지않도록 수의 min값을 정의한다. 

유사도가 높은 순서대로 top k 개의 인덱스를 구한다. 

추천결과를 리스트에 저장하도록 리스트 정의. 

for문을 돌려  zip(top_scores.tolist(), top_indices.tolist()) score와 index를 하나씩 pairing해 묶어주는 함수 zip을 사용해 점수와 노래위치를 함께 사용해 어떤 노래가 얼마의 유사도를 가지는지 처리한다. zip을 사용하지않으면 인덱슬르 직접 관리해야하고 사용하면 더 간결하고 실수를 줄일 수있다 .

song에 음악할당. 

점수추가. 

추천리스트에 추가 .

recommendations mood에 가장 높은 점수를 받은 감정으로 사용. return

import json
from pathlib import Path

import torch


class MusicRecommender:
    """
    PyTorch Tensor 기반 음악 추천 클래스입니다.

    이 클래스는 사용자의 감정 문장과 음악 데이터의 감성 태그를
    같은 단어 사전 벡터로 변환한 뒤 Cosine Similarity로 유사도를 계산합니다.

    대규모 딥러닝 학습 모델은 아니지만,
    추천 시스템의 핵심 개념인 '사용자 입력과 콘텐츠 간 유사도 계산'을
    PyTorch로 직접 확인할 수 있도록 구성한 교육용 모델입니다.
    """

    def __init__(self, dataset_path: str | Path):
        """
        추천 모델 초기화 함수입니다.

        dataset_path:
            음악 추천 데이터가 들어 있는 JSON 파일 경로입니다.
        """

        # 문자열 경로를 Path 객체로 변환하여 파일 경로 처리를 안전하게 합니다.
        self.dataset_path = Path(dataset_path)

        # JSON 파일을 읽어 음악 데이터 리스트를 메모리에 저장합니다.
        self.songs = self._load_dataset()

        # 음악 데이터의 모든 tag 값을 모아 추천 모델에서 사용할 단어 사전을 만듭니다.
        self.vocabulary = self._build_vocabulary()

        # 각 음악 데이터를 PyTorch Tensor 벡터로 미리 변환합니다.
        self.song_vectors = self._build_song_vectors()

    def _load_dataset(self) -> list[dict]:
        """
        JSON 음악 데이터셋을 읽어 리스트로 반환합니다.
        """

        # UTF-8 인코딩으로 JSON 파일을 엽니다.
        with self.dataset_path.open("r", encoding="utf-8") as file:
            # JSON 문자열을 Python 리스트/딕셔너리 구조로 변환합니다.
            return json.load(file)

    def _build_vocabulary(self) -> list[str]:
        """
        음악 데이터의 모든 감성 태그를 중복 없이 모아 단어 사전을 만듭니다.
        """

        # 중복 제거를 위해 set 자료구조를 사용합니다.
        vocab_set: set[str] = set()

        # 모든 음악 데이터를 반복합니다.
        for song in self.songs:
            # 각 음악에 등록된 tags 리스트를 반복합니다.
            for tag in song.get("tags", []):
                # 태그 앞뒤 공백을 제거한 뒤 소문자 형태로 저장합니다.
                vocab_set.add(tag.strip().lower())

        # 결과가 항상 같은 순서가 되도록 정렬해서 리스트로 반환합니다.
        return sorted(vocab_set)

    def _text_to_vector(self, text: str, extra_tags: list[str] | None = None) -> torch.Tensor:
        """
        입력 문장을 단어 사전 크기의 PyTorch Tensor 벡터로 변환합니다.

        text:
            사용자 입력 문장 또는 음악 태그 문자열입니다.

        extra_tags:
            이미 분리된 태그 리스트가 있는 경우 추가로 반영합니다.
        """

        # 단어 사전의 크기만큼 0으로 채워진 Tensor를 생성합니다.
        vector = torch.zeros(len(self.vocabulary), dtype=torch.float32)

        # 비교를 쉽게 하기 위해 입력 문장을 소문자로 변환합니다.
        normalized_text = text.lower()

        # 단어 사전의 각 단어가 입력 문장에 들어 있는지 확인합니다.
        for index, keyword in enumerate(self.vocabulary):
            # 키워드가 문장에 포함되어 있으면 해당 위치의 값을 1로 설정합니다.
            if keyword in normalized_text:
                vector[index] = 1.0

        # extra_tags가 전달된 경우 태그 기반으로도 벡터를 보강합니다.
        if extra_tags:
            # 태그 리스트를 반복합니다.
            for tag in extra_tags:
                # 태그 문자열을 정리합니다.
                normalized_tag = tag.strip().lower()

                # 태그가 단어 사전에 있으면 해당 위치를 1로 설정합니다.
                if normalized_tag in self.vocabulary:
                    vector[self.vocabulary.index(normalized_tag)] = 1.0

        # 사용자가 직접적인 태그 단어를 쓰지 않는 경우를 보완하기 위한 간단한 동의어 규칙입니다.
        synonym_map = {
            "우울": ["슬픔", "위로"],
            "힘들": ["위로", "힐링"],
            "외로": ["그리움", "슬픔"],
            "보고": ["그리움"],
            "사랑": ["사랑", "설렘"],
            "좋아": ["행복", "기쁨"],
            "기뻐": ["행복", "기쁨"],
            "화": ["화남", "분노", "진정"],
            "공부": ["집중"],
            "운동": ["활력", "운동"],
        }

        # 동의어 규칙을 반복하면서 사용자 표현을 감성 태그로 연결합니다.
        for cue, mapped_tags in synonym_map.items():
            # cue가 입력 문장에 포함되어 있는지 확인합니다.
            if cue in normalized_text:
                # 연결된 태그들을 벡터에 반영합니다.
                for tag in mapped_tags:
                    normalized_tag = tag.lower()
                    if normalized_tag in self.vocabulary:
                        vector[self.vocabulary.index(normalized_tag)] = 1.0

        # 완성된 Tensor 벡터를 반환합니다.
        return vector

    def _build_song_vectors(self) -> torch.Tensor:
        """
        모든 음악 데이터를 Tensor 행렬로 변환합니다.

        반환 Tensor의 모양:
            [음악 개수, 단어 사전 크기]
        """

        # 음악별 벡터를 저장할 리스트입니다.
        vectors: list[torch.Tensor] = []

        # 모든 음악 데이터를 반복합니다.
        for song in self.songs:
            # title, artist, mood, tags를 하나의 텍스트로 합칩니다.
            text = " ".join([
                song.get("title", ""),
                song.get("artist", ""),
                song.get("mood", ""),
                " ".join(song.get("tags", [])),
            ])

            # 합쳐진 텍스트와 태그를 Tensor 벡터로 변환합니다.
            vectors.append(self._text_to_vector(text, song.get("tags", [])))

        # 벡터 리스트를 하나의 2차원 Tensor로 묶습니다.
        return torch.stack(vectors)

    def recommend(self, user_message: str, top_k: int = 3) -> tuple[str, list[dict]]:
        """
        사용자 입력 문장에 맞는 음악을 추천합니다.

        user_message:
            사용자의 감정 또는 상황 문장입니다.

        top_k:
            상위 몇 개의 음악을 추천할지 결정합니다.

        반환값:
            추정 감정 문자열과 추천 음악 리스트입니다.
        """

        # 사용자 입력 문장을 Tensor 벡터로 변환합니다.
        user_vector = self._text_to_vector(user_message)

        # 사용자가 감성 키워드를 전혀 입력하지 않은 경우 기본 추천을 위해 작은 값을 더합니다.
        if torch.sum(user_vector) == 0:
            # 모든 음악이 완전히 0점이 되는 것을 방지하기 위해 전체 벡터에 작은 값을 부여합니다.
            user_vector = torch.ones_like(user_vector) * 0.01

        # 사용자 벡터와 음악 벡터 사이의 Cosine Similarity를 계산합니다.
        scores = torch.nn.functional.cosine_similarity(
            user_vector.unsqueeze(0),
            self.song_vectors,
            dim=1,
        )

        # 추천 개수가 전체 음악 개수보다 커지지 않도록 보정합니다.
        safe_top_k = min(top_k, len(self.songs))

        # 유사도가 높은 순서대로 top_k개의 인덱스를 구합니다.
        top_scores, top_indices = torch.topk(scores, k=safe_top_k)

        # 추천 결과를 저장할 리스트입니다.
        recommendations: list[dict] = []

        # 상위 추천 결과를 하나씩 구성합니다.
        for score, index in zip(top_scores.tolist(), top_indices.tolist()):
            # 원본 음악 데이터를 복사합니다.
            song = dict(self.songs[index])

            # API 응답용 유사도 점수를 추가합니다.
            song["score"] = round(float(score), 4)

            # 추천 리스트에 추가합니다.
            recommendations.append(song)

        # 가장 높은 점수를 받은 음악의 mood를 대표 감정으로 사용합니다.
        detected_mood = recommendations[0]["mood"] if recommendations else "일반"

        # 대표 감정과 추천 리스트를 반환합니다.
        return detected_mood, recommendations

 

 

 

chat_schema.py

pydantic에서 basemodel과 field를 가져와 스키마를 설정한다 .

ChatMessage()에는 role과 content를 정의한다. 이것은 이전대화기록 하나를 표현한다 .

MusicChatRequest() message, conversation_history에는 list형으로 ,top_k도 설정한다. 이것은 사용자가 음악추천챗봇에 보내는 요청스키마이다. 

RecommendedSong() title, artist, mood, reason, url, score 로 정의

MusicChatResponse() answer, detected_mod, recommendatios으로 추천챗봇 응답스키마이다. 

from pydantic import BaseModel, Field


class ChatMessage(BaseModel):
    """
    이전 대화 기록 하나를 표현하는 스키마입니다.

    role은 user 또는 assistant 값을 사용합니다.
    content에는 실제 대화 문장을 저장합니다.
    """

    role: str = Field(..., description="대화 발화자 역할: user 또는 assistant")
    content: str = Field(..., description="대화 내용")


class MusicChatRequest(BaseModel):
    """
    음악 추천 챗봇에 사용자가 보내는 요청 스키마입니다.
    """

    message: str = Field(..., description="사용자 입력 문장")
    conversation_history: list[ChatMessage] = Field(
        default_factory=list,
        description="브라우저에서 보관한 이전 대화 기록",
    )
    top_k: int = Field(
        default=3,
        ge=1,
        le=5,
        description="추천받을 음악 개수",
    )


class RecommendedSong(BaseModel):
    """
    추천 결과로 반환할 음악 한 곡의 정보입니다.
    """

    title: str = Field(..., description="음악 제목")
    artist: str = Field(..., description="가수 또는 연주자")
    mood: str = Field(..., description="대표 감정")
    reason: str = Field(..., description="추천 이유")
    url: str = Field(..., description="음악 또는 영상 URL")
    score: float = Field(..., description="PyTorch 유사도 점수")


class MusicChatResponse(BaseModel):
    """
    음악 추천 챗봇 응답 스키마입니다.
    """

    answer: str = Field(..., description="OpenAI ChatGPT가 생성한 자연어 답변")
    detected_mood: str = Field(..., description="추천 모델이 추정한 대표 감정")
    recommendations: list[RecommendedSong] = Field(..., description="추천 음악 목록")

 

 

 

 

config.py

functools의 lru_cache 함수의 실행결과를 메모리에 캐싱해 저장된 결과를 반환하는 데코레이더 

pydantic_setting  환경변수를 자동으로 읽어오는 basesettings, 환경설정을 어떻게 읽을지 지정하는 설정클래스 settingsconfigdict

Settings()

title지정.

개발모드여부 debug 설정, api key 및 모델, temperature, top p, maxoutput tokens, 설정. 

settingsconfigDict()를 활용해 env 파일을 읽어 model_config에 할당

get_settings() settings() 수행

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """
    앱 전체에서 공통으로 사용할 환경설정 클래스입니다.

    BaseSettings를 사용하면 .env 파일 또는 운영체제 환경변수에서
    설정값을 자동으로 읽어올 수 있습니다.
    """

    # FastAPI 문서와 브라우저 제목에 사용할 앱 이름입니다.
    APP_TITLE: str = "OpenAI Music Recommendation Chatbot"

    # 개발 모드 여부를 저장합니다.
    DEBUG: bool = True

    # OpenAI API 호출에 필요한 API Key입니다.
    OPENAI_API_KEY: str = ""

    # 기본 OpenAI 모델명입니다.
    # gpt-5.5를 기본값으로 두되, .env에서 다른 모델로 변경할 수 있습니다.
    OPENAI_MODEL: str = "gpt-5.5"

    # 일반 GPT 모델에서 사용할 창의성 조절값입니다.
    # 값이 높으면 다양한 답변을 생성하고, 낮으면 안정적인 답변을 생성합니다.
    OPENAI_TEMPERATURE: float = 0.7

    # 일반 GPT 모델에서 사용할 누적 확률 샘플링 값입니다.
    OPENAI_TOP_P: float = 0.95

    # OpenAI 응답의 최대 출력 토큰 수입니다.
    OPENAI_MAX_OUTPUT_TOKENS: int = 900

    # .env 파일을 읽도록 설정합니다.
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )


@lru_cache
def get_settings() -> Settings:
    """
    Settings 객체를 한 번만 생성해서 재사용하는 함수입니다.

    lru_cache를 사용하면 요청이 들어올 때마다 .env 파일을 반복해서 읽지 않고
    이미 만들어진 설정 객체를 재사용할 수 있습니다.
    """
    return Settings()

 

 

 

 

openai_service

import openai과 app.core의 config을 가져왔다. 

class OpenAIChatService

__init__()

settings 을 실행해 변수에 저장한다. openai()  클라이언트 생성 openai api key가 비어있으면 클라이언트를 만들지 않는다 .

_is_reasoning_model() 모델명을 lower(), model prefix를 튜플로 정의해 할당해 return. 

모델명이 해당 prefiz중 하나로 시작하는지를 반환한다. normalized_model.startswith()

build_music_prompt()

for문으로 recommendations를 돌려 각 음악의 제목, 가수, 감정, 이유, url을 한줄로 정리해 줄바꿈 문자열로 합쳐 프롬프트에 첨부해 return.

generate_music_answer() 

client가 없을경우 return.한다.

instructions에 시스템 지시문 할당

위 설정들을 request_params에 담아 

build_music_prompt() 수행해 prompt에 할당, 

_is_reasoning_model reasoning 모델이면 temperature, top_p를 보내지않도록하고 나머지에는 전달한다. 

client,response,create() 호출해 output_text해 return.

_fallback_answer

답변의 첫문장을 만들어 detected_mood를 반영한다. 

for문을 돌려 음악을 번호 목록으로 추가 .lines.append()

저작권 안내 문장을 마지막에 추가. 

return.

from openai import OpenAI

from app.core.config import Settings


class OpenAIChatService:
    """
    OpenAI ChatGPT API 호출을 담당하는 서비스 클래스입니다.

    FastAPI 라우터가 직접 OpenAI API를 호출하지 않고
    이 서비스 클래스를 거치도록 만들면 코드 역할이 명확해집니다.
    """

    def __init__(self, settings: Settings):
        """
        서비스 초기화 함수입니다.

        settings:
            .env 파일에서 읽은 OpenAI API Key, 모델명, 토큰 설정 등을 포함합니다.
        """

        # 설정 객체를 인스턴스 변수에 저장합니다.
        self.settings = settings

        # OpenAI API Key가 비어 있으면 클라이언트를 만들지 않습니다.
        # 이렇게 하면 API Key 없이도 추천 로직과 UI 화면은 확인할 수 있습니다.
        self.client = OpenAI(api_key=settings.OPENAI_API_KEY) if settings.OPENAI_API_KEY else None

    def _is_reasoning_model(self, model: str) -> bool:
        """
        temperature/top_p를 지원하지 않는 reasoning 계열 모델인지 판단합니다.

        gpt-5, gpt-5.5, o1, o3, o4 계열은 일부 샘플링 파라미터를
        지원하지 않을 수 있으므로 요청 파라미터를 분기합니다.
        """

        # 모델명을 소문자로 변환합니다.
        normalized_model = model.lower()

        # reasoning 성격의 모델 prefix를 튜플로 정의합니다.
        reasoning_prefixes = ("gpt-5", "o1", "o3", "o4")

        # 모델명이 해당 prefix 중 하나로 시작하는지 반환합니다.
        return normalized_model.startswith(reasoning_prefixes)

    def build_music_prompt(self, user_message: str, detected_mood: str, recommendations: list[dict]) -> str:
        """
        OpenAI에 전달할 음악 추천용 Prompt를 구성합니다.
        """

        # 추천 음악 목록을 사람이 읽기 쉬운 문자열로 변환합니다.
        song_lines = []

        # 추천 음악을 순서대로 반복합니다.
        for index, song in enumerate(recommendations, start=1):
            # 각 음악의 제목, 가수, 감정, 이유, URL을 한 줄로 정리합니다.
            song_lines.append(
                f"{index}. 제목: {song['title']} / 가수: {song['artist']} / "
                f"감정: {song['mood']} / 추천이유: {song['reason']} / URL: {song['url']}"
            )

        # 리스트를 줄바꿈 문자열로 합칩니다.
        songs_text = "\n".join(song_lines)

        # ChatGPT에게 줄 작업 지시 Prompt를 작성합니다.
        return f"""
사용자 입력:
{user_message}

PyTorch 추천 모델이 추정한 대표 감정:
{detected_mood}

PyTorch 추천 모델이 선택한 음악 목록:
{songs_text}

위 정보를 바탕으로 사용자에게 음악 추천 챗봇처럼 답변하시오.

답변 규칙:
1. 먼저 사용자의 감정을 공감하시오.
2. 추천 음악 3개를 번호 목록으로 제시하시오.
3. 각 음악마다 추천 이유를 짧게 설명하시오.
4. URL은 그대로 포함하시오.
5. 저작권이 있는 음악 또는 영상은 원저작자와 플랫폼 정책을 확인해야 한다는 안내를 마지막에 한 문장으로 포함하시오.
"""

    def generate_music_answer(
        self,
        user_message: str,
        detected_mood: str,
        recommendations: list[dict],
        conversation_history: list[dict] | None = None,
    ) -> str:
        """
        OpenAI API로 자연어 음악 추천 답변을 생성합니다.

        API Key가 없는 경우에도 앱 실습이 가능하도록
        PyTorch 추천 결과만으로 기본 답변을 생성하는 fallback을 제공합니다.
        """

        # OpenAI API Key가 없거나 클라이언트가 생성되지 않았으면 fallback 답변을 반환합니다.
        if self.client is None:
            return self._fallback_answer(user_message, detected_mood, recommendations)

        # OpenAI에 전달할 사용자 Prompt를 생성합니다.
        prompt = self.build_music_prompt(user_message, detected_mood, recommendations)

        # 시스템 지시문을 작성합니다.
        instructions = """
당신은 사용자의 감정을 이해하고 상황에 맞는 음악을 추천하는 친절한 음악 추천 챗봇입니다.
답변은 한국어로 작성합니다.
사용자가 힘든 감정을 표현하면 과장하지 말고 따뜻하게 공감합니다.
추천 결과는 PyTorch 추천 모델의 결과를 우선 사용합니다.
"""

        # OpenAI Responses API 요청 파라미터의 공통 부분을 구성합니다.
        request_params = {
            "model": self.settings.OPENAI_MODEL,
            "instructions": instructions,
            "input": prompt,
            "max_output_tokens": self.settings.OPENAI_MAX_OUTPUT_TOKENS,
        }

        # reasoning 모델이면 temperature/top_p를 보내지 않습니다.
        if self._is_reasoning_model(self.settings.OPENAI_MODEL):
            # reasoning 모델에는 추론 노력 수준을 낮게 지정하여 응답 속도를 높입니다.
            request_params["reasoning"] = {"effort": "low"}
        else:
            # 일반 모델은 temperature와 top_p를 함께 전달할 수 있습니다.
            request_params["temperature"] = self.settings.OPENAI_TEMPERATURE
            request_params["top_p"] = self.settings.OPENAI_TOP_P

        # OpenAI Responses API를 호출합니다.
        response = self.client.responses.create(**request_params)

        # SDK가 제공하는 output_text 속성에서 최종 텍스트를 반환합니다.
        return response.output_text

    def _fallback_answer(self, user_message: str, detected_mood: str, recommendations: list[dict]) -> str:
        """
        OpenAI API Key가 없을 때 사용할 기본 답변 생성 함수입니다.

        이 함수는 외부 API를 호출하지 않고 추천 결과를 문자열로 조합합니다.
        """

        # 답변의 첫 문장을 만듭니다.
        lines = [
            f"입력하신 문장에서 '{detected_mood}' 분위기가 느껴집니다.",
            "아래 음악을 추천합니다.",
            "",
        ]

        # 추천 음악을 번호 목록으로 추가합니다.
        for index, song in enumerate(recommendations, start=1):
            lines.append(f"{index}. {song['title']} - {song['artist']}")
            lines.append(f"   추천 이유: {song['reason']}")
            lines.append(f"   URL: {song['url']}")
            lines.append("")

        # 저작권 안내 문장을 마지막에 추가합니다.
        lines.append("음악과 영상 URL을 사용할 때는 원저작자와 플랫폼의 저작권 정책을 확인해야 합니다.")

        # 줄바꿈으로 연결해 최종 답변을 반환합니다.
        return "\n".join(lines)

 

 

 

 

chat_router.py

import 

Path, 경로

fastapi APIRouter, Depends API 들을 하나의 그룹 라우터로 묶는 클래스, 의존성 주입해 같은 객체를 여러 api에서 사용할때 매번 생성하지않도 자동으로 전달받음

app.core에서 config, app.recommendermusic_recommender, app.schemas의 chat_schema, app.services.의 openai_service를 가져왔다. 

현재파일기준으로 path지정, router 객체 생성, 추천모델 객체 생성해 변수 musuc_recommender에 할당

from pathlib import Path

from fastapi import APIRouter, Depends

from app.core.config import Settings, get_settings
from app.recommender.music_recommender import MusicRecommender
from app.schemas.chat_schema import MusicChatRequest, MusicChatResponse, RecommendedSong
from app.services.openai_service import OpenAIChatService


# 현재 파일 기준으로 프로젝트의 app/data/music_dataset.json 경로를 계산합니다.
DATASET_PATH = Path(__file__).resolve().parents[1] / "data" / "music_dataset.json"

# FastAPI 라우터 객체를 생성합니다.
router = APIRouter(prefix="/api/chat", tags=["Music Chatbot"])

# PyTorch 추천 모델은 요청마다 새로 만들 필요가 없으므로 모듈 로딩 시 1회 생성합니다.
music_recommender = MusicRecommender(DATASET_PATH)

 

 

get_openai_service()

OpenAIChatService 객체 return.

@router.post /music일때 

chat_music() 

music_recommender.recommend 매개변수를 활용해 대표감정, 추천음악 목록을 구한다. 

대화기록을 정리.

openai_service.generate_music_answer() 답변생성. 

response_recommendations 응답스키마에 맞게 변환.

MusicChatResponse 최종 객체 반환

@router.get("/songs")

list_songs() 현재 등록된 음악 추천데이터 전체를 반환. 

from pathlib import Path

from fastapi import APIRouter, Depends

from app.core.config import Settings, get_settings
from app.recommender.music_recommender import MusicRecommender
from app.schemas.chat_schema import MusicChatRequest, MusicChatResponse, RecommendedSong
from app.services.openai_service import OpenAIChatService


# 현재 파일 기준으로 프로젝트의 app/data/music_dataset.json 경로를 계산합니다.
DATASET_PATH = Path(__file__).resolve().parents[1] / "data" / "music_dataset.json"

# FastAPI 라우터 객체를 생성합니다.
router = APIRouter(prefix="/api/chat", tags=["Music Chatbot"])

# PyTorch 추천 모델은 요청마다 새로 만들 필요가 없으므로 모듈 로딩 시 1회 생성합니다.
music_recommender = MusicRecommender(DATASET_PATH)


def get_openai_service(settings: Settings = Depends(get_settings)) -> OpenAIChatService:
    """
    OpenAIChatService 의존성 주입 함수입니다.

    FastAPI의 Depends 기능을 사용하면 라우터 함수에서 필요한 객체를
    깔끔하게 전달받을 수 있습니다.
    """
    return OpenAIChatService(settings)


@router.post("/music", response_model=MusicChatResponse)
def chat_music(
    request: MusicChatRequest,
    openai_service: OpenAIChatService = Depends(get_openai_service),
):
    """
    사용자 입력을 받아 PyTorch 추천 결과와 OpenAI 자연어 답변을 함께 반환합니다.
    """

    # PyTorch 추천 모델로 대표 감정과 추천 음악 목록을 구합니다.
    detected_mood, recommendations = music_recommender.recommend(
        user_message=request.message,
        top_k=request.top_k,
    )

    # Pydantic 모델에서 일반 dict 리스트로 변환하기 위해 대화 기록을 정리합니다.
    conversation_history = [message.model_dump() for message in request.conversation_history]

    # OpenAI ChatGPT API로 자연스러운 챗봇 답변을 생성합니다.
    answer = openai_service.generate_music_answer(
        user_message=request.message,
        detected_mood=detected_mood,
        recommendations=recommendations,
        conversation_history=conversation_history,
    )

    # 추천 음악 리스트를 응답 스키마에 맞게 변환합니다.
    response_recommendations = [
        RecommendedSong(
            title=song["title"],
            artist=song["artist"],
            mood=song["mood"],
            reason=song["reason"],
            url=song["url"],
            score=song["score"],
        )
        for song in recommendations
    ]

    # 최종 응답 객체를 반환합니다.
    return MusicChatResponse(
        answer=answer,
        detected_mood=detected_mood,
        recommendations=response_recommendations,
    )


@router.get("/songs")
def list_songs():
    """
    현재 앱에 등록된 음악 추천 데이터 전체를 반환합니다.
    """
    return {
        "count": len(music_recommender.songs),
        "songs": music_recommender.songs,
    }

 

 

 

 

 

 

main.py

import

fastapi FastAPI, Request  FastAPI 애플리케이션 객체를 생성하는 클래스, Request 클라이언트가 보낸 HTTP 요청 정보를 담는 객체

fastapi.responses HTMLResponse 기본적으로 json을 반환하는 Fastapi이지만 html페이지를 반환하고 싶을때 

fastapi.staticfiles staticfiles 정적파일을 제공하는 기능

fastapi.templating Jinja2Templates html 템플릿을 렌더링하는 클래스, python데이터와 html을 연결하는 템플릿엔진 jinja2

app.api의 chat_router와 cpp.core의 config를 가져왔다. 

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from app.api.chat_router import router as chat_router
from app.core.config import get_settings

 

 

settings env 설정값 불러오기 

fastapi 객체 설정하기

app.mount() 정적 파일 제공하도록 설정하기

Jinja2Templates() html파일 폴더 지정하기

include_router() 라우터 앱에 등록하기

@app.get("/") 사용자가 메인일때 return templates.TemplateResponse(). index.html를 반환한다. 

@app.get("/api/health") 서버가 정상이라고 알려주기

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from app.api.chat_router import router as chat_router
from app.core.config import get_settings


# .env 설정값을 읽어옵니다.
settings = get_settings()

# FastAPI 앱 객체를 생성합니다.
app = FastAPI(
    title=settings.APP_TITLE,
    description="OpenAI ChatGPT API와 PyTorch 추천 모델을 사용하는 음악 추천 챗봇",
    version="1.0.0",
)

# /static URL로 CSS, JavaScript 같은 정적 파일을 제공하도록 설정합니다.
app.mount("/static", StaticFiles(directory="app/static"), name="static")

# HTML 템플릿 파일이 있는 폴더를 지정합니다.
templates = Jinja2Templates(directory="app/templates")

# 음악 추천 챗봇 API 라우터를 앱에 등록합니다.
app.include_router(chat_router)


@app.get("/", response_class=HTMLResponse)
def index(request: Request):
    """
    브라우저에서 접속했을 때 채팅 UI 페이지를 반환합니다.
    """
    return templates.TemplateResponse(
        "index.html",
        {
            "request": request,
            "app_title": settings.APP_TITLE,
        },
    )


@app.get("/api/health")
def health_check():
    """
    서버가 정상 실행 중인지 확인하는 헬스 체크 API입니다.
    """
    return {
        "status": "ok",
        "message": "Music recommendation chatbot server is running.",
    }