from langchain import PromptTemplate from langchain.chat_models import ChatOpenAI from langchain.chains.summarize import load_summarize_chain from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import DirectoryLoader from wordcloud import WordCloud, STOPWORDS import numpy as np from langchain.embeddings import OpenAIEmbeddings from sklearn.cluster import KMeans from sklearn.metrics import silhouette_score import os from langchain.docstore.document import Document import re from collections import Counter # import nltk from nltk.corpus import stopwords os.environ["OPENAI_API_KEY"] = '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' class Extract_Summary: def __init__(self,text_input, file_path=None, chunks=2000, chunking_strategy=None, LLM_Model="gpt-3.5-turbo", temperature=1, top_p=None, top_k=None): self.chunks = chunks self.file_path = file_path self.text_input = text_input self.chuking_strategy = chunking_strategy self.LLM_Model = LLM_Model self.temperature = temperature self.top_p = top_p self.top_k = top_k def doc_summary(self, docs): # print(f'You have {len(docs)} documents') num_words = sum([len(doc.page_content.split(" ")) for doc in docs]) # print(f"You have {num_words} words in documents") return num_words, len(docs) def load_docs(self): if self.file_path is not None: docs = DirectoryLoader(self.file_path, glob="**/*.txt").load() else: docs = Document(page_content=f"{self.text_input}", metadata={"source": "local"}) docs = [docs] # docs = self.text_input tokens, documents_count = self.doc_summary(docs) if documents_count > 8 or tokens > 6000: ## Add token checks as well. Add Model availabilty checks docs = self.chunk_docs(docs) ## Handling Large Document with token more than 6000 docs = self.summarise_large_documents(docs) tokens, documents_count = self.doc_summary(docs) if tokens > 2000: docs = self.chunk_docs(docs) chain_type = 'map_reduce' else: chain_type = 'stuff' print("=="*20) print(tokens) print(chain_type) return docs, chain_type ## Add ensemble retriver for this as well. def summarise_large_documents(self, docs): print("=="*20) print('Orignial Docs size : ' ,len(docs)) embeddings = OpenAIEmbeddings() vectors = embeddings. embed_documents([x.page_content for x in docs]) # Silhoute Score n_clusters_range = range(2, 11) silhouette_scores = [] for i in n_clusters_range: kmeans = KMeans(n_clusters=i, init='k-means++', max_iter=300, n_init=10, random_state=0) kmeans.fit(vectors) score = silhouette_score(vectors, kmeans.labels_) silhouette_scores.append(score) optimal_n_clusters = n_clusters_range[np.argmax(silhouette_scores)] # n_clusters = 5 kmeans = KMeans(n_clusters=optimal_n_clusters, random_state=42).fit(vectors) # Getting documents closers to centeriod closest_indices = [] # Loop through the number of clusters you have for i in range(optimal_n_clusters): # Get the list of distances from that particular cluster center distances = np.linalg.norm( vectors - kmeans.cluster_centers_[i], axis=1) # Find the list position of the closest one (using argmin to find the smallest distance) closest_index = np.argmin(distances) # Append that position to your closest indices list closest_indices.append(closest_index) sorted_indices = sorted(closest_indices) selected_docs = [docs[doc] for doc in sorted_indices] print('Selected Docs size : ' ,len(selected_docs)) return selected_docs def chunk_docs(self, docs): text_splitter = RecursiveCharacterTextSplitter( chunk_size=self.chunks, chunk_overlap=50, length_function=len, is_separator_regex=False, ) splitted_document = text_splitter.split_documents(docs) return splitted_document def get_key_information_stuff(self): prompt_template = """ Extract Key Informtion from the text below. This key information can include People Names & their Role/rank, Locations, Organization,Nationalities,Religions, Events such as Historical, social, sporting and naturally occurring events, Products , Address & email, URL, Date & Time, Provide the list of Key information each should be labeled with thier crossponding category.if key information related to category is not present,add "Not mentioned" in the response. {text} """ prompt = PromptTemplate( template=prompt_template, input_variables=['text']) return prompt def get_key_information_map_reduce(self): map_prompts = """ Extract Key Informtion from the text below. This key information can include People Names & their Role/rank, Locations, Organization,Nationalities,Religions, Events such as Historical, social, sporting and naturally occurring events, Products , Address & email, URL, Date & Time, Provide the list of Key information each should be labeled with thier crossponding category.if key information related to category is not present, add Not mentioned in the response. {text} """ combine_prompt = """ Below Text contains Key Information that was extracted from text. You job is to combine the Key Information and Return the results.This key information can include People Names & their Role/rank, Locations, Organization,Nationalities,Religions,Events such as Historical, social, sporting and naturally occurring events, Products , Address & email, URL, Date & Time, Provide the list of Key information each should be labeled with thier crossponding category. if key information related to category is not present, add Not mentioned in the response. {text} """ map_template = PromptTemplate(template=map_prompts,input_variables=['text'] ) # combine_template = PromptTemplate(template=combine_prompt,input_variables=['Summary_type','Summary_strategy','Target_Person_type','Response_lenght','Writing_style','text'] # ) combine_template = PromptTemplate(template=combine_prompt,input_variables=['text']) return map_template, combine_template def get_stuff_prompt(self): prompt_template = """ Write a {Summary_type} and {Summary_strategy} for {Target_Person_type} lenght of the summary should be of {Response_length} words and writing style should be of {Writing_style}. From the text below by identifying most important topics based on their importance in text corpus and summary should be based on these important topics. {text} """ # prompt = PromptTemplate.from_template(prompt_template,input_variables=['Summary_type','Summary_strategy','Target_Person_type','Response_lenght','Writing_style','text']) prompt = PromptTemplate( template=prompt_template, input_variables=['Summary_type','Summary_strategy','Target_Person_type','Response_length','Writing_style','text']) return prompt def define_prompts(self): map_prompts = """ "Identify the key topics in the following text. in your response only add the most relevant and most important topics and Concised yet eloborative summary of text below. Dont add all the topics that you find.if you didnt find any important topic,dont return anything in response.Also provide me importance score of each idenfied topics out of 1. 'Your response should be like this , eg: Summary of text: blah blah blah,list of comma saperated topic names `Topic 1 Topic 2 Topic 3` and list of comma saperated importance scores for these topics `1 , 0.5,0.2`, so response should be formated like this. Summary: blah Blah blah Topic Names : Topic 1, Topic 2, Topic 3 Importance Score: 1,0.4,0.3 {text} """ combine_prompt = """ Here is list of summaries ,Topics Names and thier respective importance score that were extracted from text. your job is to provide best possible summary based on the list of summaries below and Use most important topics present based on thier importance score. Write a {Summary_type} and {Summary_strategy} for {Target_Person_type} lenght of the summary should be of {Response_length} words and writing style should be of {Writing_style}. {text} output Format should be like this.Dont try Return to multiple summaries.Only return one combined summary for above mentioned summaries. Summary: blah blah blah """ map_template = PromptTemplate(template=map_prompts, input_variables=['text'] ) combine_template = PromptTemplate( template=combine_prompt, input_variables=['Summary_type','Summary_strategy','Target_Person_type','Response_length','Writing_style','text']) return map_template, combine_template # pass def define_chain(self,Summary_type,Summary_strategy, Target_Person_type,Response_length,Writing_style,chain_type=None,key_information=False): docs, chain_type = self.load_docs() llm = ChatOpenAI(model='gpt-3.5-turbo', temperature=0) if chain_type == 'stuff': if key_information: prompt = self.get_key_information_stuff() else: prompt = self.get_stuff_prompt() chain = load_summarize_chain( llm=llm, chain_type='stuff', verbose=False,prompt=prompt) elif chain_type == 'map_reduce': if key_information: map_prompts, combine_prompt = self.get_key_information_map_reduce() else: map_prompts, combine_prompt = self.define_prompts() chain = load_summarize_chain( llm=llm, map_prompt=map_prompts, combine_prompt=combine_prompt, chain_type='map_reduce', verbose=False) # elif chain_type == 'refine': # chain = load_summarize_chain(llm=llm, question_prompt=map_prompts, # refine_prompt=combine_prompt, chain_type='refine', verbose=False) if ~key_information: output = chain.run(Summary_type=Summary_type,Summary_strategy=Summary_strategy, Target_Person_type=Target_Person_type,Response_length=Response_length,Writing_style=Writing_style,input_documents = docs) else: output = chain.run(input_documents = docs) # self.create_wordcloud(output=output) # display(Markdown(f"Text: {docs}")) # display(Markdown(f"Summary Response: {output}")) return output def parse_key_information(self,text): lines = [line.strip() for line in text.split('\n') if line.strip()] # Initialize the dictionary to store information info_dict = {} current_category = None # Iterate through each line and process the information for line in lines: if re.match(r'^[A-Z][\w\s&/-]*:', line): current_category = line.rstrip(':') info_dict[current_category] = [] else: if line != '- Not mentioned': info_dict[current_category].append(line.replace('- ', '')) # Remove categories with no entries info_dict = {category: entries for category, entries in info_dict.items() if entries} return info_dict # def create_wordcloud(self, output): # wc = WordCloud(stopwords=STOPWORDS, height=500, width=300) # wc.generate(output) # wc.to_file('WordCloud.png') def create_word_count(text): # Split the text into words, convert them to lowercase words = text.split() words = [word.lower() for word in words] # Get a list of English stop words stop_words = set(stopwords.words('english')) # Filter out stop words from the list of words filtered_words = [word for word in words if word not in stop_words] # Count the frequencies of each word word_counts = Counter(filtered_words) # Convert the Counter object to a dictionary word_count_dict = dict(word_counts) return word_count_dict class AudioBookNarration: def __init__(self,text_input ,file_path=None, chunks=2000, chunking_strategy=None, LLM_Model="gpt-3.5-turbo", temperature=1, top_p=None, top_k=None): self.chunks = chunks self.file_path = file_path self.text_input = text_input self.chuking_strategy = chunking_strategy self.LLM_Model = LLM_Model self.temperature = temperature self.top_p = top_p self.top_k = top_k def doc_summary(self, docs): # print(f'You have {len(docs)} documents') num_words = sum([len(doc.page_content.split(" ")) for doc in docs]) # print(f"You have {num_words} words in documents") return num_words, len(docs) def load_docs(self): if self.file_path is not None: docs = DirectoryLoader(self.file_path, glob="**/*.txt").load() else: docs = Document(page_content=f"{self.text_input}", metadata={"source": "local"}) docs = [docs] # docs = self.text_input tokens, documents_count = self.doc_summary(docs) if documents_count > 8 or tokens > 6000: ## Add token checks as well. Add Model availabilty checks docs = self.chunk_docs(docs) ## Handling Large Document with token more than 6000 docs = self.summarise_large_documents(docs) tokens, documents_count = self.doc_summary(docs) if tokens > 2000: docs = self.chunk_docs(docs) chain_type = 'map_reduce' else: chain_type = 'stuff' print("=="*20) print(tokens) print(chain_type) return docs, chain_type ## Add ensemble retriver for this as well. def summarise_large_documents(self, docs): print("=="*20) print('Orignial Docs size : ' ,len(docs)) embeddings = OpenAIEmbeddings() vectors = embeddings. embed_documents([x.page_content for x in docs]) # Silhoute Score n_clusters_range = range(2, 11) silhouette_scores = [] for i in n_clusters_range: kmeans = KMeans(n_clusters=i, init='k-means++', max_iter=300, n_init=10, random_state=0) kmeans.fit(vectors) score = silhouette_score(vectors, kmeans.labels_) silhouette_scores.append(score) optimal_n_clusters = n_clusters_range[np.argmax(silhouette_scores)] # n_clusters = 5 kmeans = KMeans(n_clusters=optimal_n_clusters, random_state=42).fit(vectors) # Getting documents closers to centeriod closest_indices = [] # Loop through the number of clusters you have for i in range(optimal_n_clusters): # Get the list of distances from that particular cluster center distances = np.linalg.norm( vectors - kmeans.cluster_centers_[i], axis=1) # Find the list position of the closest one (using argmin to find the smallest distance) closest_index = np.argmin(distances) # Append that position to your closest indices list closest_indices.append(closest_index) sorted_indices = sorted(closest_indices) selected_docs = [docs[doc] for doc in sorted_indices] print('Selected Docs size : ' ,len(selected_docs)) return selected_docs def chunk_docs(self, docs): text_splitter = RecursiveCharacterTextSplitter( chunk_size=self.chunks, chunk_overlap=50, length_function=len, is_separator_regex=False, ) splitted_document = text_splitter.split_documents(docs) return splitted_document def get_stuff_prompt(self): prompt_template = """ Create a {Narration_style} narration for this below text. This narration will be used for audiobook generation. So provide the output that is verbose, easier to understand and full of expressions. {text} """ prompt = PromptTemplate( template=prompt_template, input_variables=['Narration_style','text']) return prompt def define_prompts(self): map_prompts = """ Create a {Narration_style} narration for this below text. This narration will be used for audiobook generation. So provide the output that is verbose, easier to understand and full of expressions. {text} """ combine_prompt = """ Below are the list of text that represent narration from the text. Your job is to combine these narrations and craete one verbose,easier to understand and full of experssions {Narration_style} narration. {text} """ map_template = PromptTemplate(template=map_prompts, input_variables=['Narration_style','text'] ) combine_template = PromptTemplate( template=combine_prompt, input_variables=['Narration_style','text']) return map_template, combine_template # pass def define_chain(self,Narration_style=None,chain_type=None): docs, chain_type = self.load_docs() llm = ChatOpenAI(model='gpt-3.5-turbo', temperature=0) if chain_type == 'stuff': prompt = self.get_stuff_prompt() chain = load_summarize_chain( llm=llm, chain_type='stuff', verbose=False,prompt=prompt) elif chain_type == 'map_reduce': map_prompts, combine_prompt = self.define_prompts() chain = load_summarize_chain( llm=llm, map_prompt=map_prompts, combine_prompt=combine_prompt, chain_type='map_reduce', verbose=False) output = chain.run(Narration_style = Narration_style,input_documents = docs) # self.create_wordcloud(output=output) # display(Markdown(f"Text: {docs}")) # display(Markdown(f"Summary Response: {output}")) return output