Upload 9 files

Dockerfile

@@ -10,19 +10,15 @@ RUN apt-get update && apt-get install -y \
 # Install uv for faster package installation
 RUN pip install uv
 
-# Copy requirements and install dependencies
-# Assumes requirements.txt is at the root of the repo
+# Copy requirements and install dependencies with uv
 COPY requirements.txt .
 RUN uv pip install --system -r requirements.txt
 
-# Copy application files
-# Assumes these files are at the root of the repo
-COPY main.py .
-COPY agent.py .
-COPY database.py .
+# Copy the entire backend package (preserves module structure)
+COPY backend/ ./backend/
 
 # Expose port 7860 (required by HF Spaces)
 EXPOSE 7860
 
-# Run the application
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]
+# Run the application – note the module path includes the package name
+CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "7860"]

add_column.py

@@ -0,0 +1,35 @@
+import os
+from dotenv import load_dotenv
+from sqlalchemy import create_engine, text
+
+# Load environment variables
+load_dotenv(".env", override=True)
+load_dotenv("../.env", override=False)
+
+DATABASE_URL = os.getenv("DATABASE_URL")
+
+if not DATABASE_URL:
+    print("Error: DATABASE_URL not found")
+    exit(1)
+
+def add_column():
+    engine = create_engine(DATABASE_URL)
+    with engine.connect() as conn:
+        try:
+            # Check if column exists first to avoid error
+            check_sql = text("SELECT column_name FROM information_schema.columns WHERE table_name='conversations' AND column_name='summary';")
+            result = conn.execute(check_sql)
+            if result.fetchone():
+                print("Column 'summary' already exists.")
+                return
+
+            print("Adding 'summary' column to 'conversations' table...")
+            sql = text("ALTER TABLE conversations ADD COLUMN summary TEXT;")
+            conn.execute(sql)
+            conn.commit()
+            print("Successfully added 'summary' column.")
+        except Exception as e:
+            print(f"Error adding column: {e}")
+
+if __name__ == "__main__":
+    add_column()

agent.py

@@ -21,8 +21,6 @@ from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
 from youtube_transcript_api import YouTubeTranscriptApi
 import yt_dlp
 
-
-
 # --- Configuration ---
 llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
 
@@ -31,6 +29,20 @@ embeddings = GoogleGenerativeAIEmbeddings(model="models/text-embedding-004")
 vector_store = Chroma(embedding_function=embeddings, persist_directory="./chroma_db")
 retriever = vector_store.as_retriever(search_kwargs={"k": 3})
 
+def clear_vector_store():
+    """Clears the vector store."""
+    global vector_store, retriever
+    try:
+        # Delete the collection
+        vector_store.delete_collection()
+        # Re-initialize
+        vector_store = Chroma(embedding_function=embeddings, persist_directory="./chroma_db")
+        retriever = vector_store.as_retriever(search_kwargs={"k": 3})
+        return True
+    except Exception as e:
+        print(f"Error clearing vector store: {e}")
+        return False
+
 # --- State Definition ---
 class AgentState(TypedDict):
     """The state of our Deep Research Agent."""
@@ -45,6 +57,29 @@ class AgentState(TypedDict):
     youtube_url: str
     youtube_captions: str
     deep_research: bool  # Flag to indicate if deep research is requested
+    conversation_id: str # For RAG isolation
+
+# --- File Processing ---
+from langchain_community.document_loaders import PyPDFLoader, TextLoader
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+
+def upload_file(file_path: str, conversation_id: str):
+    """Process uploaded file and add to vector store with metadata."""
+    if file_path.endswith(".pdf"):
+        loader = PyPDFLoader(file_path)
+    else:
+        loader = TextLoader(file_path)
+        
+    docs = loader.load()
+    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
+    splits = text_splitter.split_documents(docs)
+    
+    # Add metadata
+    for split in splits:
+        split.metadata["conversation_id"] = conversation_id
+        
+    vector_store.add_documents(splits)
+    return splits
 
 # --- Data Models ---
 class Plan(BaseModel):
@@ -63,10 +98,14 @@ def extract_video_id(url):
 
 def get_video_duration(url):
     """Gets video duration in seconds using yt-dlp."""
