Skip to content

Backend

__main__.py

Entry point that launches a uvicorn server and connects the backend reading, generation, and writing functionalities

main()

Start uvicorn server

Source code in src/chatbot_util/__main__.py
15
16
17
18
19
20
21
22
23
24
def main() -> None:
    """Start uvicorn server"""
    host = "127.0.0.1"
    port = 8080

    uvicorn.run(
        "chatbot_util.api:app",
        host=host,
        port=port,
    )

start()

Read info from files, append generated questions, then write to new file, indicating whether the run was interrupted

Source code in src/chatbot_util/__main__.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def start() -> bool:
    """Read info from files, append generated questions, then write to new file, indicating whether the run was interrupted"""

    # Check interrupt status between operations to avoid undesired behavior
    if chain.handle_interrupt():
        return True

    # Read topics and organic questions
    utils.logger.info("Reading topics, questions, employees, and answers.")
    store, teams, employees, phrases, answers, nums = file_io.read()

    if chain.handle_interrupt():
        return True

    # Generate synthetic questions
    permutated_store = chain.generate(store, phrases)

    if chain.handle_interrupt():
        return True

    # Recreate Permutated.csv w/ synthetic questions appended
    utils.logger.info(f'Writing to "{file_io.FILENAMES["permutated"]}."')
    file_io.write(permutated_store, teams, employees, answers, nums)
    utils.logger.info("Done")

    return False

verify()

Determine whether the updated Permutated.csv has any modified or missing entries (and not just new ones)

Source code in src/chatbot_util/__main__.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def verify() -> bool:
    """Determine whether the updated `Permutated.csv` has any modified or missing entries (and not just new ones)"""
    verified = True
    writefile = file_io.FILENAMES["permutated"]
    backupfile = writefile + ".backup"

    # return early when there's nothing to verify against
    if not os.path.exists(backupfile):
        return verified

    # pipe the diff into grep to ignore added entries
    diff = subprocess.run(
        [
            "diff",
            "--strip-trailing-cr",
            "-y",
            "--suppress-common-lines",
            writefile,
            backupfile,
        ],
        stdout=subprocess.PIPE,
    )
    return_code = subprocess.run(
        ["grep", ">\\||"],
        input=diff.stdout,
        stdout=subprocess.PIPE,
    ).returncode

    if return_code != 1:
        verified = False

    return verified

api.py

API routes and implementations for the uvicorn server launched in __main__.py

config()

Return CDN links to FAQ - Enter Here.csv and Other.txt

Status codes

  • 200 = no errors encountered while reading links from config

Config keys

  • faq = CDN link to FAQ - Enter Here.csv
  • other = CDN link to Other.txt
Source code in src/chatbot_util/api.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@app.get("/api/config")
def config() -> dict[str, str]:
    """Return CDN links to `FAQ - Enter Here.csv` and `Other.txt`

    Status codes

    - `200` = no errors encountered while reading links from config

    Config keys

    - `faq` = CDN link to `FAQ - Enter Here.csv`
    - `other` = CDN link to `Other.txt`
    """
    return file_io.read_config()

files()

Report on data file status

Status codes

  • 200 = no errors encountered while scanning for files

present

  • True = all files are present
  • False = some files are missing
Source code in src/chatbot_util/api.py
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@app.get("/api/files")
def files() -> dict[str, bool]:
    """Report on data file status

    Status codes

    - `200` = no errors encountered while scanning for files

    `present`

    - `True` = all files are present
    - `False` = some files are missing
    """
    present = True
    filenames = [file_io.FILENAMES["faq"], file_io.FILENAMES["other"]]

    for f in filenames:
        if not os.path.exists(f):
            present = False
            break

    return {"present": present}

generate(response)

Create chain, read info from files, append generated questions, then write to new file

Status codes

  • 201 = successfully generated Permutated.csv
  • 429 = request denied because generation is in progress

verified

  • True = verified
  • False = unverified, check diff
  • None = generation interrupted
