AgentsWithoutBorders / rag_data_layer.py
Landhoff's picture
Upload rag_data_layer.py
adb343e verified
raw
history blame
27.4 kB
# -*- 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.
"""