Update Jul 25

FC_tool_main.py

@@ -7,26 +7,22 @@ the LangChain library for natural language processing tasks and the YouTube Tran
 API for fetching video transcripts.
 
 Classes:
-    MainPointsExtractor: 
-        Extracts and formats main points from YouTube video transcripts. 
-        Timestamps are formatted for direct use in YouTube comments, enabling clickable 
-        links to specific video sections when pasted.
-    SummaryExtractor: 
-        Handles the extraction and formatting of video summaries.
+    YouTubeTranscriptPointsExtractor: 
+        Extracts and formats comments with clickable timestamps from a YouTube video transcript.
     QuestionAnswerExtractor: 
         Processes user questions and extracts answers from video transcripts.
     YouTubeAgent: 
         Manages the overall agent setup for interacting with YouTube videos and processing user queries.
 
 Key Features:
-    - Main points summarization in multiple formats
-    - Video content summarization
+    - Main points formatted as youtube comment with clickable timestamps
     - Question answering based on video content
     - Flexible AI agent for handling various YouTube video-related tasks
 """
 
 import os
 import openai
+import json
 from typing import List, Dict, Any, Union, Type
 from youtube_transcript_api import YouTubeTranscriptApi
 from langchain_core.pydantic_v1 import BaseModel, Field
@@ -40,7 +36,6 @@ from langchain_core.utils.function_calling import convert_to_openai_function
 from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
 from langchain.agents.format_scratchpad import format_to_openai_functions
 from langchain.memory import ConversationBufferWindowMemory
-from dotenv import load_dotenv, find_dotenv
 
 # _ = load_dotenv(find_dotenv()) # read local .env file
 openai.api_key = os.getenv('OPENAI_API_KEY') #os.environ['OPENAI_API_KEY']
@@ -55,59 +50,63 @@ def set_temperature(new_temperature):
     get_temperature = new_get_temperature
     # print(f"Temperature set to: {get_temperature()}")
 
-class Points_1(BaseModel):
+class TimestampedPoint_1(BaseModel):
     """Pydantic model for representing extracted points from Youtube-Transcript"""
-    timestamp: float = Field(description="The timestamp (in floating-point number) of when main points are discussed or talked about in the video.")
+    timestamp: float = Field(description="The timestamp (in floating-point number) of when main points are discussed in the video.")
     main_point: str = Field(description="A title for Main point.")
-    summary: str = Field(description="A summary of main points discussed at that timestamp. End with fullstop.")
+    summary: str = Field(description="A summary of main points discussed at that timestamp.")
     emoji: str = Field(description="An emoji that matches the summary.")
-
-class Points_2(BaseModel):
+    
+class TimestampedPoint_2(BaseModel):
     """Pydantic model for representing extracted points."""
     main_point: str = Field(description="The main topic, theme, or subject extracted from the subtitle.")
+    timestamp: float = Field(description="The timestamp (in floating-point number) from the video where the main point is mentioned.")
     summary: str = Field(description="The context or brief explanation of the main point.")
     emoji: str = Field(description="An emoji that represents or summarizes the main point.")
-    timestamp: float = Field(description="The timestamp (in floating-point number) from the video where the main point is mentioned.")
-
-class MainPointsExtractor:
+    
+class YouTubeTranscriptPointsExtractor:
     """
-    A tool for extracting and formatting main points from YouTube video transcripts.
+    A tool for extracting and formatting main points with clickable timestamps from YouTube video transcripts.
 
-    This class provides methods to process transcripts and identify key points
-    using natural language processing techniques.
+    This class provides methods to process transcripts, identify key points,
+    and format them for use in YouTube comments with clickable timestamps.
     """
 
-    class Info_1(BaseModel):
-        """Pydantic model for representing a collection of points."""
-        points: List[Points_1]
+    class PointsCollection_1(BaseModel):
+        """Pydantic model for representing a collection of timestamped points."""
+        points: List[TimestampedPoint_1]
     
-    class Info_2(BaseModel):
-        """Pydantic model for representing a collection of points."""
-        points: List[Points_2]
+    class PointsCollection_2(BaseModel):
+        """Pydantic model for representing a collection of timestamped points."""
+        points: List[TimestampedPoint_2]
     
     @staticmethod
     @tool(return_direct=True)
-    def get_youtube_video_main_points(youtube_video_id: str) -> str:
+    def extract_clickable_points(youtube_video_id: str) -> str:
         """