Source code in src/chatbot_util/api.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@app.post("/api/generate", status_code=fastapi.status.HTTP_201_CREATED)
def generate(response: fastapi.Response) -> dict[str, bool | None]:
    """Create chain, read info from files, append generated questions, then write to new file

    Status codes

    - `201` = successfully generated `Permutated.csv`
    - `429` = request denied because generation is in progress

    `verified`

    - `True` = verified
    - `False` = unverified, check diff
    - `None` = generation interrupted
    """

    # generate new Permutated.csv
    verified = None
    interrupted = False
    global allow_generate
    if allow_generate:
        allow_generate = False
        interrupted = __main__.start()
        allow_generate = True
    else:
        response.status_code = fastapi.status.HTTP_429_TOO_MANY_REQUESTS

        # involuntary interruption prevents setting verStatus when rate-limited
        interrupted = True

    # identify whether new Permutated.csv is verified, ignoring interruption case
    if not interrupted:
        verified = __main__.verify()

    return {"verified": verified}

health(response)

Health check for both uvicorn and ollama servers

Status codes

  • 200 = uvicorn and ollama are ready
  • 500 = ollama is not available
Source code in src/chatbot_util/api.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@app.get("/api/health")
def health(response: fastapi.Response) -> None:
    """Health check for both uvicorn and ollama servers

    Status codes

    - `200` = uvicorn and ollama are ready
    - `500` = ollama is not available
    """
    try:
        ollama.show("mistral")
    except Exception:
        response.status_code = fastapi.status.HTTP_500_INTERNAL_SERVER_ERROR
        utils.logger.error("You must install Ollama before using this utility.")

interrupt()

Interrupt the current generation task

Status codes

  • 200 = successfully interrupted generation of Permutated.csv
Source code in src/chatbot_util/api.py
101
102
103
104
105
106
107
108
109
110
111
@app.get("/api/interrupt")
def interrupt() -> None:
    """Interrupt the current generation task

    Status codes

    - `200` = successfully interrupted generation of `Permutated.csv`
    """
    chain.interrupt = True
    while chain.progress.index != 0:
        time.sleep(0.1)

progress()

Report on generation progress

Status codes

  • 200 = successfully retrieved progress status

Response keys

  • index = index of topic currently generating queries for, [1-total]
  • total = total number of topics to generate queries for
Source code in src/chatbot_util/api.py
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@app.get("/api/progress")
def progress() -> dict[str, int]:
    """Report on generation progress

    Status codes

    - `200` = successfully retrieved progress status

    Response keys

    - `index` = index of topic currently generating queries for, [1-total]
    - `total` = total number of topics to generate queries for
    """
    return {"index": chain.progress.index, "total": chain.progress.total}

upload(files)

Replace data files via upload and return list of status codes

Status codes

  • 201 = no errors encountered when replacing file(s)
  • 422 = validation error while processing file(s)

uploaded

  • True = succssfully replaced all files
  • False = some files failed to be replaced
Source code in src/chatbot_util/api.py
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@app.post("/api/upload", status_code=fastapi.status.HTTP_201_CREATED)
def upload(files: list[fastapi.UploadFile]) -> dict[str, bool]:
    """Replace data files via upload and return list of status codes

    Status codes

    - `201` = no errors encountered when replacing file(s)
    - `422` = validation error while processing file(s)

    `uploaded`

    - `True` = succssfully replaced all files
    - `False` = some files failed to be replaced
    """
    uploaded = True
    for f in files:
        if file_io.create_file(f) is not True:
            uploaded = False
            break

    return {"uploaded": uploaded}

chain.py

Setup language model and output parser, then generate and append new questions

Progress

Representation of generation progress with utility method for updating and logging

Source code in src/chatbot_util/chain.py
11
12
13
14
15
16
17
18
19
20
21
22
23
class Progress:
    """Representation of generation progress with utility method for updating and logging"""

    index: int = 0
    total: int

    def __init__(self, total: int) -> None:
        self.total = total

    def update(self, index: int) -> None:
        """Update and log generation progress"""
        self.index = index
        utils.logger.info(f"Generating similar queries for: {self.index}/{self.total}.")

update(index)

Update and log generation progress

