온프레미스 LLM+RAG 시스템

# 주요 기술 구성
LLM       : Bllossom-8B # (LoRA 파인튜닝)
Serving   : vLLM        # (SamplingParams 최적화)
Vector DB : FAISS       
Embedding : jhgan-ko-sroberta-multitask-strans
Framework : LangChain
UI        : Streamlit
Deployment: 온프레미스 서버

# 시도 1: 8-bit 양자화
bnb_config = BitsAndBytesConfig(
    load_in_8bit=True,
    llm_int8_threshold=6.0,
    llm_int8_skip_modules=None,
    llm_int8_enable_fp32_cpu_offload=False
)
# 시도 2: GPU 메모리 분산 처리
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    device_map="auto",  # 자동 메모리 분산
    quantization_config=bnb_config
)
# accelerate 라이브러리 활용 모델 샤딩

@st.cache_resource
def load_qna_model():
    tokenizer = AutoTokenizer.from_pretrained(VLLM_MODEL_PATH)
    tokenizer.pad_token = tokenizer.eos_token
    tokenizer.padding_side = 'right'

    llm = LLM(
        model=VLLM_MODEL_PATH,
        dtype="float16",
        max_model_len=2048,         # VRAM 효율적 활용을 위한 최적 토큰 길이
        gpu_memory_utilization=0.4,  # 다른 프로세스와 안정적 공유 (0.6까지 가능하지만 안정성 고려)
        enforce_eager=True,         # Streamlit 동기 처리로 사용자 만족도 개선
    )
    return tokenizer, llm

# 추론 설정 최적화
sampling = SamplingParams(
    stop_token_ids=[tokenizer.eos_token_id],
    temperature=0.7,
    top_p=0.9,
    max_tokens=1024,    # 답변 길이 최적화로 낭비 방지
)

RULE_FILTER: list = [
    {
      "rule_type": "blocked",
      "keywords": ["blocked keywords"],
      "reason": "금지된 내용입니다.",
      "response": "해당 질문은 처리할 수 없습니다."
    }
  ]

def load_filters() -> dict:
    with open(RULE_FILTER_PATH, "r", encoding="utf-8") as f:
        return json.load(f)

def filtering_by_rules(user_input: str) -> tuple[bool, str | None, str | None]:
    rules = load_filters()
    lowered_input = user_input.lower()
    
    for rule in rules:
        rule_type = rule.get("rule_type")
        
        for keyword in rule.get("keywords", []):
            if keyword in lowered_input:
                reason = rule.get("reason", "")
                response = rule.get("response", None)
                return FilterResult(
                    passed=False,
                    reason=reason,
                    response=response,
                    rule_type=rule_type
                )
    return FilterResult(passed=True)        

# 2-1. SentenceTransformer 에 Embed 모델 적용
def load_similarity_classifier() -> tuple:
    model = SentenceTransformer(EMBED_MODEL_PATH)   

    with open(SEMANTIC_CLF_EX_PATH, "r", encoding="UTF-8") as f:
        examples = json.load(f)

    example_embeddings = {
        label: model.encode(sentences) for label, sentences in examples.items()
    }

    return model, example_embeddings

#2-2. 질문 -> 임베딩 -> 예시와 비교, 유사도 분류
def classify_by_similarity(
    user_input: str, 
    model: SentenceTransformer, 
    example_embeddings: Dict[str, np.ndarray]) -> str:
    input_embedding = model.encode([user_input])
    similarities = {
        label: np.mean(cosine_similarity(input_embedding, embs))\
             for label, embs in example_embeddings.items()
    }
    
    # Debug
    logger.debug(f"user input: {user_input}")
    for label, score in similarities.items():
        logger.debug(f"DEBUG {label} 유사도: {score:.4f}")

    best_label = max(similarities, key=similarities.get)
    return best_label

