init

.gitignore

@@ -0,0 +1,4 @@
+__pycache__
+cvdata/
+.vscode/
+.venv/

.python-version

@@ -0,0 +1 @@
+3.12

README.md

@@ -1,13 +1,15 @@
 ---
-title: Cvchatbot
-emoji: 🏃
-colorFrom: red
+title: cvchatbot
+emoji: 💬
+colorFrom: yellow
 colorTo: purple
 sdk: gradio
-sdk_version: 6.5.1
-python_version: '3.12'
+sdk_version: 6.3.0
+python_version: "3.12"
 app_file: app.py
 pinned: false
+hf_oauth: true
+hf_oauth_scopes:
+- inference-api
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,80 @@
+import argparse
+import logging
+import gradio as gr
+from huggingface_hub import login
+import spaces
+
+
+parser = argparse.ArgumentParser("cvchatbot")
+parser.add_argument("--debug", help="Enable debug logging", action='store_true')
+parser.add_argument("--generate", dest="generate_cache", help="Generate new cache", action='store_true')
+parser.add_argument("--download-cv-data", dest="download_cv_data", help="Download CV data from Cloud Storage", action='store_true')
+args = parser.parse_args()
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
+
+
+from config import CV_DATA_DIRNAME, EMBEDDING_MODEL_ID, HF_TOKEN, LLM_ID, \
+    WELCOME_MESSAGE_TEMPLATE, NO_MATCH_RESPONSE, RESET_QUERY, \
+    CV_DATA_FILE_EDUCATION, CV_DATA_FILE_JOBS, CV_DATA_FILE_PAPERS, CV_DATA_FILE_SKILLS, \
+    EDUCATION_DESCRIPTION, JOBS_DESCRIPTION, PAPERS_DESCRIPTION, SKILLS_DESCRIPTION
+from helpers import get_education_documents, get_job_documents, get_paper_documents, get_skill_documents
+from knowledge import ChatContext, DomainKnowledgeConfig, KnowledgeBase, KnowledgeConfig
+
+
+logger.info("Logging in to Hugging Face Hub...")
+login(token=HF_TOKEN)
+
+logger.info("Initializing CV Knowledge Base...")
+cv_knowledge_base = KnowledgeBase(
+    KnowledgeConfig(
+        WELCOME_MESSAGE_TEMPLATE,
+        NO_MATCH_RESPONSE,
+        RESET_QUERY,
+        [
+            DomainKnowledgeConfig(JOBS_DESCRIPTION, CV_DATA_DIRNAME + CV_DATA_FILE_JOBS, get_job_documents),
+            DomainKnowledgeConfig(SKILLS_DESCRIPTION, CV_DATA_DIRNAME + CV_DATA_FILE_SKILLS, get_skill_documents),
+            DomainKnowledgeConfig(EDUCATION_DESCRIPTION, CV_DATA_DIRNAME + CV_DATA_FILE_EDUCATION, get_education_documents),
+            DomainKnowledgeConfig(PAPERS_DESCRIPTION, CV_DATA_DIRNAME + CV_DATA_FILE_PAPERS, get_paper_documents),
+        ],
+    ),
+    args.generate_cache,
+    args.download_cv_data,
+)
+
+
+@spaces.GPU
+def respond(
+    message: str,
+    history: list[gr.MessageDict],
+    chat_context: ChatContext,
+):
+    yield from cv_knowledge_base.answer_query(message, history, chat_context)
+
+
+with gr.Blocks() as demo:
+    # login button
+    with gr.Sidebar():
+        gr.LoginButton()
+
+    # chat context
+    context_state = gr.State(ChatContext())
+    
+    # chat interface
+    gr.ChatInterface(
+        respond,
+        chatbot=gr.Chatbot(
+            label=f"{LLM_ID} + {EMBEDDING_MODEL_ID}",
+            value=[gr.MessageDict(role="assistant", content=cv_knowledge_base.knowledge.welcome_message)],
+            avatar_images=(None, "./assets/chatbot_icon.png"),
+        ),
+        textbox=gr.Textbox(
+            submit_btn=True,
+        ),
+        additional_inputs=[context_state], # Pass the updated state as an input
+        additional_outputs=[context_state], # Receive the updated state as an output
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()

config.py

@@ -0,0 +1,66 @@
+import os
+from string import Template
+
+
+HF_TOKEN = os.getenv('HF_TOKEN')
+# LLM_ID = "google/gemma-3-1b-it"
+LLM_ID = "google/gemma-3-4b-it"
+EMBEDDING_MODEL_ID = "google/embeddinggemma-300M"
+PERSON_NAME = "William"
+PERSON_PRONOUN = "he"
+
+# cv document data files
+CV_DATA_GCS_BUCKET = "wlee-jsons"
+CV_DATA_DIRNAME = "cvdata/"
+CV_DATA_FILE_EDUCATION = "education.json"
+CV_DATA_FILE_JOBS = "jobs.json"
+CV_DATA_FILE_PAPERS = "papers.json"
+CV_DATA_FILE_SKILLS = "skills.json"
+RESET_QUERY = "restart chat"
+WELCOME_MESSAGE_TEMPLATE = Template(f"Hi! I'm William's Career Assistant. I can answer questions you might have regarding William's resume. "
+                   "To start, tell me which one of the following domains you would like to focus on:\n"
+                   "$domains\n"
+                   f"At any point in chat, say \'{RESET_QUERY}\' to reset chat and return to current state.")
+NO_MATCH_RESPONSE = "I don't have that specific information right now."
+EDUCATION_DESCRIPTION = "Academic degrees and relevant coursework"
+JOBS_DESCRIPTION = "Work experience"
+PAPERS_DESCRIPTION = "Papers and publications"
+SKILLS_DESCRIPTION = "Technical skills, such as programming language, framework, and tools"
+
+# cache files
+KNOWLEDGE_CACHE_FILE = "knowledge.pkl"
+
+# embedding parameters
+DOMAIN_SELECTION_THRESHOLD = 0.5
+DOCUMENT_SELECTION_THRESHOLD = 0.2
+DOCUMENT_FOLLOWUP_THRESHOLD = 0.5
+
+# generation parameters
+RESPONSE_MAX_NEW_TOKENS = 512
+RESPONSE_SAMPLING_TOP_P = 0.95
+RESPONSE_SAMPLING_TEMPERATURE = 0.7
+
+# helper configs
+COURSE_DEPARTMENT_NAMES = {
+    "MAT": "Mathematics",
+    "ECO": "Economics",
+    "ORF": "Operations Research and Financial Engineering",
+    "COS": "Computer Science",
+    "PHI": "Philosophy",
+    "LIN": "Linguistics",
+    "AST": "Astronomy",
+    "CEE": "Civil and Environmental Engineering",
+    "PHY": "Physics",
+    "MUS": "Music",
+}
+SKILL_COMPETENCY_LEVELS = {
+    5: "Expert",
+    4: "Proficient",
+    3: "Competent",
+    2: "Advanced Beginner",
+    1: "Novice",
+}
+JOB_TYPES = [
+    "Software Engineer",
+    "Research Intern",
+]

helpers.py

@@ -0,0 +1,164 @@
+from datetime import timedelta, datetime
+from collections import defaultdict
+import json
+from typing import Tuple
+from google.cloud.storage import Client, transfer_manager
+
+from config import COURSE_DEPARTMENT_NAMES, JOB_TYPES, PERSON_NAME, PERSON_PRONOUN, RESET_QUERY, SKILL_COMPETENCY_LEVELS, \
+    CV_DATA_DIRNAME, CV_DATA_GCS_BUCKET
+
+
+def download_cv_data() -> None:
+    # download cv data from google cloud storage
+    storage_client = Client()
+    bucket = storage_client.bucket(CV_DATA_GCS_BUCKET)
+    cv_document_blobs = [blob.name for blob in bucket.list_blobs(prefix=CV_DATA_DIRNAME)]
+    results = transfer_manager.download_many_to_path(bucket, cv_document_blobs)
+    if not all(result is None for result in results):
+        exception_message = ""
+        for i, download_result in enumerate(results):
+            if download_result is not None:
+                exception_message += f"❗ Exception occurred while downloading {cv_document_blobs[i]} from {CV_DATA_GCS_BUCKET}: {download_result}; "
+        
+        raise RuntimeError(exception_message)
+
+def get_job_type(position: str) -> str:
+    position_lower = position.lower()
+    for job_type in JOB_TYPES:
+        if job_type.lower() in position_lower:
+            return job_type
+    raise ValueError(f"Position '{position}' does not match any known job types.")
+
+def join_grammatically(words: list[str]) -> str:
+    if len(words) == 0:
+        raise ValueError("words must contain at least one word.")
+    elif len(words) == 1:
+        return words[0]
+    return ", ".join(words[:-2] + [" and ".join(words[-2:])])
+
+def td_to_yoe_str(td: timedelta) -> str:
+    secs_in_year = 365.2425 * 24 * 3600
+    td_secs = td.total_seconds()
+    if td_secs < secs_in_year:
+        secs_in_month = 30.436875 * 24 * 3600
+        months = round(td_secs / secs_in_month)
+        return f"{months} {'month' if months <= 1 else 'months'}"
+    else:
+        return f"{td_secs / secs_in_year:.1f} years"
+
+def get_job_documents(jobs_json: dict) -> Tuple[str, list[str]]:
+    jobs_docs = []
+    positions_in_order = []
+    yoe_by_job_type = defaultdict(timedelta)
+    for job_json in jobs_json["jobs"].values():
+        company_word = f"[{job_json["company"]["name"]}]({job_json["company"]["link"]}) in {job_json["company"]["location"]}"
+        for position in job_json["positions"]:
+            start_date = datetime.strptime(position["startDate"], "%Y-%m-%d")
+            end_date = datetime.strptime(position["endDate"], "%Y-%m-%d")
+            yoe_by_job_type[get_job_type(position["name"])] += end_date - start_date
+
+            job_doc = f"{PERSON_NAME} worked as {position["name"]} at {company_word} from {start_date.strftime('%B %Y')} to {end_date.strftime('%B %Y')}."
+            job_doc += f" Following are highlights of this position: {'; '.join(position['highlights'])}."
+            jobs_docs.append(job_doc)
+
+            positions_in_order.append(f"➢ {position["name"]} at {company_word} ({start_date.strftime('%B %Y')} - {end_date.strftime('%B %Y')})")
+    
+    yoe_by_job_type_strs = [f"{td_to_yoe_str(yoe)} of experience as {job_type}" for job_type, yoe in yoe_by_job_type.items()]
+    intro_message = f"{PERSON_NAME} has {join_grammatically(yoe_by_job_type_strs)}. Following are positions {PERSON_PRONOUN} has worked as:\n"
+    intro_message += "\n".join(positions_in_order)
+    intro_message += f"\nYou can ask me about more details on any of these positions. As a reminder, you can say \'{RESET_QUERY}\' to reset chat."
+
+    return intro_message, jobs_docs
+
+def get_education_documents(education_json: dict) -> Tuple[str, list[str]]:
+    education_docs = []
+
+    university_name = education_json["school"]["name"]
+    university_end_date = datetime.strptime(education_json["endDate"], "%Y-%m-%d").strftime("%B %Y")
+    degree_full_str = f"{education_json["degree"]["name"]}"
+    minors = [minor["name"] for minor in education_json["minors"]]
+    if len(minors) == 1:
+        degree_full_str += f" with minor in {minors[0]}"
+    elif len(minors) > 1:
+        degree_full_str += f" with minors in {", ".join(minors)}"
+
+    dept_courses = defaultdict(list[str])
+    for course_json in education_json["courses"]:
+        course_number = course_json["number"]
+        course_dept = course_number.split(" ")[0]
+        dept_courses[course_dept].append(f"➢ [{course_json["number"]}]({course_json["link"]}): {course_json["title"]}")
+
+    sorted_depts = sorted(dept_courses, key=lambda k: len(dept_courses[k]), reverse=True)
+    for dept in sorted_depts:
+        education_docs.append(f"In department of {COURSE_DEPARTMENT_NAMES[dept]}, {PERSON_NAME} took \n{"\n".join(dept_courses[dept])}.")
+    sorted_dept_names = "\n".join([f"➢ {COURSE_DEPARTMENT_NAMES[dept]} ({dept})" for dept in sorted_depts])
+
+    intro_message = (f"{PERSON_NAME} graduated from {university_name} in {university_end_date} with {degree_full_str}. "
+                     f"{PERSON_PRONOUN.title()} took {len(education_json["courses"])} courses across following departments:\n{sorted_dept_names}."
+                     f"\nYou can ask me about detailed list of courses {PERSON_NAME} took in each department. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")
+    
+    return intro_message, education_docs
+
+def get_paper_documents(papers_json: dict) -> Tuple[str, list[str]]:
+    intro_messages = []
+    paper_docs = []
+    for i, paper_json in enumerate(papers_json["papers"]):
+        paper_date = datetime.strptime(paper_json["date"], "%Y-%m")
+        intro_messages.append(f"{i + 1}: {paper_json["title"]} ({paper_date.strftime("%Y")})")
+        paper_doc = f"{i + 1}: Title of this paper is {paper_json["title"]}."
+
+        location_str = f" in {paper_json["location"]}" if "location" in paper_json else ""
+        paper_doc += f" It was published on {paper_date.strftime("%B %Y")}{location_str}."
+
+        if "conference" in paper_json:
+            paper_doc += f" It was published in {paper_json["conference"]}."
+
+        paper_doc += f" Link to paper is {paper_json["paperUrl"]}."
+
+        paper_doc += f" Following is description of this paper. {paper_json["desc"]}"
+
+        paper_docs.append(paper_doc)
+    
+    intro_message = (f"{PERSON_NAME} wrote {len(intro_messages)} papers. Here are titles of these papers:\n{"\n".join(intro_messages)}"
+                     f"\nYou can ask me about more details on any of these papers. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")
+    
+    return intro_message, paper_docs
+
+def get_skill_documents(skills_json: dict) -> Tuple[str, list[str]]:
+    skill_competency = defaultdict(int)
+    skills_by_category = defaultdict(list)
+    for skill_json in skills_json["skills"]:
+        skill_competency[skill_json["title"]] = skill_json["competency"]
+        for skill_category in skill_json["category"]:
+            skills_by_category[skill_category].append(skill_json["title"])
+    
+    skill_docs = []
+    for category, skills in skills_by_category.items():
+        skills.sort(key=lambda x: skill_competency[x], reverse=True)
+        skill_docs.append(f"Technical skills related to {category} that {PERSON_NAME} has are: {", ".join(skills)}.")
+
+    sorted_skill_categories = list(skills_by_category.keys())
+    sorted_skill_categories.sort()
+    intro_message = (f"{PERSON_NAME} has various technical skills across following categories:\n"
+                     f"{'\n'.join([f"➢ {category}" for category in sorted_skill_categories])}"
+                     f"\nYou can ask me which technical skills {PERSON_NAME} has in one of these areas. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")
+
+    return intro_message, skill_docs
+
+def print_docs(name, file_name, parse_func):
+    with open(f"cvdata/{file_name}") as f:
+        intro_message, docs = parse_func(json.load(f))
+
+    print(f'\n>>>>>>>>>> {name} <<<<<<<<<<')
+    print('========== Intro')
+    print(intro_message)
+    print('\n========== Docs')
+    for doc in docs:
+        print(doc + '\n')
+
+
+if __name__ == "__main__":
+    print_docs("Jobs", "jobs.json", get_job_documents)
+    print_docs("Education", "education.json", get_education_documents)
+    print_docs("Papers", "papers.json", get_paper_documents)
+    print_docs("Skills", "skills.json", get_skill_documents)

knowledge.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:85a4b6d3b961b41367c6f09f73a522513ba8bae2f9e3c3c08eae6c40bd97e459
+size 156587

knowledge.py

@@ -0,0 +1,304 @@
+import json
+import logging
+import os
+import pickle
+from collections.abc import Callable, Generator
+from string import Template
+from threading import Thread
+from typing import Any, Tuple
+from gradio import MessageDict
+from accelerate import Accelerator
+from sentence_transformers import SentenceTransformer
+from transformers import TextIteratorStreamer, pipeline
+import torch
+
+from config import EMBEDDING_MODEL_ID, LLM_ID, KNOWLEDGE_CACHE_FILE, \
+    DOCUMENT_FOLLOWUP_THRESHOLD, DOCUMENT_SELECTION_THRESHOLD, DOMAIN_SELECTION_THRESHOLD, \
+    RESPONSE_MAX_NEW_TOKENS, RESPONSE_SAMPLING_TEMPERATURE, RESPONSE_SAMPLING_TOP_P
+from helpers import download_cv_data
+
+logger = logging.getLogger(__name__)
+accelerator = Accelerator()
+device = accelerator.device
+
+class DomainKnowledge:
+    def __init__(
+        self, 
+        desc: str, 
+        intro_message: str,
+        documents: list[str],
+        document_embeddings: torch.Tensor,
+    ):
+        self.description = desc
+        self.intro_message = intro_message
+        self.documents = documents
+        self.document_embeddings = document_embeddings
+
+
+class DomainKnowledgeConfig:
+    def __init__(self, description, file_name: str, parse_file_fn: Callable[[Any], Tuple[str, list[str]]]):
+        self.desc = description
+        self.file_name = file_name
+        self.parse_file = parse_file_fn
+
+    def initialize(self, embedding_model: SentenceTransformer) -> DomainKnowledge:
+        with open(self.file_name) as f:
+            document_file = json.load(f)
+
+        intro_message, documents = self.parse_file(document_file)
+        document_embeddings: torch.Tensor = embedding_model.encode_document(
+            documents, 
+            convert_to_tensor=True, 
+            device=device,
+        ) # type: ignore
+        return DomainKnowledge(self.desc, intro_message, documents, document_embeddings)
+    
+
+class Knowledge:
+    def __init__(
+        self,
+        welcome_message: str,
+        no_match_response: str,
+        reset_query: str,
+        domain_knowledges: list[DomainKnowledge],
+        domain_knowledge_description_embeddings: torch.Tensor,
+    ):
+        self.welcome_message = welcome_message
+        self.no_match_response = no_match_response
+        self.reset_query = reset_query
+        self.domain_knowledges = domain_knowledges
+        self.dk_desc_embeddings = domain_knowledge_description_embeddings
+
+
+class KnowledgeConfig:
+    def __init__(
+        self,
+        welcome_message_template: Template,
+        no_match_response: str,
+        reset_query: str,
+        domain_knowledge_configs: list[DomainKnowledgeConfig],
+    ):
+        domain_descriptions = "\n".join([f"➢ {dk_cfg.desc}" for dk_cfg in domain_knowledge_configs])
+        self.welcome_message = welcome_message_template.substitute(domains=domain_descriptions)
+        self.no_match_response = no_match_response
+        self.reset_query = reset_query
+        self.domain_knowledge_configs = domain_knowledge_configs
+    
+    def initialize(
+        self,
+        embedding_model: SentenceTransformer,
+        generate_cache: bool,
+        download_data: bool,
+    ) -> Knowledge:
+        if generate_cache or not os.path.exists(KNOWLEDGE_CACHE_FILE):
+            logger.info("⚠️ Initializing cv knowledge...")
+            if download_data:
+                logger.info("⚠️ Downloading cv data...")
+                download_cv_data()
+
+            domain_knowledges = []
+            domain_knowledge_descriptions = []
+            for dk_cfg in self.domain_knowledge_configs:
+                domain_knowledge = dk_cfg.initialize(embedding_model)
+                domain_knowledges.append(domain_knowledge)
+                domain_knowledge_descriptions.append(domain_knowledge.description)
+            
+            domain_knowledge_description_embeddings: torch.Tensor = embedding_model.encode_document(
+                domain_knowledge_descriptions, 
+                convert_to_tensor=True, 
+                device=device,
+            ) # type: ignore
+
+            knowledge = Knowledge(
+                self.welcome_message,
+                self.no_match_response,
+                self.reset_query,
+                domain_knowledges,
+                domain_knowledge_description_embeddings,
+            )
+
+            with open(KNOWLEDGE_CACHE_FILE, 'wb') as f:
+                pickle.dump(knowledge, f)
+
+            logger.info(f"✅ Cached cv knowledge to '{KNOWLEDGE_CACHE_FILE}'.")
+        else:
+            logger.info(f"✅ Found cache for cv knowledge. Loading data from '{KNOWLEDGE_CACHE_FILE}'...")
+            with open(KNOWLEDGE_CACHE_FILE, 'rb') as f:
+                knowledge = pickle.load(f)
+
+        return knowledge
+
+
+class ChatContext:
+    """
+    Holds the conversational state that uses knowledge base and LLM to answer questions.
+    Context tree:
+    1. Learn about jobs
+        a. position a, company x
+        b. position b, company x
+        ...
+    2. Learn about tech stack and skills
+        a. languages
+        b. domain of expertise
+        c. tech stacks
+    3. Learn about papers
+        a. paper 1
+        b. paper 2
+        ...
+    4. Learn about education
+    """
+    def __init__(self):
+        self.domain_index = -1
+        self.document_index = -1
+
+
+class KnowledgeBase:
+    def __init__(
+            self, 
+            knowledge_config: KnowledgeConfig,
+            generate_cache: bool,
+            download_data: bool,
+        ):
+        self.embedding_model = SentenceTransformer(EMBEDDING_MODEL_ID, device='cuda')
+        self.llm_pipeline = pipeline(task="text-generation", model=LLM_ID, device='cuda')
+        self.knowledge = knowledge_config.initialize(self.embedding_model, generate_cache, download_data)
+
+    def semantic_search_domains(self, query: str) -> int:
+        """symmetric semantic search on domain descriptions"""
+        query_embedding: torch.Tensor = self.embedding_model.encode_query(query, convert_to_tensor=True) # type: ignore
+        similarity_scores = self.embedding_model.similarity(query_embedding, self.knowledge.dk_desc_embeddings)
+        if similarity_scores.numel() == 0:
+            logger.debug("Calculating similarities returned an empty tensor.")
+            return -1
+        
+        best_index = similarity_scores.argmax()
+        best_similarity = similarity_scores[0, best_index]
+        if best_similarity >= DOMAIN_SELECTION_THRESHOLD:
+            logger.debug(f"domain description index {best_index} with similarity {best_similarity} exceeds threshold {DOMAIN_SELECTION_THRESHOLD}.")
+            return int(best_index)
+
+        logger.debug(f"best domain description at {best_index} with similarity {best_similarity} is below threshold {DOMAIN_SELECTION_THRESHOLD}.")
+        return -1
+
+    def semantic_search_documents(self, query: str, chat_context: ChatContext) -> int:
+        """asymmetric semantic search on documents"""
+        query_embedding: torch.Tensor = self.embedding_model.encode_query(query, convert_to_tensor=True) # type: ignore
+        similarity_threshold = DOCUMENT_SELECTION_THRESHOLD if chat_context.document_index == -1 else DOCUMENT_FOLLOWUP_THRESHOLD
+        document_embeddings = self.knowledge.domain_knowledges[chat_context.domain_index].document_embeddings
+
+        similarity_scores = self.embedding_model.similarity(query_embedding, document_embeddings)
+        if similarity_scores.numel() == 0:
+            logger.debug("Calculating similarities returned an empty tensor.")
+            return -1
+
+        best_index = similarity_scores.argmax()
+        best_similarity = similarity_scores[0, best_index]
+        if best_similarity >= similarity_threshold:
+            logger.debug(f"in domain {chat_context.domain_index}, document at {best_index} with similarity {best_similarity} exceeds threshold {similarity_threshold}.")
+            return int(best_index)
+
+        logger.debug(f"in domain {chat_context.domain_index}, document at {best_index} with best similarity {best_similarity} is below threshold {similarity_threshold}.")
+        return -1
+    
+    def get_system_prompt(self, context_str: str) -> str:
+        return f"""
+            \rAnswer the following QUESTION based only on the CONTEXT provided. If the answer cannot be found in the CONTEXT, write \"{self.knowledge.no_match_response}\"
+            \r---
+            \rCONTEXT:
+            \r{context_str}"""
+
+    def get_llm_prompt(self, chat_context: ChatContext, user_input: str, history: list[MessageDict]) -> list[MessageDict]:
+        """Returns the prompt to ask LLM to generate response."""
+        document_content = self.knowledge.domain_knowledges[chat_context.domain_index].documents[chat_context.document_index]
+
+        llm_prompt = list()
+        llm_prompt.append(MessageDict(role="system", content=self.get_system_prompt(document_content)))
+        llm_prompt.extend(history[1:]) # skip welcome message
+        llm_prompt.append(MessageDict(role="user", content=f"QUESTION:\n{user_input}"))
+
+        logger.debug("generated llm prompt:")
+        logger.debug(f"system: {llm_prompt[0]["content"]}")
+        logger.debug(f"user: {llm_prompt[-1]["content"]}")
+        
+        return llm_prompt        
+
+    def intro_response(self, chat_context: ChatContext) -> Generator[Tuple[str, ChatContext]]:
+        yield self.knowledge.welcome_message, chat_context
+
+    def no_match_response(self, chat_context: ChatContext) -> Generator[Tuple[str, ChatContext]]:
+        yield self.knowledge.no_match_response, chat_context
+
+    def answer_query(
+        self, 
+        user_message: str, 
+        history: list[MessageDict], 
+        chat_context: ChatContext,
+    ) -> Generator[Tuple[str, ChatContext]]:
+        """Generates a response based on the user message and chat context."""
+        logger.debug(f"user input: {user_message}")
+        
+        if user_message == self.knowledge.reset_query:
+            # reset chat
+            logger.debug("resetting chat")
+            chat_context.domain_index = -1
+            chat_context.document_index = -1
+            yield from self.intro_response(chat_context)
+            return
+        elif len(history) == 0:
+            # new convo
+            logger.debug("starting a new chat")
+            yield from self.intro_response(chat_context)
+            return
+        elif chat_context.domain_index == -1:
+            # domain selection chat
+            domain_index = self.semantic_search_domains(user_message)
+            if domain_index >= 0:
+                chat_context.domain_index = domain_index
+                yield self.knowledge.domain_knowledges[domain_index].intro_message, chat_context
+                return
+            else:
+                logger.debug("failed to find matching domain")
+                yield from self.no_match_response(chat_context)
+                return
+        else:
+            # domain chat
+            document_index = self.semantic_search_documents(user_message, chat_context)
+            if document_index >= 0:
+                # found content similar to user's input. use that content.
+                chat_context.document_index = document_index    
+                logger.debug(f"using document at index {chat_context.document_index} in domain {chat_context.domain_index} that matches user's input.")
+            elif chat_context.document_index >= 0:
+                # no specific content matching user's input, so use previous document in chat_context.
+                logger.debug(f"no content matches user's input. using previous document at index {chat_context.document_index} in domain {chat_context.domain_index}.")
+            else:
+                # no content found and no previous content. yield no match.
+                logger.debug("no content found and no previous content.")
+                yield from self.no_match_response(chat_context)
+                return
+
+        # Generate response using LLM
+        llm_prompt = self.get_llm_prompt(chat_context, user_message, history)
+
+        llm_response_streamer = TextIteratorStreamer(self.llm_pipeline.tokenizer, skip_prompt=True)  # type: ignore
+        generation_args = dict(
+            text_inputs=llm_prompt,
+            streamer=llm_response_streamer,
+            skip_special_tokens=True,
+            max_new_tokens=RESPONSE_MAX_NEW_TOKENS,
+            do_sample=True,
+            top_p=RESPONSE_SAMPLING_TOP_P,
+            temperature=RESPONSE_SAMPLING_TEMPERATURE,
+        )
+
+        thread = Thread(
+            target=self.llm_pipeline,
+            kwargs=generation_args
+        )
+        thread.start()
+
+        response = ""
+        for new_text in llm_response_streamer:
+            response += new_text
+            yield response, chat_context
+
+        logger.debug(f"[assistant] {response}")

pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "cvchatbot_wjlee"
+version = "0.1.0"
+description = "Chatbot app that answers questions about CV"
+readme = "README.md"
+requires-python = "==3.12.*"
+dependencies = [
+    "accelerate>=1.12.0",
+    "google-cloud-storage>=3.8.0",
+    "gradio[oauth]>=6.3.0",
+    "hf-xet>=1.2.0",
+    "sentence-transformers>=5.2.0",
+    "spaces>=0.46.0",
+    "transformers>=4.57.5",
+]

requirements.txt

@@ -0,0 +1,16 @@
+# This file was autogenerated by uv via the following command:
+#    uv pip compile pyproject.toml -o requirements.txt --no-deps --no-strip-extras
+accelerate==1.12.0
+    # via cvchatbot-wjlee (pyproject.toml)
+google-cloud-storage==3.8.0
+    # via cvchatbot-wjlee (pyproject.toml)
+gradio[oauth]==6.3.0
+    # via cvchatbot-wjlee (pyproject.toml)
+hf-xet==1.2.0
+    # via cvchatbot-wjlee (pyproject.toml)
+sentence-transformers==5.2.0
+    # via cvchatbot-wjlee (pyproject.toml)
+spaces==0.46.0
+    # via cvchatbot-wjlee (pyproject.toml)
+transformers==4.57.5
+    # via cvchatbot-wjlee (pyproject.toml)