Source code in src/chatbot_util/chain.py
20
21
22
23
def update(self, index: int) -> None:
    """Update and log generation progress"""
    self.index = index
    utils.logger.info(f"Generating similar queries for: {self.index}/{self.total}.")

generate(store, phrases)

Generate and append new questions to store

Source code in src/chatbot_util/chain.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def generate(
    store: dict[str, list[str]], phrases: list[list[str]]
) -> dict[str, list[str]]:
    """Generate and append new questions to store"""

    # Calculate total number of questions to generate
    index, total = 1, 0
    for topic in store:
        for question in store[topic]:
            total += 1

    # Init progress
    global progress
    progress = Progress(total)

    # Iterate over each topic (e.g., 'Reach', 'CEN', etc.)
    for topic in store:
        new_questions: list[list[str]] = []

        # Iterate over each question in the topic, generating 5 new ones / question
        for question in store[topic]:
            # return early if interrupted
            if interrupt:
                return store

            progress.update(index)
            prompt = INSTRUCTION + question
            new_questions.append(invoke(prompt, phrases))
            index += 1

        # Append synthetic questions for each organic question in this topic
        for new_sub_question in new_questions:
            for new_question in new_sub_question:
                store[topic].append(new_question)

    # Reset progress
    progress = Progress(0)
    return store

handle_interrupt()

Helper to reset chain state when interrupted

Source code in src/chatbot_util/chain.py
33
34
35
36
37
38
39
40
41
42
43
44
def handle_interrupt() -> bool:
    """Helper to reset chain state when interrupted"""
    interrupted = False
    global interrupt
    global progress
    if interrupt:
        interrupt = False
        progress = Progress(0)
        utils.logger.info("Interrupted")
        interrupted = True

    return interrupted

invoke(prompt, phrases)

Define chat model, then create the chain

Source code in src/chatbot_util/chain.py
79
80
81
82
83
84
85
86
87
88
89
def invoke(prompt: str, phrases: list[list[str]]) -> list[str]:
    """Define chat model, then create the chain"""
    options = {"seed": 39}
    response = ollama.generate(
        model="mistral",
        prompt=prompt,
        options=options,
    )

    cleaned_response = parse(response["response"], phrases)
    return cleaned_response

parse(response, phrases)

Parse lines from LLM and clean them before returning

Source code in src/chatbot_util/chain.py
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def parse(response: str, phrases: list[list[str]]) -> list[str]:
    """Parse lines from LLM and clean them before returning"""

    def remove_empty(lines: list[str]) -> list[str]:
        """Remove empty lines"""
        for line in lines:
            if line == "":
                lines.remove(line)
        return lines

    def remove_numbers(lines: list[str]) -> list[str]:
        """Remove numbers"""
        for i, line in enumerate(lines):
            if line[1] == ".":
                lines[i] = line[3:]
        return lines

    def remove_phrases(lines: list[str], phrases: list[list[str]]) -> list[str]:
        """Find and replace phrases"""
        for i, line in enumerate(lines):
            for phrase in phrases:
                line = line.replace(phrase[0], phrase[1])
            lines[i] = line
        return lines

    lines = response.strip().split("\n")
    nonempty_lines = remove_empty(lines)
    no_num_lines = remove_numbers(nonempty_lines)
    cleaned_lines = remove_phrases(no_num_lines, phrases)
    return cleaned_lines

file_io.py

Reads toml config, csv, employees, and answers, and writes generated result to csv

create_file(f)

Create or replace a data file

Source code in src/chatbot_util/file_io.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def create_file(f: fastapi.UploadFile) -> bool | None:
    """Create or replace a data file"""
    # determine whether anything should be done with the uploaded file
    created: bool | None = True
    filename: str | None = None
    if str(f.filename) == FAQ:
        filename = FILENAMES["faq"]
    elif str(f.filename) == OTHER:
        filename = FILENAMES["other"]
    elif str(f.filename) == CONFIG:
        filename = FILENAMES["config"]

    # try to replace the relevant file bytewise
    if filename is not None:
        try:
            contents = f.file.read()

            # create the data dir if it doesn't exist
            if not os.path.exists(DIR):
                os.makedirs(DIR)

            # move the old file if it exists
            if os.path.exists(filename):
                os.rename(
                    filename,
                    filename + ".backup",
                )

            with open(filename, "wb") as openfile:
                openfile.write(contents)

        except Exception:
            created = False
    else:
        created = None

    return created

