import itertools import json import re from functools import partial from pathlib import Path import pandas as pd import requests import streamlit as st import webvtt from transformers import AutoTokenizer from generate_text_api import TextGenerator from model_inferences.utils.chunking import Truncater from model_inferences.utils.files import get_captions_from_vtt, get_transcript USE_PARAGRAPHING_MODEL = True def get_sublist_by_flattened_index(A, i): current_index = 0 for sublist in A: sublist_length = len(sublist) if current_index <= i < current_index + sublist_length: return sublist, A.index(sublist) current_index += sublist_length return None, None import requests def get_talk_metadata(video_id): url = "https://www.ted.com/graphql" headers = { "Content-Type": "application/json", "Accept": "application/json", "x-operation-name": "Transcript", # Replace with the actual operation name } data = { "query": """ query GetTalk($videoId: ID!) { video(id: $videoId) { title, presenterDisplayName, nativeDownloads {medium} } } """, "variables": { "videoId": video_id, # Corrected key to "videoId" }, } response = requests.post(url, json=data, headers=headers) if response.status_code == 200: result = response.json() return result else: print(f"Error: {response.status_code}, {response.text}") class OfflineTextSegmenterClient: def __init__(self, host_url): self.host_url = host_url.rstrip("/") + "/segment" def segment(self, text, captions=None, generate_titles=False, threshold=0.4): payload = { 'text': text, 'captions': captions, 'generate_titles': generate_titles, "prefix_titles": True, "threshold": threshold, } headers = { 'Content-Type': 'application/json' } response = requests.post(self.host_url, data=json.dumps(payload), headers=headers).json() #segments = response["annotated_segments"] if "annotated_segments" in response else response["segments"] return {'segments':response["segments"], 'titles': response["titles"], 'sentences': response["sentences"]} class Toc: def __init__(self): self._items = [] self._placeholder = None def title(self, text): self._markdown(text, "h1") def header(self, text): self._markdown(text, "h2", " " * 2) def subheader(self, text): self._markdown(text, "h3", " " * 4) def placeholder(self, sidebar=False): self._placeholder = st.sidebar.empty() if sidebar else st.empty() def generate(self): if self._placeholder: self._placeholder.markdown("\n".join(self._items), unsafe_allow_html=True) def _markdown(self, text, level, space=""): key = re.sub(r'[^\w-]', '', text.replace(" ", "-").replace("'", "-").lower()) st.markdown(f"<{level} id='{key}'>{text}", unsafe_allow_html=True) self._items.append(f"{space}* {text}") # custom_css = "" # st.write(custom_css, unsafe_allow_html=True) def concat_prompt(prompt_text, text, model_name): if 'flan' in model_name: input_ = prompt_text + "\n\n" + text elif 'galactica' in model_name: input_ = text + "\n\n" + prompt_text return input_ endpoint = "http://hiaisc.isl.iar.kit.edu/summarize" ENDPOINTS = {"http://hiaisc.isl.iar.kit.edu/summarize": "meta-llama/Llama-2-13b-chat-hf",} client = OfflineTextSegmenterClient("http://hiaisc.isl.iar.kit.edu/chapterize") if USE_PARAGRAPHING_MODEL: paragrapher = OfflineTextSegmenterClient("http://hiaisc.isl.iar.kit.edu/paragraph") summarizer = TextGenerator(endpoint) tokenizer = AutoTokenizer.from_pretrained(ENDPOINTS[endpoint], use_fast=False) # TLDR PROMPT SYSTEM_PROMPT = "You are an assistant who replies with a summary to every message." TLDR_PROMPT_TEMPLATE = """[INST] <> {system_prompt} <> {user_message} [/INST] Sure! Here is a summary of the research presentation in a single, short sentence:""" TLDR_USER_PROMPT = "Summarize the following research presentation in a single, short sentence:\n\n{input}" TLDR_PROMPT = TLDR_PROMPT_TEMPLATE.format(system_prompt=SYSTEM_PROMPT, user_message=TLDR_USER_PROMPT) TLDR_PROMPT_LENGTH = tokenizer(TLDR_PROMPT, return_tensors="pt")["input_ids"].size(1) BP_PROMPT_TEMPLATE = """[INST] <> {system_prompt} <> {user_message} [/INST] Sure! Here is a summary of the research presentation using three bullet points:\n\n\u2022""" BP_USER_PROMPT = "Summarize the following research presentation using three bullet points:\n\n{input}" BP_PROMPT = BP_PROMPT_TEMPLATE.format(system_prompt=SYSTEM_PROMPT, user_message=TLDR_USER_PROMPT) BP_PROMPT_LENGTH = tokenizer(BP_PROMPT, return_tensors="pt")["input_ids"].size(1) CONTEXT_LENGTH = 3072 MAX_SUMMARY_LENGTH = 1024 TLDR_MAX_INPUT_LENGTH = CONTEXT_LENGTH - MAX_SUMMARY_LENGTH - TLDR_PROMPT_LENGTH - 1 BP_MAX_INPUT_LENGTH = CONTEXT_LENGTH - MAX_SUMMARY_LENGTH - BP_PROMPT_LENGTH - 1 text_generator = TextGenerator(endpoint) temperature = 0.7 import re def replace_newlines(text): updated_text = re.sub(r'\n+', r'\n\n', text) return updated_text def generate_summary(summarizer, generated_text_box, input_, prompt, max_input_length, prefix=""): all_generated_text = prefix truncater = Truncater(tokenizer, max_length=max_input_length) input_ = truncater(input_) input_ = prompt.format(input=input_) for generated_text in summarizer.generate_text_stream(input_, max_new_tokens=MAX_SUMMARY_LENGTH, do_sample=True, temperature=temperature): all_generated_text += replace_newlines(generated_text) generated_text_box.info(all_generated_text) print(all_generated_text) return all_generated_text.strip() st.header("Demo: Intelligent Recap") if not hasattr(st, 'global_state'): st.global_state = {'NIPS 2021 Talks': None, 'TED Talks': None} # NIPS 2021 Talks transcript_files = itertools.islice(Path("demo_data/nips-2021/").rglob("transcript_whisper_large-v2.vtt"), 15) # get titles from metadata.json transcripts_map = {} for transcript_file in transcript_files: base_path = transcript_file.parent metadata = base_path / "metadata.json" txt_file = base_path / "transcript_whisper_large-v2.txt" with open(metadata) as f: metadata = json.load(f) title = metadata["title"] transcript = get_transcript(txt_file) captions = get_captions_from_vtt(transcript_file) transcripts_map[title] = {"transcript": transcript, "captions": captions, "video": base_path / "video.mp4"} st.global_state['NIPS 2021 Talks'] = transcripts_map data = pd.read_json("demo_data/ted_talks.json") video_ids = data.talk_id.tolist() transcripts = data.text.apply(lambda x: " ".join(x)).tolist() transcripts_map = {} for video_id, transcript in zip(video_ids, transcripts): metadata = get_talk_metadata(video_id) title = metadata["data"]["video"]["title"] presenter = metadata["data"]["video"]["presenterDisplayName"] print(metadata["data"]) if metadata["data"]["video"]["nativeDownloads"] is None: continue video_url = metadata["data"]["video"]["nativeDownloads"]["medium"] transcripts_map[title] = {"transcript": transcript, "video": video_url, "presenter": presenter} st.global_state['TED Talks'] = transcripts_map def get_lecture_id(path): return int(path.parts[-2].split('-')[1]) transcript_files = Path("demo_data/lectures/").rglob("English.vtt") sorted_path_list = sorted(transcript_files, key=get_lecture_id) transcripts_map = {} for transcript_file in sorted_path_list: base_path = transcript_file.parent lecture_id = base_path.parts[-1] transcript = " ".join([c["text"].strip() for c in get_captions_from_vtt(transcript_file)]).replace("\n", " ") video_path = Path(base_path, "video.mp4") transcripts_map["Machine Translation: " + lecture_id] = {"transcript": transcript, "video": video_path} st.global_state['KIT Lectures'] = transcripts_map type_of_document = st.selectbox('What kind of document do you want to test it on?', list(st.global_state.keys())) transcripts_map = st.global_state[type_of_document] selected_talk = st.selectbox("Choose a document...", list(transcripts_map.keys())) st.video(str(transcripts_map[selected_talk]['video']), format="video/mp4", start_time=0) input_text = st.text_area("Transcript", value=transcripts_map[selected_talk]['transcript'], height=300) toc = Toc() summarization_todos = [] with st.expander("Adjust Thresholds"): threshold = st.slider('Chapter Segmentation Threshold', 0.00, 1.00, value=0.4, step=0.05) paragraphing_threshold = st.slider('Paragraphing Threshold', 0.00, 1.00, value=0.5, step=0.05) if st.button("Process Transcript"): with st.sidebar: st.header("Table of Contents") toc.placeholder() st.header(selected_talk, divider='rainbow') # if 'presenter' in transcripts_map[selected_talk]: # st.markdown(f"### *by **{transcripts_map[selected_talk]['presenter']}***") captions = transcripts_map[selected_talk]['captions'] if 'captions' in transcripts_map[selected_talk] else None result = client.segment(input_text, captions, generate_titles=True, threshold=threshold) if USE_PARAGRAPHING_MODEL: presult = paragrapher.segment(input_text, captions, generate_titles=False, threshold=paragraphing_threshold) paragraphs = presult['segments'] segments, titles, sentences = result['segments'], result['titles'], result['sentences'] if USE_PARAGRAPHING_MODEL: prev_chapter_idx = 0 prev_paragraph_idx = 0 segment = [] for i, sentence in enumerate(sentences): chapter, chapter_idx = get_sublist_by_flattened_index(segments, i) paragraph, paragraph_idx = get_sublist_by_flattened_index(paragraphs, i) if (chapter_idx != prev_chapter_idx and paragraph_idx == prev_paragraph_idx) or (paragraph_idx != prev_paragraph_idx and chapter_idx != prev_chapter_idx): print("Chapter / Chapter & Paragraph") segment_text = " ".join(segment) toc.subheader(titles[prev_chapter_idx]) if len(segment_text) > 1200: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment_text, BP_PROMPT, BP_MAX_INPUT_LENGTH, prefix="\u2022")) elif len(segment_text) > 450: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment_text, TLDR_PROMPT, TLDR_MAX_INPUT_LENGTH)) st.write(segment_text) segment = [] elif paragraph_idx != prev_paragraph_idx and chapter_idx == prev_chapter_idx: print("Paragraph") segment.append("\n\n") segment.append(sentence) prev_chapter_idx = chapter_idx prev_paragraph_idx = paragraph_idx segment_text = " ".join(segment) toc.subheader(titles[prev_chapter_idx]) if len(segment_text) > 1200: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment_text, BP_PROMPT, BP_MAX_INPUT_LENGTH, prefix="\u2022")) elif len(segment_text) > 450: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment_text, TLDR_PROMPT, TLDR_MAX_INPUT_LENGTH)) st.write(segment_text) else: segments = [" ".join([sentence for sentence in segment]) for segment in segments] for title, segment in zip(titles, segments): toc.subheader(title) if len(segment) > 1200: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment, BP_PROMPT, BP_MAX_INPUT_LENGTH, prefix="\u2022")) elif len(segment) > 450: generated_text_box = st.info("") summarization_todos.append(partial(generate_summary, summarizer, generated_text_box, segment, TLDR_PROMPT, TLDR_MAX_INPUT_LENGTH)) st.write(segment) toc.generate() for summarization_todo in summarization_todos: summarization_todo()