import gradio as gr from huggingface_hub import InferenceClient from litellm import completion from dotenv import load_dotenv import json import re load_dotenv() # ------------------------------- # 1) 고정 10개 질문 정의 # ------------------------------- FIXED_QUESTIONS = [ "Q1) 본인의 전공/업무 분야는 무엇인가요?", "Q2) 최근 가장 집중한 프로젝트는 무엇이었나요?", "Q3) 해당 프로젝트에서 가장 어려웠던 점은 무엇이었나요?", "Q4) 즐겨 쓰는 개발 스택(언어/프레임워크/라이브러리)을 알려주세요.", "Q5) 협업 시 가장 중요하게 생각하는 원칙은 무엇인가요?", "Q6) 성능 개선을 위해 가장 자주 사용하는 방법은 무엇인가요?", "Q7) 테스트/검증을 어떤 방식으로 진행하나요?", "Q8) 데이터/리소스가 제한될 때 어떤 전략을 쓰시나요?", "Q9) 최근 배운 것 중 가장 유용했던 내용은 무엇이었나요?", "Q10) 앞으로 다뤄보고 싶은 주제나 기술이 있나요?", ] # ------------------------------- # 2) 꼬리질문 20개 생성 프롬프트 # (고정 10문답을 바탕으로 추출) # ------------------------------- FOLLOWUP_SYSTEM = ( "You are an excellent interviewer. Based on the given 10 Q/A pairs, " "generate 20 SHORT, concrete, non-overlapping follow-up questions that deeply probe the user's answers. " "Each question should be standalone and specific. Output as a numbered list 1..20." ) def build_followup_user_prompt(qa_pairs): """ qa_pairs: list[tuple(question, answer)] """ lines = ["Below are 10 Q/A pairs. Generate 20 short follow-up questions.\n"] for i, (q, a) in enumerate(qa_pairs, 1): lines.append(f"[Q{i}] {q}") lines.append(f"[A{i}] {a}\n") lines.append("Return only the 20 questions as a numbered list (1..20).") return "\n".join(lines) def parse_numbered_list_to_lines(text, expected_n=20): # 1) Remove code fences text = re.sub(r"^```.*?\n|\n```$", "", text, flags=re.DOTALL).strip() # 2) Split lines on numbering # e.g. "1) ..." or "1. ..." or "1 - ..." etc candidates = re.split(r"(?:^\s*\d+\s*[\)\.\-\:]\s*)", text, flags=re.MULTILINE) # The split keeps text fragments; we need to reassemble meaningful lines. # An easier approach is to capture lines that start with a number: lines = re.findall(r"^\s*\d+\s*[\)\.\-\:]\s*(.+)$", text, flags=re.MULTILINE) lines = [l.strip() for l in lines if l.strip()] # Fallback: if no pattern matched, split by newline and filter bullets if not lines: for raw in text.splitlines(): s = raw.strip() if s and not s.startswith("#"): lines.append(s) # Trim to expected_n if overshoot; if undershoot, keep whatever we have return lines[:expected_n] # -------------------------------- # 3) 메인 응답 함수 # -------------------------------- def respond( message, history: list[dict[str, str]], system_message, max_tokens, temperature, top_p, # (OAuth 버튼은 유지하되, 아래 구현에서는 사용하지 않음) hf_token: gr.OAuthToken, # --- 상태 값들 --- phase, # 1 -> 2 -> 3 asked, # 직전에 질문을 던졌는지 여부 (True면 이번 사용자의 입력은 답변으로 간주) i1, i2, i3, # 각 단계 인덱스 gen_questions,# 2단계에서 사용할 20개 질문 (list[str]) fixed_answers,# 1단계 답변(10개 저장용) gen_answers, # 2단계 답변(20개 저장용) rep_answers, # 3단계 답변(10개 저장용) ): """ 대화 흐름: - asked == False: 이번 호출에서는 '다음 질문'을 내보내고 asked=True 로 전환 - asked == True : 이번 호출의 message를 '답변'으로 저장하고 인덱스를 증가시킨 뒤, 다음 질문을 내보내며 asked=True 유지 """ model = "gemini/gemini-2.5-flash" # 최초 진입(사용자 첫 메시지): 질문을 던질 차례로 맞춘다. if phase is None: phase = 1 if asked is None: asked = False if i1 is None: i1 = 0 if i2 is None: i2 = 0 if i3 is None: i3 = 0 if gen_questions is None: gen_questions = [] if fixed_answers is None: fixed_answers = [] if gen_answers is None: gen_answers = [] if rep_answers is None: rep_answers = [] # 헬퍼: 현재 단계에서 "다음 질문 텍스트"를 리턴 def next_question(): nonlocal phase, i1, i2, i3, gen_questions if phase == 1: return FIXED_QUESTIONS[i1] if i1 < len(FIXED_QUESTIONS) else None elif phase == 2: return gen_questions[i2] if i2 < len(gen_questions) else None elif phase == 3: return FIXED_QUESTIONS[i3] if i3 < len(FIXED_QUESTIONS) else None return None # 헬퍼: 1→2 단계 전환 시 꼬리질문 20개 생성 def ensure_followups(): nonlocal gen_questions if gen_questions: return # 이미 생성됨 # 1단계 Q/A 페어 구성 qa_pairs = list(zip(FIXED_QUESTIONS, fixed_answers)) user_prompt = build_followup_user_prompt(qa_pairs) followup_msgs = [ {"role": "system", "content": FOLLOWUP_SYSTEM}, {"role": "user", "content": user_prompt}, ] res = completion( model=model, messages=followup_msgs, temperature=temperature, top_p=top_p, # max_tokens=max_tokens, # 필요 시 해제 ) res_json = res.choices[0].message.model_dump() followup_text = res_json["content"].strip() gen_questions = parse_numbered_list_to_lines(followup_text, expected_n=20) # 혹시 20개 미만이면 보충(간단한 백업) while len(gen_questions) < 20: gen_questions.append(f"(추가) 관심 주제에 대해 더 자세히 설명해 주실 수 있나요? [{len(gen_questions)+1}]") # -------------------------------- # (A) asked == False → 질문 던지기 # -------------------------------- if not asked: if phase == 1 and i1 == 0: intro = ( "안녕하세요! 다음 순서로 진행할게요:\n" "1) 고정 10문항에 먼저 답변합니다.\n" "2) 이어서 LLM이 방금 답변을 바탕으로 20문항을 생성해 질문합니다.\n" "3) 마지막으로 처음의 10문항을 다시 묻습니다.\n\n" "그럼 시작하겠습니다!" ) # 첫 안내 후 첫 질문 q = next_question() asked = True yield f"{intro}\n\n{q}", phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return # 그 외 일반 케이스: 다음 질문 q = next_question() if q is not None: asked = True yield q, phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return # 질문이 더 없다면(모든 단계 완료) yield "모든 질문이 완료되었습니다. 참여해 주셔서 감사합니다! 🎉", phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return # -------------------------------- # (B) asked == True → 방금 받은 message를 답변으로 저장하고 다음 질문 # -------------------------------- if asked: if phase == 1: # 1단계 답변 저장 fixed_answers.append(message) i1 += 1 asked = False # 다음 턴에는 질문을 내보내도록 if i1 >= len(FIXED_QUESTIONS): # 2단계로 이동: 꼬리질문 20개 생성 phase = 2 ensure_followups() # 곧바로 다음 질문 던지기 q = next_question() asked = True yield f"좋습니다. 1단계를 마쳤습니다. 이제 2단계(20문항)로 넘어갈게요.\n\n{q}", phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return else: # 다음 1단계 질문 q = next_question() asked = True yield q, phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return elif phase == 2: # 2단계 답변 저장 gen_answers.append(message) i2 += 1 asked = False if i2 >= len(gen_questions): # 3단계로 이동 phase = 3 q = next_question() asked = True yield f"좋아요. 2단계를 마쳤습니다. 마지막으로 1단계의 10문항을 다시 묻겠습니다.\n\n{q}", phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return else: q = next_question() asked = True yield q, phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return elif phase == 3: # 3단계 답변 저장 rep_answers.append(message) i3 += 1 asked = False if i3 >= len(FIXED_QUESTIONS): # 완료 summary = { "phase1_fixed": [{"q": FIXED_QUESTIONS[i], "a": fixed_answers[i]} for i in range(len(fixed_answers))], "phase2_generated": [{"q": gen_questions[i], "a": gen_answers[i]} for i in range(len(gen_answers))], "phase3_repeat": [{"q": FIXED_QUESTIONS[i], "a": rep_answers[i]} for i in range(len(rep_answers))], } done_text = ( "모든 질문이 완료되었습니다. 참여에 감사드립니다! 🎉\n" "필요하시다면 아래 JSON 요약을 복사해가세요.\n\n" + "```json\n" + json.dumps(summary, ensure_ascii=False, indent=2) + "\n```" ) yield done_text, phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return else: q = next_question() asked = True yield q, phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers return # 안전망 (도달하지 않아야 함) yield "상태 전환 중 예기치 못한 상황이 발생했습니다. 다시 시도해 주세요.", phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers # -------------------------------- # 4) ChatInterface 구성 # -------------------------------- with gr.Blocks() as demo: with gr.Sidebar(): # 로그인 버튼 유지 (hf_token은 현재 구현에서 사용하지 않지만 UI는 그대로 둠) oauth = gr.LoginButton() gr.Markdown( "### 진행 순서\n" "1) 고정 10문항에 먼저 답하기\n" "2) LLM이 생성한 20문항 꼬리질문에 답하기\n" "3) 고정 10문항을 다시 한 번 답하기" ) phase = gr.State(1) asked = gr.State(False) i1 = gr.State(0) i2 = gr.State(0) i3 = gr.State(0) gen_questions= gr.State([]) fixed_answers= gr.State([]) gen_answers = gr.State([]) rep_answers = gr.State([]) # system/max_tokens/temperature/top_p 입력 컴포넌트도 블록 안에서 만들고 넘깁니다 sys_msg = gr.Textbox(value="You are a friendly Chatbot.", label="System message") max_toks = gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens") temp = 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)") chatbot = gr.ChatInterface( respond, type="messages", # ✅ 파라미터 순서: system_message, max_tokens, temperature, top_p, hf_token, (states…) additional_inputs=[ sys_msg, max_toks, temp, top_p, oauth, # <-- ✅ hf_token 자리에 놓기! phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers, ], # ✅ 출력에도 동일한 State 인스턴스 재사용 additional_outputs=[ phase, asked, i1, i2, i3, gen_questions, fixed_answers, gen_answers, rep_answers, ], ) chatbot.render() if __name__ == "__main__": demo.launch()