read()

Read questions from csv file, read teams, employees, phrases and answers from text files

Source code in src/chatbot_util/file_io.py
204
205
206
207
208
209
210
211
212
213
214
215
216
def read() -> tuple[
    dict[str, list[str]],
    list[str],
    dict[str, list[str]],
    list[list[str]],
    utils.Answers,
    utils.Nums,
]:
    """Read questions from csv file, read teams, employees, phrases and answers from text files"""
    teams, employees, phrases, answers = read_other(FILENAMES["other"])
    store, nums = read_entries(FILENAMES["faq"], teams)

    return store, teams, employees, phrases, answers, nums

read_answers(lines)

Read and return answers for cen and teams

Source code in src/chatbot_util/file_io.py
170
171
172
173
174
175
176
177
178
179
def read_answers(lines: list[list[str]]) -> utils.Answers:
    """Read and return answers for cen and teams"""
    cen_answers = read_cen(lines[0])
    other_answers = [read_basic(lines[1]), read_basic(lines[2])]

    answers: utils.Answers = {
        "cen_answers": cen_answers,
        "other_answers": other_answers,
    }
    return answers

read_basic(lines)

Read and return basic answers for topics other than CEN

Source code in src/chatbot_util/file_io.py
160
161
162
163
164
165
166
167
def read_basic(lines: list[str]) -> list[str]:
    """Read and return basic answers for topics other than CEN"""
    basic_answers: list[str] = []
    for line in lines:
        clean_line = line.strip("\n")
        basic_answers.append(clean_line)

    return basic_answers

read_cen(lines)

Read and return cen_answers

Source code in src/chatbot_util/file_io.py
147
148
149
150
151
152
153
154
155
156
157
def read_cen(lines: list[str]) -> dict[str, list[str]]:
    """Read and return `cen_answers`"""
    cen_answers: dict[str, list[str]] = {}
    for i, line in enumerate(lines):
        if len(line) >= 3:
            part1, part2 = line[:-1].split(sep=":")
        else:
            part1, part2 = line.split(sep=":")
        cen_answers[f"cen_{i}"] = [part1, part2]

    return cen_answers

read_config()

Read links from config file

Source code in src/chatbot_util/file_io.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def read_config() -> dict[str, str]:
    """Read links from config file"""
    links: dict[str, str] = {"faq": "", "other": ""}

    try:
        with open(FILENAMES["config"], "rb") as f:
            config = tomllib.load(f)

        links: dict[str, str] = config["links"]

    except Exception:
        utils.logger.warning("Failed to load config.toml. Defaulting to empty links")

    return links

read_employees(lines)

Read and return employee list

Source code in src/chatbot_util/file_io.py
123
124
125
126
127
128
129
130
131
132
133
def read_employees(lines: list[str]) -> dict[str, list[str]]:
    """Read and return employee list"""
    employees: dict[str, list[str]] = {}
    for line in lines:
        if len(line) >= 3:
            employee, role, pronoun = line[:-1].split(sep=":")
        else:
            employee, role, pronoun = line.split(sep=":")
        employees[employee] = [role, pronoun]

    return employees

read_entries(filename, teams)

Read and return topics and basic answers