-        Extracts and formats main points with Timestamps from YouTube video transcripts. Timestamps are formatted for direct use in YouTube comments, enabling clickable links to specific video sections when pasted.
+        Extracts and formats comments with clickable timestamps from a YouTube video transcript.
 
         Args:
             youtube_video_id (str): The ID of the YouTube video.
 
         Returns:
-            str: Formatted string of main points extracted from the video.
+            str: Formatted string of main points with clickable timestamps, ready for use in YouTube comments.
         """
         try:
-            transcript = MainPointsExtractor._get_youtube_video_transcript(youtube_video_id)
-            main_points_1 = MainPointsExtractor._extract_main_points(transcript, MainPointsExtractor.Info_1)
-            main_points_2 = MainPointsExtractor._extract_main_points(transcript, MainPointsExtractor.Info_2)
-            formatted_output = f"""Main points extracted from YouTube video (ID: {youtube_video_id})\nStyle_1:\n```\n{main_points_2}\n```\nStyle_2:\n```\n{main_points_1}\n```\nChoose the style that best suits your needs for presenting the main points of the video."""
-            return formatted_output
+            transcript = YouTubeTranscriptPointsExtractor._fetch_transcript(youtube_video_id)
+            extracted_points_1 = YouTubeTranscriptPointsExtractor._process_transcript(transcript, YouTubeTranscriptPointsExtractor.PointsCollection_1)
+            formatted_output_1 = YouTubeTranscriptPointsExtractor._format_for_youtube_comment(extracted_points_1, True)
+            formatted_output_1a = YouTubeTranscriptPointsExtractor._format_for_youtube_comment(extracted_points_1, False)
+            
+            extracted_points_2 = YouTubeTranscriptPointsExtractor._process_transcript(transcript, YouTubeTranscriptPointsExtractor.PointsCollection_2)
+            formatted_output_2 = YouTubeTranscriptPointsExtractor._format_for_youtube_comment(extracted_points_2, True)
+            formatted_output_2a = YouTubeTranscriptPointsExtractor._format_for_youtube_comment(extracted_points_2, False)
+            return f"""Main points extracted from YouTube video (ID: {youtube_video_id})\nOutput_style_1:\n```\n{formatted_output_1}\n```\nOutput_Style_1a:\n```\n{formatted_output_1a}\n```\nOutput_Style_2a:\n```\n{formatted_output_2}\n```\nOutput_Style_2a:\n```\n{formatted_output_2a}\n```\nChoose the style that best suits your needs for presenting the main points of the video."""
         except Exception as e:
             raise
 
     @staticmethod
-    def _get_youtube_video_transcript(youtube_video_id: str) -> str:
+    def _fetch_transcript(youtube_video_id: str) -> str:
         """
         Fetches the transcript for a YouTube video.
 
@@ -128,12 +127,12 @@ class MainPointsExtractor:
             raise
 
     @staticmethod
-    def _extract_main_points(transcript: str, info_model: Union[Type[Info_1], Type[Info_2]]) -> List[Dict[str, Any]]:
+    def _process_transcript(transcript: str, info_model: Union[Type[PointsCollection_1], Type[PointsCollection_2]]) -> List[Dict[str, Any]]:
         """
         Extracts main points from the transcript using NLP techniques.
         
         This method maintains a conversation history to provide context for subsequent calls.
-
+        
         Args:
             transcript (str): The full transcript of the video.
         
@@ -143,22 +142,19 @@ class MainPointsExtractor:
         main_points_extraction_function = [convert_to_openai_function(info_model)]
         
         model = ChatOpenAI(temperature=get_temperature())
-        
-        extraction_model = model.bind(functions=main_points_extraction_function)
+
+        extraction_model = model.bind(functions=main_points_extraction_function, function_call={"name": info_model.__name__})
         
         system_message = f"""