-    ydl_opts = {'quiet': True}
-    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-        info = ydl.extract_info(url, download=False)
-        return info.get('duration', 0)
+    ydl_opts = {'quiet': True, 'no_warnings': True}
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+            return info.get('duration', 0)
+    except Exception as e:
+        print(f"Warning: Could not check duration: {e}")
+        return 0 # Return 0 to skip duration check on error
 
 # --- Nodes ---
 
@@ -118,29 +157,91 @@ def youtube_node(state: AgentState):
     # Check Duration
     try:
         duration = get_video_duration(url)
-        if duration > 600: # 10 minutes
-            return {"final_report": f"Error: Video is too long ({duration//60} mins). Limit is 10 minutes.", "steps": ["Video rejected: Too long"]}
+        if duration > 1200: # 20 minutes limit (increased)
+            return {"final_report": f"Error: Video is too long ({duration//60} mins). Limit is 20 minutes.", "steps": ["Video rejected: Too long"]}
     except Exception as e:
-        return {"final_report": f"Error checking duration: {e}", "steps": ["Failed to check duration"]}
+        print(f"Error checking duration: {e}")
+        # Continue anyway if duration check fails (might be network issue)
         
     # Get Captions
+    transcript_text = ""
     try:
-        print(f"DEBUG: Fetching captions for {video_id}")
-        yt = YouTubeTranscriptApi()
-        transcript_list = yt.list(video_id)
-        # Try to find English, or fallback to first available
+        print(f"Fetching captions for {video_id}")
+        # Method 1: Try YouTubeTranscriptApi
         try:
-            transcript = transcript_list.find_transcript(['en'])
-        except:
-            # Fallback to the first available transcript
-            transcript = next(iter(transcript_list))
+            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
+            transcript = None
+            try:
+                transcript = transcript_list.find_transcript(['en'])
+            except:
+                # Get any available
+                for t in transcript_list:
+                    transcript = t
+                    break
+            
+            if transcript:
+                transcript_data = transcript.fetch()
+                # Handle both dictionary and object formats
+                transcript_text = " ".join([
+                    entry.text if hasattr(entry, 'text') else entry['text'] 
+                    for entry in transcript_data
+                ])
+        except Exception as e:
+            print(f"YouTubeTranscriptApi failed: {e}. Trying yt-dlp fallback...")
             
-        transcript_data = transcript.fetch()
-        # Handle if it returns objects or dicts (it seems to be objects in this version)
-        transcript_text = " ".join([t.text for t in transcript_data])
+            # Method 2: Fallback to yt-dlp
+            import requests
+            ydl_opts = {
+                'skip_download': True,
+                'writesubtitles': True,
+                'writeautomaticsub': True,
+                'subtitleslangs': ['en'],
+                'quiet': True
+            }
+            
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(url, download=False)
+                subtitles = info.get('subtitles', {})
+                auto_captions = info.get('automatic_captions', {})
+                
+                sub_url = None
+                if 'en' in subtitles:
+                    for fmt in subtitles['en']:
+                        if fmt['ext'] == 'json3':
+                            sub_url = fmt['url']
+                            break
+                    if not sub_url and subtitles['en']:
+                        sub_url = subtitles['en'][0]['url']
+                elif 'en' in auto_captions:
+                    for fmt in auto_captions['en']:
+                        if fmt['ext'] == 'json3':
+                            sub_url = fmt['url']
+                            break
+                    if not sub_url and auto_captions['en']:
+                        sub_url = auto_captions['en'][0]['url']
+                
+                if sub_url:
+                    print(f"Fetching captions from: {sub_url}")
+                    r = requests.get(sub_url)
+                    data = r.json()
+                    events = data.get('events', [])
+                    for event in events:
+                        if 'segs' in event:
+                            for seg in event['segs']:
+                                if 'utf8' in seg:
+                                    transcript_text += seg['utf8']
+                            transcript_text += " "
+                else:
+                    raise Exception("No captions found via yt-dlp")
+
+        if not transcript_text:
+             return {"final_report": "No captions available for this video.", "steps": ["No captions found"]}
+             
     except Exception as e:
-        print(f"DEBUG: Caption Error: {e}")
-        return {"final_report": f"Error fetching captions: {e}", "steps": ["Failed to fetch captions"]}
+        print(f"Caption Error: {e}")
+        import traceback
+        traceback.print_exc()
+        return {"final_report": f"Error fetching captions: {e}. \n\nPossible reasons:\n1. Video has no captions.\n2. Network restrictions.\n3. Video is private.", "steps": ["Failed to fetch captions"]}
     
     # Generate Title
     system = """You are a YouTube Expert. Analyze the provided video transcript and generate 3 catchy, AI-enhanced title options.
@@ -160,7 +261,11 @@ def youtube_node(state: AgentState):
     )
     
     chain = prompt | llm | StrOutputParser()
-    raw_titles = chain.invoke({"transcript": transcript_text[:5000], "task": task})
+    try:
+        raw_titles = chain.invoke({"transcript": transcript_text[:10000], "task": task}) # Increased limit
+    except Exception as e:
+        raw_titles = "VIRAL: Error generating titles\nSEO: Error\nPROFESSIONAL: Error"
+        print(f"Title generation error: {e}")
     
     # Manually construct a beautifully formatted report with MAXIMUM SPACING
     report = "# YouTube Video Analysis\n\n\n"
@@ -180,11 +285,21 @@ def youtube_node(state: AgentState):
     if "caption" in task.lower() or "transcript" in task.lower():
         report += "---\n\n\n"
         report += "## 📝 Full Captions\n\n\n"
-        report += f"```text\n{transcript_text}\n```\n\n\n"
+        report += f"```text\n{transcript_text[:5000]}...\n```\n\n(Truncated for display)\n\n"
     else:
         report += "---\n\n\n"
         report += "> **Note:** Captions are available for this video! Add 'with captions' to your request to see them.\n\n\n"
         
+    # Add summary of video content
+    summary_system = "Summarize the following video transcript in 3-5 bullet points."
+    summary_prompt = ChatPromptTemplate.from_messages([("system", summary_system), ("human", "{transcript}")])
+    summary_chain = summary_prompt | llm | StrOutputParser()
+    try:
+        summary = summary_chain.invoke({"transcript": transcript_text[:10000]})
+        report += "## 📹 Video Summary\n\n" + summary + "\n\n"
+    except:
+        pass
+
     return {
         "final_report": report, 
         "youtube_captions": transcript_text,
@@ -203,7 +318,11 @@ def quick_response_node(state: AgentState):
     
     # Try to get relevant context from vector store
     try:
-        docs = retriever.invoke(task)
+        conversation_id = state.get("conversation_id")
+        filter_dict = {"conversation_id": conversation_id} if conversation_id else None
+        
+        # Use similarity_search directly to support filtering
+        docs = vector_store.similarity_search(task, k=3, filter=filter_dict)
         context = "\n\n".join([d.page_content for d in docs]) if docs else ""
     except Exception as e:
         print(f"Retriever error: {e}")
@@ -211,7 +330,7 @@ def quick_response_node(state: AgentState):
     
     # Check if we should do a quick web search (for real-time info)
     # If context is empty OR if the query implies real-time data
-    real_time_keywords = ["price", "current", "news", "latest", "today", "now", "live", "rate", "stock", "weather", "forecast", "score", "result", "vs", "when", "where", "who"]
+    real_time_keywords = ["price", "current", "news", "latest", "today", "now", "live", "rate", "stock", "weather", "forecast", "score", "result", "vs", "when", "where", "who", "what"]
     should_search = any(k in task.lower() for k in real_time_keywords)
     
     web_context = ""
@@ -246,6 +365,7 @@ def quick_response_node(state: AgentState):
     4. If the user asks for "price", "news", or "current" info, prioritize the Web Search Results.
     5. Keep responses focused. Do NOT write a long report.
     6. If the question requires extensive research, suggest the user ask for "deep research".
+    7. If you don't know the answer and have no context, use your general knowledge to answer as best as possible.
     """
     
     if full_context:
@@ -321,7 +441,10 @@ def research_node(state: AgentState):
         steps_log.append(f"Researching: {step}")
         
         # 1. Try Vector Store first
-        docs = retriever.invoke(step)
+        conversation_id = state.get("conversation_id")
+        filter_dict = {"conversation_id": conversation_id} if conversation_id else None
+        
+        docs = vector_store.similarity_search(step, k=3, filter=filter_dict)
         if docs:
             context = "\n".join([d.page_content for d in docs])
             content.append(f"Source: Local Documents\nTopic: {step}\nContent: {context}")