Source code in src/chatbot_util/file_io.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def read_entries(
    filename: str, teams: list[str]
) -> tuple[dict[str, list[str]], utils.Nums]:
    """Read and return topics and basic answers"""
    with open(filename, "r", encoding="utf-8") as f:
        # it seems as though the delimiter doesn't actually
        # matter as long as it's not the default: ",". updating
        # to 3.13 broke "\n", and alternative approaches (e.g.,
        # topic = line[0] w/ delimiter = ",") broke Ollama's
        # determinism
        reader = csv.reader(f, delimiter="\t")
        store: dict[str, list[str]] = {}
        cur_topic = ""
        nums: utils.Nums = {
            "num_cen": 0,
            "num_other": [0] * len(teams),
        }

        for i, line in enumerate(reader):
            if i != 0:
                if len(line) > 0:
                    topic = line[0].split(",")[0]
                    question = line[0].removeprefix(topic + ",")
                    if topic != "" and question != "":
                        cur_topic = topic
                        store[cur_topic] = []
                    if question != "":
                        store[cur_topic].append(question)
                        if cur_topic == "CEN":
                            nums["num_cen"] += 1
                        elif cur_topic in teams:
                            nums["num_other"][teams.index(cur_topic)] += 1

    return store, nums

read_other(filename)

Read and return employees, phrases, and answers

Source code in src/chatbot_util/file_io.py
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def read_other(
    filename: str,
) -> tuple[list[str], dict[str, list[str]], list[list[str]], utils.Answers]:
    """Read and return employees, phrases, and answers"""
    with open(filename, "r", encoding="utf-8") as f:
        raw_lines = f.readlines()
        lines: list[list[str]] = [[], [], [], [], [], [], []]

        cur = 0
        for raw_line in raw_lines:
            if raw_line == "\n":
                cur += 1
                continue
            lines[cur].append(raw_line)

    teams = read_teams(lines[0])
    employees = read_employees(lines[1])
    phrases = read_phrases(lines[2])
    answers = read_answers(lines[3:])
    return teams, employees, phrases, answers

read_phrases(lines)

Read and return phrases to find and replace

Source code in src/chatbot_util/file_io.py
136
137
138
139
140
141
142
143
144
def read_phrases(lines: list[str]) -> list[list[str]]:
    """Read and return phrases to find and replace"""
    phrases: list[list[str]] = []
    for line in lines:
        find, replace = line.split(sep=":")
        replace = replace.strip("\n")
        phrases.append([find, replace])

    return phrases

read_teams(lines)

Read and return teams

Source code in src/chatbot_util/file_io.py
82
83
84
def read_teams(lines: list[str]) -> list[str]:
    """Read and return teams"""
    return [line.strip() for line in lines]

write(store, teams, employees, answers, nums)

Format questions and topics, write to csv file

Source code in src/chatbot_util/file_io.py
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def write(
    store: dict[str, list[str]],
    teams: list[str],
    employees: dict[str, list[str]],
    answers: utils.Answers,
    nums: utils.Nums,
) -> None:
    """Format questions and topics, write to csv file"""
    if os.path.exists(FILENAMES["permutated"]):
        os.rename(FILENAMES["permutated"], FILENAMES["permutated"] + ".backup")

    with open(FILENAMES["permutated"], "w", encoding="utf-8") as csvfile:
        csvfile.write('"question","answer"\n')
        indices: utils.Indices = {
            "cen_index": 0,
            "other_index": [0] * len(teams),
        }
        for topic in store:
            for question in store[topic]:
                # Write cleaned entry to csv
                answer, indices = utils.create_answer(
                    topic,
                    question,
                    teams,
                    employees,
                    answers,
                    nums,
                    indices,
                )
                entry = utils.clean_entry(question, answer)
                csvfile.write(entry)

utils.py

Utilities for logging and creating + cleaning answers based on file content

Answers

Bases: TypedDict

Representation of the answer dict variants expected for each topic

Source code in src/chatbot_util/utils.py
25
26
27
28
29
class Answers(TypedDict):
    """Representation of the answer dict variants expected for each topic"""

    cen_answers: dict[str, list[str]]
    other_answers: list[list[str]]

Indices

Bases: TypedDict

Representation of the indices dict variants expected for each topic

Source code in src/chatbot_util/utils.py
39
40
41
42
43
class Indices(TypedDict):
    """Representation of the indices dict variants expected for each topic"""

    cen_index: int
    other_index: list[int]

Nums

Bases: TypedDict

Representation of the nums dict variants expected for each topic

Source code in src/chatbot_util/utils.py
32
33
34
35
36
class Nums(TypedDict):
    """Representation of the nums dict variants expected for each topic"""

    num_cen: int
    num_other: list[int]