-        You are an AI assistant that extracts info from video transcripts.
-        When extracting info, ensure that:
-        1. Each point has a unique timestamp.
+        You are an AI assistant that extracts essential info from video transcripts.
+        You have the authority to make improvements as you see fit.
+        
+        Rules To Follow:
+        - Refining the summaries for clarity and conciseness.
+        - Adjusting emoji choices to better represent the content.
+        - Removing redundant information.
+        - Grouping two points into a single point if the timestamps are close enough.
         
-        In addition to these specific requirements, you have the authority to make other improvements as you see fit. This may include:
-
-        - Refining the summaries for clarity and conciseness
-        - Adjusting emoji choices to better represent the content
-        - Reorganizing points for better logical flow
-        - Removing redundant information
-        - Adding context where necessary
-
         Your goal is to produce a refined and accurate representation of the main points from the video transcript. Use your judgment to balance adherence to the specific rules with overall improvement of the extracted information.
         """
         
@@ -169,11 +165,11 @@ class MainPointsExtractor:
         
         extraction_chain = prompt | extraction_model | JsonKeyOutputFunctionsParser(key_name="points")
 
-        text_splitter = RecursiveCharacterTextSplitter(chunk_overlap=0, chunk_size=8192, separators=[f" {char}" for char in "123456789"])
+        text_splitter = RecursiveCharacterTextSplitter(chunk_overlap=0, chunk_size=16000, separators=[f" {char}" for char in "123456789"])
         
         prep = RunnableLambda(lambda x: [{"input": doc} for doc in text_splitter.split_text(x)])
         
-        chain = prep | extraction_chain.map() | MainPointsExtractor._flatten | MainPointsExtractor._format_youtube_comment
+        chain = prep | extraction_chain.map() | YouTubeTranscriptPointsExtractor._flatten
         
         result_1 = chain.invoke(transcript)
                 
@@ -183,17 +179,19 @@ class MainPointsExtractor:
     def _flatten(matrix):
         """Flattens a 2D list into a 1D list."""
         return [item for row in matrix for item in row]
-
+    
     @staticmethod
-    def _format_youtube_comment(json_data: List[Dict[str, Any]]) -> str:
+    def _format_for_youtube_comment(points: List[Dict[str, Any]], detailed: bool = True) -> str:
         """
-        Formats extracted main points into a YouTube-style comment.
+        Formats extracted main points into a YouTube-style comment with clickable timestamps.
 
         Args:
-            json_data (List[Dict[str, Any]]): List of dictionaries containing main points.
+            points (List[Dict[str, Any]]): List of dictionaries containing main points with timestamps.
+            detailed (bool): If True, returns a detailed format with emojis and summaries. 
+                             If False, returns a simpler format with just timestamps and main points.
 
         Returns:
-            str: Formatted string representing the main points as a YouTube comment.
+            str: Formatted string representing the main points as a YouTube comment with clickable timestamps.
         """
         def _format_timestamp(seconds):
             hours = int(seconds // 3600)
@@ -202,125 +200,24 @@ class MainPointsExtractor:
             return f"{hours:02}:{minutes:02}:{seconds:02}"
 
         formatted_comment = ""
-        for entry in json_data:
-            timestamp = _format_timestamp(entry['timestamp'])
-            emoji = entry['emoji']
-            summary = entry['summary']
+        for point in points:
+            timestamp = _format_timestamp(point['timestamp'])
+            main_point = point['main_point'].rstrip('.')
             
-            if entry['main_point'].endswith('.'):
-                point = entry['main_point'][:-1]
+            if detailed:
+                emoji = point['emoji']
+                summary = point['summary']
+                formatted_comment += f"{timestamp} {emoji} {main_point}: {summary}\n"
             else:
-                point = entry['main_point']
-                
-            formatted_comment += f"{timestamp} {emoji} {point}: {summary}\n"
+                formatted_comment += f"{timestamp} {main_point}\n"
             
         return formatted_comment.strip()
-    
-#######################################################################################################################################    
-
-class Summary(BaseModel):
-    """Pydantic model for representing extracted summary."""
-    summary: str = Field(description="Extract detailed information from the content.")
-
-class SummaryExtractor:
-    """
-    A tool for extracting and formatting summaries from YouTube video transcripts.
-
-    This class provides methods to process transcripts and generate concise summaries
-    using natural language processing techniques.
-    """
-    
-    class Info(BaseModel):
-        """Pydantic model for representing a collection of summaries."""
-        summary: List[Summary]
-
-    @staticmethod
-    @tool(return_direct=False)
-    def get_youtube_video_summary(youtube_video_id: str) -> str:
-        """
-        Extracts and formats a summary from a YouTube video transcript.
-
-        Args:
-            youtube_video_id (str): The ID of the YouTube video.
-
-        Returns:
-            str: Formatted string of the summary extracted from the video.
-        """
-        try:
-            transcript = SummaryExtractor._get_youtube_video_transcript(youtube_video_id)
-            summary = SummaryExtractor._extract_summary(transcript)
-            return SummaryExtractor._format_summary(summary)
-        except Exception as e:
-            return f"Error extracting summary: {str(e)}"
-
-    @staticmethod
-    def _get_youtube_video_transcript(youtube_video_id: str) -> str:
-        """
-        Fetches the transcript for a YouTube video.
-
-        Args:
-            youtube_video_id (str): The ID of the YouTube video.
-
-        Returns:
-            str: The full transcript of the video.
-
-        Raises:
-            Exception: If there's an error fetching the transcript.
-        """
-        try:
-            transcript_json = YouTubeTranscriptApi.get_transcript(youtube_video_id)
-            transcript_data = [entry['text'] for entry in transcript_json]
-            return " ".join(transcript_data)
-        except Exception as e:
-            raise
-
-    @staticmethod
-    def _extract_summary(transcript: str) -> List[Summary]:
-        """
-        Extracts a summary from a YouTube video transcript.
-
-        Args:
-            transcript (str): The full transcript of the video.
-
-        Returns:
-            Summary: A Summary object containing the extracted summary.
-        """
-        summary_extraction_function = [convert_to_openai_function(SummaryExtractor.Info)]
-        
-        model = ChatOpenAI(temperature=get_temperature())
-        
-        extraction_model = model.bind(functions=summary_extraction_function)
-        
-        prompt = ChatPromptTemplate.from_messages([("human", "{input}")])
-        
-        extraction_chain = prompt | extraction_model | JsonKeyOutputFunctionsParser(key_name="summary")
-        
-        text_splitter = RecursiveCharacterTextSplitter(chunk_overlap=0, chunk_size=8192, separators=[f" {char}" for char in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"])
-        
-        prep = RunnableLambda(lambda x: [{"input": doc} for doc in text_splitter.split_text(x)])
-        
-        chain = prep | extraction_chain.map() | MainPointsExtractor._flatten
-        
-        return chain.invoke(transcript)
-        
-    @staticmethod
-    def _format_summary(summaries: List[Summary]) -> str:
-        """
-        Formats the list of summaries into a single string.
-        Args:
-            summaries (List[Summary]): List of Summary objects.
-        Returns:
-            str: A formatted string containing all summaries.
-        """
-        return "\n\n".join([s["summary"] for s in summaries])
-
-#############################################################################################################################################################
 
 class Answer(BaseModel):
     """Pydantic model for representing an answer to a question."""
     answer: str = Field(description="The answer to the user's question based on the video transcript.")
     confidence: float = Field(description="A confidence score between 0 and 1 indicating how certain the model is about the answer.")
-
+    
 class QuestionAnswerExtractor:
     """
     A tool for answering questions about YouTube videos based on their transcripts.
@@ -349,7 +246,7 @@ class QuestionAnswerExtractor:
         try:
             transcript = QuestionAnswerExtractor._get_youtube_video_transcript(youtube_video_id)
             answer = QuestionAnswerExtractor._extract_answer(transcript, question)
-            return QuestionAnswerExtractor._format_answer(answer)
+            return answer
         except Exception as e:
             return f"Error answering question: {str(e)}"
 
@@ -384,7 +281,7 @@ class QuestionAnswerExtractor:
             question (str): The user's question about the video.
         
         Returns:
-            List[Answer]: A list of Answer objects containing the extracted answers.
+            List[Answer]: A list containing a single Answer object with the consolidated answer.
         """
         answer_extraction_function = [convert_to_openai_function(QuestionAnswerExtractor.Info)]
         
@@ -408,34 +305,37 @@ class QuestionAnswerExtractor:
         
         chain = prep | extraction_chain.map() | QuestionAnswerExtractor._flatten
         
-        return chain.invoke({"transcript": transcript, "question": question})
+        # Get partial answers
+        partial_answers = chain.invoke({"transcript": transcript, "question": question})
         
-    @staticmethod
-    def _flatten(matrix):
-        """Flattens a 2D list into a 1D list."""
-        return [item for row in matrix for item in row]
+        # Filter out low-confidence answers
+        filtered_answers = [answer for answer in partial_answers if answer['confidence'] > 0.4]
         
-    @staticmethod
-    def _format_answer(answers: List[Answer]) -> str:
-        """
-        Formats the list of answers into a single string.
+        # If all answers were filtered out, return a low-confidence "no answer" response
+        if not filtered_answers:
+            return "I couldn't find a reliable answer to your question based on the video transcript."
         
-        Args:
-            answers (List[Answer]): List of Answer objects.
-
-        Returns:
-            str: A formatted string containing the best answer and its confidence score.
-        """
-        if not answers:
-            return "I couldn't find an answer to your question based on the video transcript."
+        # Consolidate filtered partial answers
+        consolidation_prompt = ChatPromptTemplate.from_messages([
+            ("system", "You are an AI assistant tasked with consolidating multiple partial answers into a comprehensive final answer."),
+            ("human", "Question: {question}\n\nPartial Answers: {partial_answers}\n\nPlease provide a consolidated, comprehensive answer to the question based on these partial answers. Ignore any information from answers with low confidence (0.5 or below).")
+        ])
         
-        # Sort answers by confidence score and take the best one
-        best_answer = max(answers, key=lambda x: x['confidence'])
+        consolidation_model = ChatOpenAI(temperature=get_temperature())
+        consolidation_chain = consolidation_prompt | consolidation_model 
         
-        return f"{best_answer['answer']}({best_answer['confidence']:.2f})"
-
-#######################################################################################################################################
-
+        final_answer = consolidation_chain.invoke({
+            "question": question,
+            "partial_answers": json.dumps(filtered_answers, indent=2)
+        })
+        
+        return final_answer.content
+        
+    @staticmethod
+    def _flatten(matrix):
+        """Flattens a 2D list into a 1D list."""
+        return [item for row in matrix for item in row]
+    
 class YouTubeAgent:
     """
     An agent for interacting with YouTube videos and processing user queries.
@@ -448,12 +348,19 @@ class YouTubeAgent:
         """Initializes the YouTubeAgent with necessary tools and components."""
         
         self.tools = [
-            MainPointsExtractor.get_youtube_video_main_points,
-            SummaryExtractor.get_youtube_video_summary,
-            QuestionAnswerExtractor.get_answer
+            QuestionAnswerExtractor.get_answer,
+            YouTubeTranscriptPointsExtractor.extract_clickable_points,
         ]
         
-        self.sys_message = "You are a helpful assistant."
+        self.sys_message = """You are a helpful assistant specialized in processing YouTube video transcripts and answering questions about video content.'
+
+        Important instructions:
+        1. Only use the 'extract_clickable_points' tool when the user explicitly asks for clickable points or timestamps from a video.
+        2. For all other queries, including general questions about video content, use the 'get_answer' tool.
+        3. If the user's query is unclear, ask for clarification before using any tools.
+        4. Always provide concise and relevant responses based on the tool outputs.
+
+        Remember to interpret the user's intent carefully and use the appropriate tool for each situation."""
         
         self.functions = [convert_to_openai_function(f) for f in self.tools]
         

app.py

@@ -66,10 +66,8 @@ with gr.Blocks() as demo:
     API for fetching video transcripts.
     
     Key Features:
-    - Main points summarization in multiple formats
-    - Video content summarization
+    - Main points formatted as youtube comment with clickable timestamps
     - Question answering based on video content
-    - Flexible AI agent for handling various YouTube video-related tasks
 
     Simply enter your question or request along with a YouTube video link, and the AI will process and respond accordingly.
     Adjust the temperature slider to control the creativity of the AI's responses.