# 2-3. 분류 모델
def classifier_system_prompt(user_input: str) -> str:
    #키워드 명시
    domain_general_keywords = ["기본개념", "일반상식", "용어설명"],
    domain_specific_keywords = ["전문용어", "특화정보", "세부규정"]
    
    # 키워드 매칭 시도
    if contains_domain_general and contains_domain_specific:
        return "domain_specific"
    elif contains_domain_specific:
        return "domain_specific"
    elif contains_domain_general:
        return "domain_general"
    
    # 2-2. SentenceTransformer 기반 2차 분류
    # 각 카테고리별 30개씩 구성된 질문 사전(총 90개)과 유사도 비교
    model, example_embeddings = load_similarity_classifier()
    return classify_by_similarity(user_input, model, example_embeddings)

# 3-1. 일반 질문에 대한 답변 명시
def general(request: PromptRequest):
    generated_prompt = generate_prompt(
        classification_result=request.classification_result,
        user_input=request.user_input,
        context=request.context,
        tokenizer=request.tokenizer
    )
    sampling = SamplingParams(
        stop_token_ids=[request.tokenizer.eos_token_id],
        temperature=0.3,
        top_p=0.7,
        max_tokens=64,
    )
    outputs = request.llm.generate(generated_prompt, sampling)
    answer = "".join([o.outputs[0].text for o in outputs])
    
    # Trained 된 모델에 대해서 응답 처리
    section_pattern = r"관련 X:.*"
    answer = re.sub(section_pattern, "", answer).strip()
    
    if len(answer) < 5:
        answer = "질문과 관련된 정보를 찾을 수 없습니다. 다른 질문을 해주세요."
        
    return generated_prompt, answer

# 3-2. 도메인 일반 / 특화 된 부분에 따라서 차별 응답
def generate_llm_response(request: PromptRequest, docs: list = None):
    if request.classification_result == "general":
        return general(request)
    
    prompt, answer = detail(request)
        
    if request.classification_result in ["domain_specific", "domain_general"]:
        answer = format_response(answer)
        
    return prompt, answer

# 다양한 청킹 전략 테스트
chunk_size_tests = [100, 300, 500, 1000]
processing_methods = ["PDF_OCR", "PDF_direct", "Markdown_conversion"]

# 최적 조합 탐색
for chunk_size in chunk_size_tests:
    for method in processing_methods:
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=200,  # 고정값
            separators=["\n\n", "\n", " "]
        )
        # 육안 검토를 통한 완성도 평가
        evaluate_chunk_quality(splitter, method)

def get_text_splitter(chunk_size=500, chunk_overlap=200):
    return RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", " "],
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        is_separator_regex=False
    )

# FAISS 벡터 데이터베이스 구성
# 한국어 특화 임베딩 모델 사용
embed_model = HuggingFaceEmbeddings(
    model_name='../models/jhgan/ko-sroberta-multitask-strans',
    model_kwargs={"device": "cuda:0"},
    encode_kwargs={"normalize_embeddings": True},
)

vector_store = FAISS.from_documents(
    split_documents,
    embedding=embed_model,
    distance_strategy=DistanceStrategy.COSINE
)

# 검색 최적화 -> faiss - merge - context로 반환
def run_retriever(user_input: str, classification_result: str):
    if classification_result == "general":
        # do not run retriever
        return [], ""
    
    retriever = get_faiss_retriever()
    docs = retriever.invoke(user_input)
    context = merge_docs(docs=docs)
    return docs, context

# Streamlit 기반 통합 서비스
import streamlit as st
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 비동기 처리 환경 설정
try:
    asyncio.get_running_loop()
except RuntimeError:
    asyncio.set_event_loop(asyncio.new_event_loop())

# 서비스 구성요소
def main()
    try:
        # wordcloud service
        if is_wordcloud_request(user_input=user_input):
            with st.spinner("워드클라우드 생성 중"):
                handle_wordcloud_request(role="assistant")
            st.success("워드클라우드 생성 완료")
            return
        
        # filtering by rules (passed, reason, response, rule_type)
        filtered_result = filtering_by_rules(user_input=user_input)

        if not filtered_result.passed: # filtered: No RAG, No LLM
            if filtered_result.response: # 답변이 있을 시
                st.chat_message("assistant").write(response)
                add_message("assistant", response)
            st.warning(f"[오류] {reason}") # 차단 내용 설명
            
            return # 답변 끝
        
        # classifier
        classification_result = classifier_system_prompt(user_input=user_input)
        
        # RAG
        docs, context = run_retriever(user_input=user_input,
                                      classification_result=classification_result)
        
        # LLM
        tokenizer, llm = load_qna_model() # load model and save in cache
        
        request = PromptRequest(
            classification_result=classification_result,
            user_input=user_input,
            context=context,
            tokenizer=tokenizer,
            llm=llm
        )
        
        prompt, answer = generate_llm_response(request=request, docs=docs)
        st.chat_message("assistant").markdown(answer) # answer format in markdown
        add_message("assistant", answer)