@@ -361,9 +484,9 @@ def writer_node(state: AgentState):
     2. **Headers**: Use headers (##, ###) to organize sections. **IMPORTANT**: Always add a blank line before and after every header.
     3. **Content**: Synthesize the information. Do not just list facts.
     4. **Formatting**:
-       - Use **bold** for key terms.
-       - Use bullet points for lists (ensure there is a blank line before the list starts).
-       - Use > Blockquotes for important summaries.
+    - Use **bold** for key terms.
+    - Use bullet points for lists (ensure there is a blank line before the list starts).
+    - Use > Blockquotes for important summaries.
     5. **Citations**: If the notes mention specific sources, cite them.
     6. **Conclusion**: End with a strong conclusion.
     

api_test.py

@@ -0,0 +1,57 @@
+import requests
+import os
+
+BASE_URL = "http://127.0.0.1:7860"
+
+def test_api():
+    print("--- Starting API Test ---")
+    
+    # 1. Create a dummy file
+    test_file = "api_test_doc.txt"
+    with open(test_file, "w") as f:
+        f.write("The API secret is 99999. Do not share this.")
+        
+    conversation_id = "api_chat_1"
+    
+    try:
+        # 2. Upload File
+        print(f"\n1. Uploading to {conversation_id}...")
+        with open(test_file, "rb") as f:
+            files = {"file": f}
+            data = {"conversation_id": conversation_id}
+            response = requests.post(f"{BASE_URL}/api/upload", files=files, data=data)
+            
+        print(f"Upload Status: {response.status_code}")
+        print(f"Upload Response: {response.json()}")
+        
+        if response.status_code != 200:
+            print("❌ Upload Failed")
+            return
+
+        # 3. Chat (Ask about the file)
+        print(f"\n2. Asking about the file in {conversation_id}...")
+        chat_data = {
+            "message": "What is the API secret?",
+            "history": [],
+            "conversation_id": conversation_id,
+            "user_id": "test_user"
+        }
+        
+        response = requests.post(f"{BASE_URL}/api/chat", json=chat_data)
+        print(f"Chat Status: {response.status_code}")
+        result = response.json()
+        print(f"Chat Response: {result.get('response')}")
+        
+        if "99999" in result.get('response', ''):
+            print("✅ Success: AI found the secret!")
+        else:
+            print("❌ Failure: AI did not find the secret.")
+            
+    except Exception as e:
+        print(f"Test Failed: {e}")
+    finally:
+        if os.path.exists(test_file):
+            os.remove(test_file)
+
+if __name__ == "__main__":
+    test_api()

check_yt.py

@@ -0,0 +1,65 @@
+
+import yt_dlp
+import requests
+import json
+
+video_id = "2dGB9Fo4hnU"
+url = f"https://www.youtube.com/watch?v={video_id}"
+
+print("\n--- Method 2: yt-dlp ---")
+try:
+    ydl_opts = {
+        'skip_download': True,
+        'writesubtitles': True,
+        'writeautomaticsub': True,
+        'subtitleslangs': ['en'],
+        'quiet': True
+    }
+    
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        info = ydl.extract_info(url, download=False)
+        
+        subtitles = info.get('subtitles', {})
+        auto_captions = info.get('automatic_captions', {})
+        
+        sub_url = None
+        if 'en' in subtitles:
+            print("Found manual English subtitles")
+            # Prefer json3
+            for fmt in subtitles['en']:
+                if fmt['ext'] == 'json3':
+                    sub_url = fmt['url']
+                    break
+            if not sub_url:
+                sub_url = subtitles['en'][0]['url']
+        elif 'en' in auto_captions:
+            print("Found auto English captions")
+            for fmt in auto_captions['en']:
+                if fmt['ext'] == 'json3':
+                    sub_url = fmt['url']
+                    break
+            if not sub_url:
+                sub_url = auto_captions['en'][0]['url']
+        
+        if sub_url:
+            print(f"Fetching: {sub_url}")
+            r = requests.get(sub_url)
+            data = r.json()
+            # print(json.dumps(data, indent=2)[:500])
+            
+            # Parse json3
+            events = data.get('events', [])
+            text = ""
+            for event in events:
+                if 'segs' in event:
+                    for seg in event['segs']:
+                        if 'utf8' in seg:
+                            text += seg['utf8']
+                    text += " "
+            print(f"Extracted text length: {len(text)}")
+            print(f"Preview: {text[:100]}")
+        else:
+            print("No English subtitles found")
+
+except Exception as e:
+    print(f"Method 2 failed: {e}")

database.py

@@ -28,6 +28,7 @@ class Conversation(Base):
     title = Column(String, nullable=False)
     created_at = Column(DateTime, default=datetime.utcnow)
     updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    summary = Column(Text, nullable=True)  # Short summary of the conversation
     
     # Relationship
     messages = relationship("Message", back_populates="conversation", cascade="all, delete-orphan")

main.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, HTTPException, UploadFile, File, Depends
+from fastapi import FastAPI, HTTPException, UploadFile, File, Depends, BackgroundTasks, Form
 from fastapi.concurrency import run_in_threadpool
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
@@ -58,27 +58,23 @@ class ConversationResponse(BaseModel):
     created_at: str
     updated_at: str
     message_count: int = 0
+    summary: Optional[str] = None
 
 @app.post("/api/upload")
-async def upload_file(file: UploadFile = File(...)):
+async def upload_file(file: UploadFile = File(...), conversation_id: str = Form(...)):
+    print(f"DEBUG: Uploading file {file.filename} to conversation {conversation_id}")
+    if not conversation_id or conversation_id == "null" or conversation_id == "undefined":
+        print("ERROR: Invalid conversation_id received in upload_file")
+        raise HTTPException(status_code=400, detail="Please start a conversation first!")
     try:
         # Save file temporarily
         file_path = f"temp_{file.filename}"
         with open(file_path, "wb") as buffer:
             shutil.copyfileobj(file.file, buffer)
         
-        # Load and split document
-        if file.filename.endswith(".pdf"):
-            loader = PyPDFLoader(file_path)
-        else:
-            loader = TextLoader(file_path)
-            
-        docs = loader.load()
-        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
-        splits = text_splitter.split_documents(docs)
-        
-        # Add to Vector Store
-        vector_store.add_documents(splits)
+        # Process file
+        from agent import upload_file as agent_upload
+        splits = agent_upload(file_path, conversation_id)
         
         # Cleanup
         os.remove(file_path)
@@ -87,8 +83,61 @@ async def upload_file(file: UploadFile = File(...)):
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
+@app.delete("/api/vector_store")
+async def clear_vector_store_endpoint():
+    try:
+        from agent import clear_vector_store
+        success = clear_vector_store()
+        if success:
+            return {"status": "success", "message": "Vector store cleared"}
+        else:
+            raise HTTPException(status_code=500, detail="Failed to clear vector store")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+async def generate_conversation_summary(conversation_id: str, db: Session):
+    """Background task to generate a summary for a conversation."""
+    try:
+        # Get messages
+        messages = db.query(DBMessage).filter(
+            DBMessage.conversation_id == conversation_id
+        ).order_by(DBMessage.created_at).limit(10).all() # Limit to first 10 for summary
+        
+        if not messages:
+            return
+
+        conversation_text = "\n".join([f"{msg.role}: {msg.content}" for msg in messages])
+        
+        from langchain_google_genai import ChatGoogleGenerativeAI
+        from langchain_core.prompts import ChatPromptTemplate
+        from langchain_core.output_parsers import StrOutputParser
+        
+        llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
+        
+        system = """You are a helpful assistant. Create a very short, 1-sentence summary (max 10 words) of this conversation topic.
+        Example: "Python script debugging", "Recipe for chocolate cake", "Travel plans to Japan".
+        """
+        
+        prompt = ChatPromptTemplate.from_messages([
+            ("system", system),
+            ("human", "Conversation:\n{text}")
+        ])
+        
+        chain = prompt | llm | StrOutputParser()
+        summary = chain.invoke({"text": conversation_text})
+        
+        # Update conversation
+        conversation = db.query(Conversation).filter(Conversation.id == conversation_id).first()
+        if conversation:
+            conversation.summary = summary.strip()
+            db.commit()
+            print(f"Generated summary for {conversation_id}: {summary}")
+            
+    except Exception as e:
+        print(f"Error generating summary: {e}")
+
 @app.post("/api/chat")
-def chat_endpoint(request: ChatRequest, db: Session = Depends(get_db)):
+def chat_endpoint(request: ChatRequest, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
     try:
         # Convert history to LangChain messages
         messages = []
@@ -108,13 +157,13 @@ def chat_endpoint(request: ChatRequest, db: Session = Depends(get_db)):
             "plan": [],
             "content": [],
             "revision_number": 0,
-            "max_revisions": 1,
-            "final_report": "",
+            "max_revisions": 2,
             "steps": [],
-            "messages": messages,
-            "deep_research": False,  # Will be set by router
+            "messages": [HumanMessage(content=request.message)],
             "youtube_url": "",
-            "youtube_captions": ""
+            "youtube_captions": "",
+            "deep_research": False, # Will be set by router
+            "conversation_id": request.conversation_id
         }
         
         result = agent_app.invoke(inputs)
@@ -163,6 +212,14 @@ def chat_endpoint(request: ChatRequest, db: Session = Depends(get_db)):
                     conversation.updated_at = datetime.utcnow()
                     
                     db.commit()
+                    
+                    # Trigger summary generation if it's the first few messages or summary is missing
+                    # We can check message count or just do it periodically
+                    # For simplicity, let's do it if message count is small (< 5) or summary is None
+                    message_count = db.query(DBMessage).filter(DBMessage.conversation_id == request.conversation_id).count()
+                    if message_count <= 4 or not conversation.summary:
+                         background_tasks.add_task(generate_conversation_summary, request.conversation_id, db)
+
             except Exception as db_error:
                 print(f"Database error: {db_error}")
                 db.rollback()
@@ -233,7 +290,8 @@ async def create_conversation(conv: ConversationCreate, db: Session = Depends(ge
             "title": new_conv.title,
             "created_at": new_conv.created_at.isoformat(),
             "updated_at": new_conv.updated_at.isoformat(),
-            "message_count": 0
+            "message_count": 0,
+            "summary": None
         }
     except Exception as e:
         db.rollback()
@@ -262,7 +320,8 @@ async def get_conversations(user_id: str, db: Session = Depends(get_db)):
                 "title": conv.title,
                 "created_at": conv.created_at.isoformat(),
                 "updated_at": conv.updated_at.isoformat(),
-                "message_count": message_count
+                "message_count": message_count,
+                "summary": conv.summary
             })
         
         return result

