# -*- coding: utf-8 -*- """RAG Data Layer.ipynb Automatically generated by Colab. Original file is located at https://colab.research.google.com/drive/17nEwUcytqDID3-27YOGBv36tRhLj3E3p """ """# Task Build a RAG data layer by loading and processing data from "knowledge_base.json", "video_transcript.json", and "youtube_transcripts (1).json", then consolidating and exporting the processed chunks into a unified "rag_data_layer.json" file. ## Load and Process Knowledge Base ### Subtask: Load the 'knowledge_base.json' file, process each item to extract relevant information, derive language and topic, assign a unique ID, and store the processed chunks. **Reasoning**: First, I'll import the `json` module to handle JSON file operations. Then, I will load the `knowledge_base.json` file, initialize an empty list to store processed chunks, and iterate through each item to extract, transform, and derive the required information such as URL, text, source type, language, topic, and a unique ID, as per the instructions. """ import json # 1. Load the content of 'knowledge_base.json' with open('/content/knowledge_base.json', 'r') as f: knowledge_base_data = json.load(f) # 2. Initialize an empty list to store the processed knowledge base chunks. processed_kb_chunks = [] # 3. Iterate through each item and process it for i, item in enumerate(knowledge_base_data): processed_chunk = {} # Extract 'url' processed_chunk['url'] = item.get('url') # Assign 'content' to 'text' processed_chunk['text'] = item.get('content') # Set 'source_type' processed_chunk['source_type'] = 'blog_article' # Determine 'language' if processed_chunk['url'] and '/ca/' in processed_chunk['url']: processed_chunk['language'] = 'ca' else: processed_chunk['language'] = 'es' # Derive 'topic' topic = None if processed_chunk['url']: # Remove 'https://blog.example.com/' or 'https://blog.example.com/ca/' to get the path path = processed_chunk['url'].replace('https://blog.example.com/ca/', '').replace('https://blog.example.com/', '') path_parts = path.split('/') # Try to find a non-empty part that could be the topic for part in path_parts: if part and '.html' not in part and part != 'ca' and part != 'es': # Exclude empty strings and file extensions topic = part break if topic == None: # Fallback for URLs that might not have a clear topic segment before an article name # If the path looks like domain/language/article-name.html, take the part before .html if len(path_parts) > 0 and '.html' in path_parts[-1]: topic = path_parts[-1].split('.html')[0].replace('-',' ') elif len(path_parts) > 1 and '.html' in path_parts[-2]: topic = path_parts[-2].split('.html')[0].replace('-',' ') if topic and topic.endswith('.html'): # Clean up if .html is still there topic = topic.replace('.html', '') if topic and '-' in topic and ' ' not in topic: # Replace hyphens with spaces if it looks like a slug topic = topic.replace('-', ' ') processed_chunk['topic'] = topic # Assign a unique 'id' processed_chunk['id'] = f"kb_{i+1}" processed_kb_chunks.append(processed_chunk) print(f"Processed {len(processed_kb_chunks)} knowledge base chunks.") # Display the first 3 processed chunks to verify print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_kb_chunks))): print(processed_kb_chunks[i]) """**Reasoning**: The next step is to load and process the `video_transcript.json` file, extracting relevant information and structuring it similarly to the `knowledge_base.json` data, while assigning a unique ID and identifying the language and topic. """ import re # Import regex for advanced topic extraction import json # Load the content of 'video_transcript.json' with open('/content/video_transcript.json', 'r') as f: video_transcript_data = json.load(f) # Initialize an empty list to store the processed video transcript chunks. processed_vt_chunks = [] # The video_transcript_data loaded from the file is a single dictionary, # not a list of dictionaries. Wrap it in a list to use the existing loop structure. items_to_process = [video_transcript_data] if isinstance(video_transcript_data, dict) else video_transcript_data # Iterate through each item and process it for i, item in enumerate(items_to_process): processed_chunk = {} # Extract 'url'. The 'video_transcript_data' example does not have a 'url' field directly. # Assigning None if not found, as per dict.get() behavior. processed_chunk['url'] = item.get('url') # Assign 'text' from the 'text' key, not 'content' processed_chunk['text'] = item.get('text') # Set 'source_type' processed_chunk['source_type'] = 'video_transcript' # Determine 'language'. Prioritize the 'language' field from the item itself if available. if item.get('language'): processed_chunk['language'] = item.get('language') elif processed_chunk['url'] and '/ca/' in processed_chunk['url']: processed_chunk['language'] = 'ca' else: # Fallback if neither 'language' key nor URL contains language info processed_chunk['language'] = 'es' # Defaulting to 'es' as per original logic # Derive 'topic' from the URL or text topic = None if processed_chunk['url']: # Original URL parsing logic (will likely not be triggered as 'url' is None for this data) path_parts = [part for part in processed_chunk['url'].split('/') if part] try: videos_index = path_parts.index('videos') if len(path_parts) > videos_index + 2 and path_parts[videos_index + 1] in ['ca', 'es']: topic = path_parts[videos_index + 2] elif len(path_parts) > videos_index + 1: topic = path_parts[videos_index + 1] if topic: topic = topic.replace('-', ' ').replace('.html', '').strip() except ValueError: pass # Fallback if topic extraction from URL is not successful or URL is None if not topic and processed_chunk['text']: # Taking the first sentence as a fallback topic first_sentence = processed_chunk['text'].split('.')[0] if len(first_sentence) > 5: # Avoid very short sentences as topics topic = first_sentence.strip() else: topic = "General Video Topic" # A more generic fallback if first sentence is too short processed_chunk['topic'] = topic # Assign a unique 'id' processed_chunk['id'] = f"vt_{i+1}" processed_vt_chunks.append(processed_chunk) print(f"Processed {len(processed_vt_chunks)} video transcript chunks.") # Display the first 3 processed chunks to verify print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_vt_chunks))): print(processed_vt_chunks[i]) import json # Define the output file name output_file_name = 'rag_data_layer.json' # Load the content of 'rag_data_layer.json' with open(output_file_name, 'r', encoding='utf-8') as f: rag_data_layer = json.load(f) print(f"Total chunks loaded from {output_file_name}: {len(rag_data_layer)}") # Display the first 5 chunks to verify the content print("\nFirst 5 chunks from rag_data_layer.json:") for i in range(min(5, len(rag_data_layer))): print(rag_data_layer[i]) """**Reasoning**: The previous code failed because it attempted to iterate over a dictionary as if it were a list of dictionaries, causing `item` to be a string (the dictionary key) which lacks the `.get()` method. Additionally, the structure of `video_transcript_data` shows a 'text' key instead of 'content' and has a 'language' key which should be prioritized. I will wrap the single dictionary into a list for iteration, update the key access for text, and refine the language extraction to use the provided 'language' key. """ import re # Import regex for advanced topic extraction import json # Load the content of 'video_transcript.json' with open('/content/video_transcript.json', 'r') as f: video_transcript_data = json.load(f) # Initialize an empty list to store the processed video transcript chunks. processed_vt_chunks = [] # The video_transcript_data loaded from the file is a single dictionary, # not a list of dictionaries. Wrap it in a list to use the existing loop structure. items_to_process = [video_transcript_data] if isinstance(video_transcript_data, dict) else video_transcript_data # Iterate through each item and process it for i, item in enumerate(items_to_process): processed_chunk = {} # Extract 'url'. The 'video_transcript_data' example does not have a 'url' field directly. # Assigning None if not found, as per dict.get() behavior. processed_chunk['url'] = item.get('url') # Assign 'text' from the 'text' key, not 'content' processed_chunk['text'] = item.get('text') # Set 'source_type' processed_chunk['source_type'] = 'video_transcript' # Determine 'language'. Prioritize the 'language' field from the item itself if available. if item.get('language'): processed_chunk['language'] = item.get('language') elif processed_chunk['url'] and '/ca/' in processed_chunk['url']: processed_chunk['language'] = 'ca' else: # Fallback if neither 'language' key nor URL contains language info processed_chunk['language'] = 'es' # Defaulting to 'es' as per original logic # Derive 'topic' from the URL or text topic = None if processed_chunk['url']: # Original URL parsing logic (will likely not be triggered as 'url' is None for this data) path_parts = [part for part in processed_chunk['url'].split('/') if part] try: videos_index = path_parts.index('videos') if len(path_parts) > videos_index + 2 and path_parts[videos_index + 1] in ['ca', 'es']: topic = path_parts[videos_index + 2] elif len(path_parts) > videos_index + 1: topic = path_parts[videos_index + 1] if topic: topic = topic.replace('-', ' ').replace('.html', '').strip() except ValueError: pass # Fallback if topic extraction from URL is not successful or URL is None if not topic and processed_chunk['text']: # Taking the first sentence as a fallback topic first_sentence = processed_chunk['text'].split('.')[0] if len(first_sentence) > 5: # Avoid very short sentences as topics topic = first_sentence.strip() else: topic = "General Video Topic" # A more generic fallback if first sentence is too short processed_chunk['topic'] = topic # Assign a unique 'id' processed_chunk['id'] = f"vt_{i+1}" processed_vt_chunks.append(processed_chunk) print(f"Processed {len(processed_vt_chunks)} video transcript chunks.") # Display the first 3 processed chunks to verify print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_vt_chunks))): print(processed_vt_chunks[i]) """## Load and Process YouTube Transcripts ### Subtask: Load the 'youtube_transcripts (1).json' file, process each item to extract relevant information, derive language and topic, assign a unique ID, and store the processed chunks. **Reasoning**: The next step is to load the 'youtube_transcripts (1).json' file, extract the relevant information, and process it into a structured format, similar to how the previous data sources were handled. This includes assigning a unique ID, identifying the language, and deriving a topic for each transcript. """ import re import json # Load the content of 'youtube_transcripts (1).json' with open('/content/youtube_transcripts (1).json', 'r') as f: youtube_transcript_data = json.load(f) # Initialize an empty list to store the processed YouTube transcript chunks. processed_yt_chunks = [] # Iterate through each item and process it for i, item in enumerate(youtube_transcript_data): processed_chunk = {} # Extract 'url' processed_chunk['url'] = item.get('url') # Assign 'text' from the 'text' key processed_chunk['text'] = item.get('text') # Set 'source_type' processed_chunk['source_type'] = 'youtube_transcript' # Determine 'language'. Prioritize the 'language' field from the item itself if available. if item.get('language'): processed_chunk['language'] = item.get('language') elif processed_chunk['url'] and '/ca/' in processed_chunk['url']: processed_chunk['language'] = 'ca' else: # Fallback if neither 'language' key nor URL contains language info processed_chunk['language'] = 'es' # Defaulting to 'es' # Derive 'topic' from the URL or text topic = None if processed_chunk['url']: path_parts = [part for part in processed_chunk['url'].split('/') if part] try: # Assuming YouTube URLs might contain 'watch?v=' and the video ID is after that # Or, if there's a more structured path for topics, use that. if 'watch?v=' in processed_chunk['url']: # Attempt to get a topic from a 'title' or 'description' if available in the item # This JSON doesn't seem to have a title/description, so falling back to text. pass # Will use the text fallback else: # Try to extract from path segments like blog_article or video_transcript # Look for a part after 'videos' and language, but before any video ID or file extension if len(path_parts) > 2 and path_parts[1] in ['ca', 'es']: topic = path_parts[2] elif len(path_parts) > 1: topic = path_parts[1] if topic: topic = topic.replace('-', ' ').replace('.html', '').strip() except Exception: pass # If any error during URL parsing, fallback to text # Fallback if topic extraction from URL is not successful or URL is None if not topic and processed_chunk['text']: # Taking the first sentence as a fallback topic first_sentence = processed_chunk['text'].split('.')[0] if len(first_sentence) > 5: # Avoid very short sentences as topics topic = first_sentence.strip() else: topic = "General YouTube Topic" # A more generic fallback processed_chunk['topic'] = topic # Assign a unique 'id' processed_chunk['id'] = f"yt_{i+1}" processed_yt_chunks.append(processed_chunk) print(f"Processed {len(processed_yt_chunks)} YouTube transcript chunks.") # Display the first 3 processed chunks to verify print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_yt_chunks))): print(processed_yt_chunks[i]) """**Reasoning**: The previous code block failed to extract 'url' and 'text' correctly from the YouTube transcript data because it was looking for keys 'url' and 'text' instead of 'video_url' and 'content', respectively. I need to update the key names to match the actual structure of the `youtube_transcript_data` and also improve the topic extraction for YouTube URLs, as the topic is not directly available in the URL path. """ import re import json # Load the content of 'youtube_transcripts (1).json' with open('/content/youtube_transcripts (1).json', 'r') as f: youtube_transcript_data = json.load(f) # Initialize an empty list to store the processed YouTube transcript chunks. processed_yt_chunks = [] # Iterate through each item and process it for i, item in enumerate(youtube_transcript_data): processed_chunk = {} # Extract 'video_url' and assign to 'url' processed_chunk['url'] = item.get('video_url') # Assign 'content' to 'text' processed_chunk['text'] = item.get('content') # Set 'source_type' processed_chunk['source_type'] = 'youtube_transcript' # Determine 'language'. Prioritize the 'language' field from the item itself if available. # The example YouTube data doesn't have a 'language' key or '/ca/' in URL, so defaulting to 'es'. if item.get('language'): processed_chunk['language'] = item.get('language') elif processed_chunk['url'] and '/ca/' in processed_chunk['url']: processed_chunk['language'] = 'ca' else: processed_chunk['language'] = 'es' # Defaulting to 'es' # Derive 'topic' from the URL or text topic = None if processed_chunk['url']: # For YouTube URLs, a direct topic from path segments is unlikely. # Attempt to extract topic from the video_id or a general placeholder. match = re.search(r'v=([\w-]+)', processed_chunk['url']) if match: # Use video_id as a temporary topic or a placeholder, as no descriptive topic is in the URL path topic = f"YouTube Video ID: {match.group(1)}" # Further refinement of topic from URL might be needed if video titles are available. # Fallback if topic extraction from URL is not successful or URL is None if not topic and processed_chunk['text']: # Taking the first sentence as a fallback topic if more descriptive topic not found first_sentence = processed_chunk['text'].split('.')[0] if len(first_sentence) > 5: # Avoid very short sentences as topics topic = first_sentence.strip() else: topic = "General YouTube Topic" # A more generic fallback processed_chunk['topic'] = topic # Assign a unique 'id' processed_chunk['id'] = f"yt_{i+1}" processed_yt_chunks.append(processed_chunk) print(f"Processed {len(processed_yt_chunks)} YouTube transcript chunks.") # Display the first 3 processed chunks to verify print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_yt_chunks))): print(processed_yt_chunks[i]) """**Reasoning**: All data sources have been processed individually. The next logical step is to combine the processed chunks from the knowledge base, video transcripts, and YouTube transcripts into a single, unified list to prepare for exporting as the final RAG data layer. """ all_processed_chunks = [] all_processed_chunks.extend(processed_kb_chunks) all_processed_chunks.extend(processed_vt_chunks) all_processed_chunks.extend(processed_yt_chunks) print(f"Total processed chunks: {len(all_processed_chunks)}") # Display the first 5 unified chunks to verify print("\nFirst 5 unified chunks:") for i in range(min(5, len(all_processed_chunks))): print(all_processed_chunks[i]) """**Reasoning**: The processed data from all sources has been unified into a single list. The final step is to export this consolidated list of chunks into a JSON file, completing the RAG data layer. """ import json # Define the output file name output_file = 'rag_data_layer.json' # Export the unified list of processed chunks to a JSON file with open(output_file, 'w', encoding='utf-8') as f: json.dump(all_processed_chunks, f, ensure_ascii=False, indent=4) print(f"Successfully exported {len(all_processed_chunks)} chunks to {output_file}") """## Process Breathing Transcript ### Subtask: Load the 'video_transcript.json' file, extract the 'text' content, and assign specific fixed values for 'url', 'source_type', 'language', and 'topic', along with a unique ID. **Reasoning**: The subtask requires loading the 'video_transcript.json' file, extracting its 'text' content, and assigning specific fixed values for 'url', 'source_type', 'language', 'topic', and a unique ID to the processed chunk. This code block will perform these operations and store the result in a new list. """ import json # 1. Load the content of 'video_transcript.json' with open('/content/video_transcript.json', 'r') as f: breathing_transcript_data = json.load(f) # 2. Initialize an empty list to store the processed breathing chunks. processed_breathing_chunks = [] # 3. Create a dictionary to represent the processed chunk and extract 'text'. processed_chunk = {} processed_chunk['text'] = breathing_transcript_data.get('text') # 4. Assign the fixed 'url' processed_chunk['url'] = 'https://video.local/breathing_exercise' # 5. Assign the fixed 'source_type' processed_chunk['source_type'] = 'breathing_video' # 6. Assign the fixed 'language' processed_chunk['language'] = 'es' # 7. Assign the fixed 'topic' processed_chunk['topic'] = 'anxiety' # 8. Assign a unique 'id' processed_chunk['id'] = 'vt_1' # 9. Append the processed chunk to the list. processed_breathing_chunks.append(processed_chunk) # 10. Print the number of processed chunks and display the first processed chunk to verify. print(f"Processed {len(processed_breathing_chunks)} breathing transcript chunks.") print("\nFirst processed chunk:") print(processed_breathing_chunks[0]) """## Process Legal Webinar Transcript ### Subtask: Load the 'youtube_transcripts (1).json' file, extract the 'url' and 'text' content, assign fixed values for 'source_type', 'language', and 'topic', along with a unique ID. **Reasoning**: The subtask requires loading the 'youtube_transcripts (1).json' file, extracting specific content, and assigning fixed metadata fields. This code block will perform these operations as per the instructions, creating a list of processed legal webinar chunks with unique IDs. """ import json # 1. Load the content of 'youtube_transcripts (1).json' with open('/content/youtube_transcripts (1).json', 'r') as f: legal_webinar_data = json.load(f) # 2. Initialize an empty list to store the processed legal webinar chunks. processed_legal_webinar_chunks = [] # 3. Iterate through each item in the loaded YouTube transcript data. # 4. For each item, create a new dictionary to represent the processed chunk. for i, item in enumerate(legal_webinar_data): processed_chunk = {} # 5. Extract the value associated with the 'video_url' key from the item and assign it to the 'url' key. processed_chunk['url'] = item.get('video_url') # 6. Extract the value associated with the 'content' key from the item and assign it to the 'text' key. processed_chunk['text'] = item.get('content') # 7. Assign the string 'legal_webinar' to the 'source_type' key. processed_chunk['source_type'] = 'legal_webinar' # 8. Assign the string 'es' to the 'language' key. processed_chunk['language'] = 'es' # 9. Assign the string 'legal' to the 'topic' key. processed_chunk['topic'] = 'legal' # 10. Assign a unique identifier (e.g., 'lw_1', 'lw_2', etc.) to the 'id' key. processed_chunk['id'] = f"lw_{i+1}" # 11. Append the processed_chunk to the processed_legal_webinar_chunks list. processed_legal_webinar_chunks.append(processed_chunk) # 12. Print the total number of processed chunks and display the first few processed chunks to verify. print(f"Processed {len(processed_legal_webinar_chunks)} legal webinar chunks.") print("\nFirst 3 processed chunks:") for i in range(min(3, len(processed_legal_webinar_chunks))): print(processed_legal_webinar_chunks[i]) """## Consolidate and Export All Corrected Chunks ### Subtask: Combine all the corrected processed chunks from the knowledge base, breathing transcript, and legal webinar into a single list, and then export this unified list to a new JSON file. **Reasoning**: The subtask requires combining all corrected processed chunks from different sources into a single list and then exporting this unified list to a JSON file. I will initialize an empty list, extend it with each of the processed chunk lists, print the total count for verification, and finally, save the combined list to 'rag_data_layer.json' with proper JSON formatting. """ import json # 1. Initialize an empty list called final_rag_data_layer. final_rag_data_layer = [] # 2. Extend final_rag_data_layer with the contents of processed_kb_chunks. final_rag_data_layer.extend(processed_kb_chunks) # 3. Extend final_rag_data_layer with the contents of processed_breathing_chunks. final_rag_data_layer.extend(processed_breathing_chunks) # 4. Extend final_rag_data_layer with the contents of processed_legal_webinar_chunks. final_rag_data_layer.extend(processed_legal_webinar_chunks) # 5. Print the total number of chunks in final_rag_data_layer to verify the count. print(f"Total chunks in final RAG data layer: {len(final_rag_data_layer)}") # 6. Save final_rag_data_layer to a JSON file named 'rag_data_layer.json', ensuring proper formatting. output_file_name = 'rag_data_layer.json' with open(output_file_name, 'w', encoding='utf-8') as f: json.dump(final_rag_data_layer, f, ensure_ascii=False, indent=4) print(f"Successfully exported {len(final_rag_data_layer)} chunks to {output_file_name}") # Display the first few chunks to verify the content print("\nFirst 3 unified chunks:") for i in range(min(3, len(final_rag_data_layer))): print(final_rag_data_layer[i]) """## Final Task ### Subtask: Confirm that the RAG data layer has been successfully built and exported into a unified JSON file, ready for further use in the hackathon. ## Summary: ### Q&A Yes, the RAG data layer has been successfully built and exported into a unified JSON file named `rag_data_layer.json`, containing all processed data and ready for further use in the hackathon. ### Data Analysis Key Findings * **Knowledge Base Processing:** The `knowledge_base.json` file was loaded, and its content was processed to extract URLs, content (mapped to 'text'), source type ('blog\_article'), language (derived from URL as 'ca' or 'es'), and a derived topic. A total of 28 knowledge base chunks were successfully processed. * **Breathing Transcript Processing:** The `video_transcript.json` file was specifically processed as a "breathing transcript". Its 'text' content was extracted, and fixed metadata values were assigned: `url` as 'https://video.local/breathing\_exercise', `source_type` as 'breathing\_video', `language` as 'es', and `topic` as 'anxiety'. A single (1) breathing transcript chunk was processed. * **Legal Webinar Transcript Processing:** The `youtube_transcripts (1).json` file was specifically processed as a "legal webinar transcript". Its 'video\_url' was mapped to 'url', 'content' to 'text', and fixed metadata values were assigned: `source_type` as 'legal\_webinar', `language` as 'es', and `topic` as 'legal'. A single (1) legal webinar chunk was processed. * **Consolidation and Export:** All processed chunks from the knowledge base, breathing transcript, and legal webinar were consolidated into a single list. This unified list contained a total of 30 chunks. * **Final Output:** The consolidated data layer, comprising 30 chunks, was successfully exported to `rag_data_layer.json` in a well-formatted JSON structure. ### Insights or Next Steps * Consider implementing a more dynamic and robust topic extraction method, potentially leveraging natural language processing (NLP) models, to reduce reliance on URL patterns or fixed assignments, especially as data sources grow. * For future iterations, centralize the metadata schema and processing rules to ensure consistency and easier maintenance across diverse data sources, instead of having source-specific hardcoded values. """