clean_entry(question, answer)

Clean up unnecessary quotes

Source code in src/chatbot_util/utils.py
128
129
130
131
132
133
134
135
136
137
138
def clean_entry(question: str, answer: str) -> str:
    """Clean up unnecessary quotes"""
    if question[0] == '"' and question[-1] == '"':
        entry = f'{question},"{answer}"\n'
    elif question[0] == '"':
        entry = f'{question}","{answer}"\n'
    elif question[-1] == '"':
        entry = f'"{question},"{answer}"\n'
    else:
        entry = f'"{question}","{answer}"\n'
    return entry

create_answer(topic, question, teams, employees, answers, nums, indices)

Convert topics to answers depending on whether the topic is a person, CEN, or other

Source code in src/chatbot_util/utils.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def create_answer(
    topic: str,
    question: str,
    teams: list[str],
    employees: dict[str, list[str]],
    answers: Answers,
    nums: Nums,
    indices: Indices,
) -> tuple[str, Indices]:
    """Convert topics to answers depending on whether the topic is a person, CEN, or other"""
    if topic == "CEN":
        answer, indices["cen_index"] = create_cen_answer(
            question,
            answers["cen_answers"],
            nums["num_cen"],
            indices["cen_index"],
        )
    elif topic in teams:
        i = teams.index(topic)
        answer, indices["other_index"][i] = create_other_answer(
            answers["other_answers"][i],
            nums["num_other"][i],
            indices["other_index"][i],
        )
    else:
        answer = create_person_answer(topic, employees)
    return answer, indices

create_cen_answer(question, cen_answers, num_cen, cen_index)

Update CEN topics to be the relevant answer

Source code in src/chatbot_util/utils.py
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def create_cen_answer(
    question: str, cen_answers: dict[str, list[str]], num_cen: int, cen_index: int
) -> tuple[str, int]:
    """Update CEN topics to be the relevant answer"""
    answer = None
    i = 0
    while not answer:
        if (
            ((cen_index in set(range(i * 2, (i + 1) * 2))) and (num_cen >= (i + 1) * 2))
            or (cen_index == 0 and num_cen == 1)
            or (cen_index in set(range(num_cen + (i * 10), num_cen + ((i + 1) * 10))))
        ):
            answer = create_cen_answer_helper(question, cen_answers[f"cen_{i}"])
        i += 1
    cen_index += 1
    return answer, cen_index

create_cen_answer_helper(question, cen_answer)

Format answer based on question content

Source code in src/chatbot_util/utils.py
56
57
58
59
60
61
62
63
64
65
66
def create_cen_answer_helper(question: str, cen_answer: list[str]) -> str:
    """Format answer based on question content"""
    if (
        ("CEN" in question)
        and ("acronym" not in question)
        and ("abbreviation" not in question)
    ):
        answer = f"{cen_answer[0]}CEN{cen_answer[1]}"
    else:
        answer = f"{cen_answer[0]}Collegiate Edu-Nation{cen_answer[1]}"
    return answer

create_other_answer(answers, num, index)

Update other topics to be the relevant answer

Source code in src/chatbot_util/utils.py
87
88
89
90
91
92
93
94
95
96
def create_other_answer(answers: list[str], num: int, index: int) -> tuple[str, int]:
    """Update other topics to be the relevant answer"""
    answer = None
    i = 0
    while not answer:
        if (index == i) or (index in set(range(num + (i * 5), num + ((i + 1) * 5)))):
            answer = answers[i]
        i += 1
    index += 1
    return answer, index

create_person_answer(topic, employees)

Update person topics to be their contact info

Source code in src/chatbot_util/utils.py
46
47
48
49
50
51
52
53
def create_person_answer(topic: str, employees: dict[str, list[str]]) -> str:
    """Update person topics to be their contact info"""
    temp = topic.split(" ")
    answer = (
        f"{topic}, CEN's {employees[topic][0]}, can help with that. "
        f"{employees[topic][1]} contact is {temp[0][0].lower()}{temp[1].lower()}@edu-nation.org"
    )
    return answer