# 실시간 디버깅 UI 구현
def debug_interface():
    #Debug
    st.markdown(f"# 질문: {user_input}")
    
    add_message("user", user_input)
    
    try:
        # wordcloud service
        if is_wordcloud_request(user_input=user_input):
            with st.spinner("워드클라우드 생성 중"):
                handle_wordcloud_request(role="assistant")
            st.success("워드클라우드 생성 완료")
            return
        
        # filtering by rules (passed, reason, response, rule_type)
        filtered_result = filtering_by_rules(user_input=user_input)

        if not filtered_result.passed: # filtered: No RAG, No LLM
            if filtered_result.response: # 답변이 있을 시
                st.chat_message("assistant").write(response)
                add_message("assistant", response)
            st.warning(f"[오류] {reason}") # 차단 내용 설명
            
            return # 답변 끝
        
        # classifier
        classification_result = classifier_system_prompt(user_input=user_input)
        
        st.markdown(f"## classification_result: {classification_result}")
        # RAG
        docs, context = run_retriever(user_input=user_input,
                                      classification_result=classification_result)
        
        st.markdown(f"#### 검색된 문서 수 : {len(docs)}")
        st.markdown("### 검색 문서 전체 내용")
        
        if docs: # not empty
            for i, doc in enumerate(docs, 1):
                st.markdown(f"**문서 {i}**")
                st.code(doc.page_content)
        else:
            st.markdown("RAG가 적용되지 않았습니다.")
            st.markdown(f"prompt type: {classification_result}")
        
        # LLM
        tokenizer, llm = load_qna_model() # load model and save in cache
        
        request = PromptRequest(
            classification_result=classification_result,
            user_input=user_input,
            context=context,
            tokenizer=tokenizer,
            llm=llm
        )
        
        prompt, answer = generate_llm_response(request=request, docs=docs)
        
        st.markdown("### 생성된 프롬프트")
        st.code(prompt[:500])

        st.markdown("### LLM 응답")
        st.chat_message("assistant").markdown(answer) # answer format in markdown
        add_message("assistant", answer)

# LoRA 설정 최적화
config = LoraConfig(
    r=8,                    # 랭크 - 메모리와 성능 간 균형
    alpha=32,               # 스케일링 팩터
    dropout=0.1,            # 오버피팅 방지
    target_modules=[
        "q_proj", "o_proj", "k_proj", "v_proj",  # 어텐션 레이어
        "gate_proj", "up_proj", "down_proj"       # MLP 레이어
    ],
    bias="none",
    task_type="CAUSAL_LM"
)

# 훈련 설정
trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    args=TrainingArguments(
        output_dir="outputs",           # 학습 결과 저장 경로
        #num_train_epochs=1,            # 학습 에폭 수
        max_steps=300,                  # 최대 학습 스텝 수 // check: 300 250318
        per_device_train_batch_size=16, # 배치 크기 설정
        gradient_accumulation_steps=4,  # 그래디언트 누적 단계 수
        optim="paged_adamw_8bit",       # 8bit 최적화 알고리즘
        warmup_steps=0,                 # 워밍업 스텝 비활성화
        learning_rate=2e-4,             # 학습률 설정
        fp16=True,                      # 16비트 연산 활성화 (메모리 사용량 감소)
        logging_steps=100,              # 로그 출력 간격
        push_to_hub=False,              # Hugging Face Hub 푸시 비활성화
        report_to='none',               # 로그 리포팅 비활성화
    ),
    peft_config=lora_config,            # LoRA 설정 전달
    formatting_func=prompts,            # 프롬프트 포맷 함수 적용
)