rag_test.py

@@ -0,0 +1,50 @@
+import os
+import shutil
+from dotenv import load_dotenv
+from agent import upload_file, vector_store
+
+# Load environment variables
+load_dotenv(".env", override=True)
+load_dotenv("../.env", override=False)
+
+def test_rag():
+    print("--- Starting RAG Test ---")
+    
+    # Create a dummy test file
+    test_file = "rag_test_doc.txt"
+    with open(test_file, "w") as f:
+        f.write("The secret code is 12345. This is a confidential document for Project X.")
+    
+    conversation_id_1 = "chat_1"
+    conversation_id_2 = "chat_2"
+    
+    try:
+        # 1. Upload to Chat 1
+        print(f"\n1. Uploading to {conversation_id_1}...")
+        upload_file(test_file, conversation_id_1)
+        
+        # 2. Retrieve from Chat 1 (Should find it)
+        print(f"\n2. Searching in {conversation_id_1}...")
+        results_1 = vector_store.similarity_search("secret code", k=1, filter={"conversation_id": conversation_id_1})
+        if results_1:
+            print(f"✅ Found: {results_1[0].page_content}")
+        else:
+            print("❌ Not found (Unexpected)")
+            
+        # 3. Retrieve from Chat 2 (Should NOT find it)
+        print(f"\n3. Searching in {conversation_id_2}...")
+        results_2 = vector_store.similarity_search("secret code", k=1, filter={"conversation_id": conversation_id_2})
+        if not results_2:
+            print("✅ Not found (Expected)")
+        else:
+            print(f"❌ Found (Unexpected): {results_2[0].page_content}")
+            
+    except Exception as e:
+        print(f"Test Failed: {e}")
+    finally:
+        # Cleanup
+        if os.path.exists(test_file):
+            os.remove(test_file)
+
+if __name__ == "__main__":
+    test_rag()