add new gradio + agent implementation

README.md

@@ -10,4 +10,72 @@ pinned: false
 license: mit
 ---
 
-An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).
+An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).
+
+# YouTwo Agent - Modal Deployment Guide
+
+This repository contains the YouTwo Agent, a conversational AI agent powered by FastRTC for real-time audio communication.
+
+## Prerequisites
+
+1. Install the Modal CLI:
+```bash
+pip install modal
+```
+
+2. Authenticate with Modal:
+```bash
+modal token new
+```
+
+3. Create the following secrets and volumes on Modal:
+   - Secret: `youtwo-secrets` with the following values:
+     - `NEBIUS_API_KEY` - Your Nebius API key for the LLM
+     - `VECTARA_API_KEY` - Your Vectara API key for RAG
+   - Volume: `youtwo-volume` - For storing any persistent data
+
+## Project Structure
+
+- `modal_app.py`: The main Modal deployment file
+- `agent.py`: Core agent logic
+- `rag.py`: Retrieval-augmented generation module
+- `schemas.py`: Data schemas for the app
+
+## Deployment Instructions
+
+### Local Development
+
+To run the app locally during development:
+
+```bash
+modal serve modal_app.py
+```
+
+This will provide you with a URL to access your app during development.
+
+### Production Deployment
+
+To deploy the app to Modal for production:
+
+```bash
+modal deploy modal_app.py
+```
+
+After deployment, you'll receive a URL where your application is accessible.
+
+## Environment Variables
+
+The application requires the following environment variables:
+
+- `NEBIUS_API_KEY`: API key for accessing the Nebius LLM
+- `VECTARA_API_KEY`: API key for Vectara RAG system
+
+These should be set up as secrets in Modal.
+
+## Troubleshooting
+
+- **Issue:** "WebRTC peer connection failed"
+  - **Solution:** Ensure your browser has permission to access your microphone and that you're using HTTPS or localhost.
+
+- **Issue:** "Modal volume not accessible"
+  - **Solution:** Verify that you've created the `youtwo-volume` volume in Modal.

app.py

@@ -1,165 +1,8 @@
-import gradio as gr
-from pathlib import Path
-from rag import is_allowed_filetype, upload_file_to_vectara, retrieve_chunks
-import logging
+# Run this gradio app with:
+# python run_gradio_app.py
 
-# ---------------------------
-# Placeholder Backend Functions
-# ---------------------------
+from src.yt_gradio.app import get_gradio_blocks
 
-def sync_lifelog_db() -> str:
-    """
-    Synchronizes local copy of lifelog (journal) database with the latest source.
-    This function acts as a placeholder for database connection logic.
-
-    Returns:
-        str: Status message with sync timestamp.
-    """
-    import datetime
-    return f"✅ Lifelog database synchronized successfully at {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
-
-
-def search_lifelogs(keyword: str, date_start: str, date_end: str) -> str:
-    """
-    Searches for lifelog entries containing the provided keyword within the specified date range.
-
-    Args:
-        keyword (str): Search term to match against entries.
-        date_start (str): Start date (YYYY-MM-DD format).
-        date_end (str): End date (inclusive, YYYY-MM-DD format).
-
-    Returns:
-        str: Simulated search result summary. Replace with actual search logic.
-    """
-    return f"🔍 Found 12 entries related to ‘{keyword}’ between {date_start} and {date_end}."
-
-
-def update_knowledge_graph_relations() -> str:
-    """
-    Updates the knowledge graph using the latest lifelog data.
-    Simulates the creation of new triples (subject-predicate-object) from lifelog content.
-
-    Returns:
-        str: Update summary with number of relations added.
-    """
-    return "🧠 15 new triples have been added to the knowledge graph."
-
-
-def natural_language_handler(query: str) -> str:
-    """
-    Processes natural language inputs to determine which system function to execute.
-    Designed to interface with NLP/LLM for future automation.
-
-    Args:
-        query (str): Free-text input from the user.
-
-    Returns:
-        str: Simulated or generated action and result.
-    """
-    chunks, response = retrieve_chunks(query, limit=5)
-    return f"💬 Got {len(chunks)} chunks for your request: “{query}”. Response: {response}"
-
-
-def placeholder(feature_name: str = "unknown") -> str:
-    """
-    Placeholder for unimplemented features.
-
-    Args:
-        feature_name (str): Name of the feature to query.
-
-    Returns:
-        str: Placeholder response for future functions.
-    """
-    return f"{feature_name} functionality not available yet."  # Replace with dynamic logic later
-
-
-# Gradio Behavior:
-# Textbox: As input component: Passes text value as a str into the function.
-# File: As input component: Passes the filepath to a temporary file object whose full path can be retrieved by file_obj.name
-# If we change the type to "binary", uploaded_file returns bytes.
-# NOTE: This means we can handle multiple files by tweaking this expected type.
-def handle_file_input(file_path: str | None, uploaded_file: gr.File | None):
-    if not uploaded_file and not file_path:
-        return "Please enter a file path or upload a file."
-    
-    if uploaded_file:
-        filepath = Path(uploaded_file.name)
-    else:
-        filepath = Path(file_path.strip())
-        
-    if not filepath.exists():
-        logging.error(f"Error: The specified file path does not exist: {filepath}")
-        return "Error: The uploaded filepath does not exist."
-    
-    if not is_allowed_filetype(filepath.suffix):
-        return f"Error: The uploaded filetype {filepath.suffix} is not supported."
-    
-    # Obtain the bytes
-    with open(filepath, "rb") as file:
-        file_contents = file.read()
-        
-    
-    upload_result = upload_file_to_vectara(file_contents, filepath.name)
-    
-    return f"Uploaded document: {upload_result['id']}"
-
-# ---------------------------
-# Gradio UI (Blocks API)
-# ---------------------------
-
-with gr.Blocks(title="Knowledge Graph Agent Interface") as demo:
-    gr.Markdown("## 🧠 Knowledge Graph Agent Interface\nBuilt with Gradio + MCP Support for LLM Tool Integration")
-
-    with gr.Tab("🔄 Sync Lifelog DB"):
-        gr.Markdown("Synchronize the lifelog database locally.")
-        sync_btn = gr.Button("Sync Database")
-        sync_out = gr.Textbox(lines=2, label="Sync Status")
-        sync_btn.click(fn=sync_lifelog_db, outputs=sync_out)
-
-    with gr.Tab("🔍 Search Lifelogs"):
-        gr.Markdown("Search lifelog entries by keyword and time range.")
-        keyword = gr.Textbox(label="Search Keyword")
-        with gr.Row():
-            start_date = gr.Textbox(label="Start Date (YYYY-MM-DD)")
-            end_date = gr.Textbox(label="End Date (YYYY-MM-DD)")
-        search_btn = gr.Button("Search Entries")
-        search_out = gr.Textbox(label="Search Results")
-        search_btn.click(fn=search_lifelogs, inputs=[keyword, start_date, end_date], outputs=search_out)
-
-    with gr.Tab("🧠 Update Knowledge Graph"):
-        gr.Markdown("Use lifelog data to update knowledge graph relations.")
-        update_btn = gr.Button("Update Graph")
-        update_out = gr.Textbox(label="Update Status")
-        update_btn.click(fn=update_knowledge_graph_relations, outputs=update_out)
-
-    with gr.Tab("🗣️ Natural Language Mode"):
-        gr.Markdown("Input natural language requests for system actions.")
-        with gr.Row():
-            user_query = gr.Textbox(label="Type your query")
-        query_btn = gr.Button("Process Request")
-        query_out = gr.Textbox(label="System Response")
-        query_btn.click(fn=natural_language_handler, inputs=user_query, outputs=query_out)
-
-    with gr.Tab("⚙️ Future Features"):
-        gr.Markdown("Placeholder area for upcoming functionalities")
-        feature = gr.Textbox(label="Feature to Check")
-        feature_btn = gr.Button("Check Feature Status")
-        feature_out = gr.Textbox(label="Status")
-        feature_btn.click(fn=placeholder, inputs=feature, outputs=feature_out)
-
-    with gr.Tab("🗣️Input File"):
-        gr.Markdown("Input file")
-        with gr.Row():
-            file_path_input = gr.Textbox(label="Enter File Path")
-            file_upload_input = gr.File(label="Upload a File", type="filepath")
-        submit_btn = gr.Button("Submit")
-        output = gr.Textbox(label="Result")
-        submit_btn.click(fn=handle_file_input, inputs=[file_path_input, file_upload_input], outputs=output)
-
-# ---------------------------
-# Launch as MCP Server
-# ---------------------------
 if __name__ == "__main__":
-    from dotenv import load_dotenv
-    load_dotenv()
-    demo.launch()
+    blocks = get_gradio_blocks()
+    blocks.launch()

chatbot-app.py

@@ -1,64 +0,0 @@
-# import gradio as gr
-# from huggingface_hub import InferenceClient
-
-# """
-# For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-# """
-# client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
-
-
-# def respond(
-#     message,
-#     history: list[tuple[str, str]],
-#     system_message,
-#     max_tokens,
-#     temperature,
-#     top_p,
-# ):
-#     messages = [{"role": "system", "content": system_message}]
-
-#     for val in history:
-#         if val[0]:
-#             messages.append({"role": "user", "content": val[0]})
-#         if val[1]:
-#             messages.append({"role": "assistant", "content": val[1]})
-
-#     messages.append({"role": "user", "content": message})
-
-#     response = ""
-
-#     for message in client.chat_completion(
-#         messages,
-#         max_tokens=max_tokens,
-#         stream=True,
-#         temperature=temperature,
-#         top_p=top_p,
-#     ):
-#         token = message.choices[0].delta.content
-
-#         response += token
-#         yield response
-
-
-# """
-# For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-# """
-# demo = gr.ChatInterface(
-#     respond,
-#     additional_inputs=[
-#         gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-#         gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-#         gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-#         gr.Slider(
-#             minimum=0.1,
-#             maximum=1.0,
-#             value=0.95,
-#             step=0.05,
-#             label="Top-p (nucleus sampling)",
-#         ),
-#     ],
-# )
-
-
-# if __name__ == "__main__":
-#     demo.launch()

requirements.txt

@@ -1,5 +1,14 @@
-gradio[mcp]==5.33.0
+gradio==5.33.0
+modal==1.0.2
+smolagents==1.17.0
 requests
 python-dotenv
-smolagents
-fastrtc
+
+# For knowledge graph
+# langgraph
+# langchain_core
+# matplotlib
+# networkx
+
+# For vectara (Useful to get types)
+# vectara

schemas.py

@@ -1,11 +0,0 @@
-from typing import TypedDict, Dict, Any
-
-class StorageUsage(TypedDict):
-    bytes_used: int
-    metadata_bytes_used: int
-
-
-class UploadResult(TypedDict):
-    id: str
-    metadata: Dict[str, Any]
-    storage_usage: StorageUsage

yt_agent/__init__.py

@@ -0,0 +1,3 @@
+"""
+Agent module for YouTwo
+""" 

yt_agent/agent.py

@@ -0,0 +1,26 @@
+import os
+
+from dotenv import load_dotenv
+from smolagents import CodeAgent, InferenceClientModel
+
+from src.yt_agent.tools import retrieve_tool, inspect_database_tool
+
+# Load environment variables
+load_dotenv()
+# Initialize models
+model = InferenceClientModel(provider="nebius", model="nebius/Qwen/Qwen3-30B-A3B", api_key=os.environ["NEBIUS_API_KEY"])
+agent = CodeAgent(
+    tools=[
+        retrieve_tool,
+        inspect_database_tool,
+    ],
+    model=model,
+    max_steps=2,
+    verbosity_level=2,
+    description="Agent used to search documents.",
+)
+
+if __name__ == "__main__":
+    agent.run("What is 2+2?")
+    messages = agent.memory.get_full_steps()
+    print(messages)

yt_agent/prompts.py

@@ -0,0 +1,8 @@
+AGENTIC_MODE_SYSTEM_PROMPT = """You are a general-purpose chatbot that answers user questions using information from uploaded documents. 
+When a question requires document-based information:
+1. Use the retrieval tool to fetch relevant document chunks
+2. Synthesize answers using these chunks
+3. Always cite your document sources
+
+If information isn't in documents, use your general knowledge but state this limitation.
+"""

yt_agent/tools.py

@@ -0,0 +1,37 @@
+from smolagents import tool
+from src.schemas import VectaraDocuments
+from src.yt_rag.rag import fetch_documents_from_corpus, retrieve_chunks
+
+@tool
+def retrieve_tool(query: str, limit: int = 5, filter_by_id: str = None) -> dict[str, list[str] | str]:
+    """
+    Retrieve chunks by relevance to a query
+
+    Args:
+        query: The query to retrieve chunks for
+        limit: The maximum number of chunks to retrieve (default: 5)
+        filter_by_id: A document ID to filter by
+    Returns:
+        A list of chunks, and a grounded summary
+    """
+    chunks, vectara_summary = retrieve_chunks(query, limit, filter_by_id)
+    return {
+        "chunks": chunks,
+        "summary": vectara_summary
+    }
+
+@tool
+def inspect_database_tool() -> str:
+    """
+    Inspect the vector database
+
+    Returns:
+        A list of documents
+    """
+    results = fetch_documents_from_corpus(limit = 50)
+    documents = VectaraDocuments(documents = results["documents"])
+    id_list = [document["id"] for document in documents["documents"]]
+    final_string = "The following documents IDs are in the vector database:\n"
+    for i, id in enumerate(id_list):
+        final_string += f"{i+1}. {id}\n"
+    return final_string

yt_gradio/__init__.py

@@ -0,0 +1,3 @@
+"""
+Gradio UI module for YouTwo
+""" 

yt_gradio/app.py

@@ -0,0 +1,145 @@
+from pprint import pprint
+import gradio as gr
+from pathlib import Path
+from src.yt_rag.rag import is_allowed_filetype, upload_file_to_vectara, retrieve_chunks
+import logging
+from src.yt_agent.agent import agent
+# ---------------------------
+# Placeholder Backend Functions
+# ---------------------------
+
+
+def update_knowledge_graph_relations() -> str:
+    """
+    Updates the knowledge graph using the latest lifelog data.
+    Simulates the creation of new triples (subject-predicate-object) from lifelog content.
+
+    Returns:
+        str: Update summary with number of relations added.
+    """
+    return "🧠 15 new triples have been added to the knowledge graph."
+
+
+def natural_language_handler(query: str) -> str:
+    """
+    Processes natural language inputs to determine which system function to execute.
+    Designed to interface with NLP/LLM for future automation.
+
+    Args:
+        query (str): Free-text input from the user.
+
+    Returns:
+        str: Simulated or generated action and result.
+    """
+    chunks, response = retrieve_chunks(query, limit=5)
+    return f"💬 Got {len(chunks)} chunks for your request: “{query}”. Response: {response}"
+
+def agent_chat(message: str, chat_history):
+    if not message.strip():
+        return chat_history, ""
+
+    # Append user message to history
+    chat_history.append({"role": "user", "content": message})
+
+    # Run your agent
+    response = agent.run(message)
+    if isinstance(response, dict):
+        parsed_response = response.get("output") or response.get("answer") or str(response)
+    else:
+        parsed_response = str(response)
+
+    # Append agent response to history
+    chat_history.append({"role": "assistant", "content": parsed_response})
+
+    return chat_history, ""
+
+# Gradio Behavior:
+# Textbox: As input component: Passes text value as a str into the function.
+# File: As input component: Passes the filepath to a temporary file object whose full path can be retrieved by file_obj.name
+# If we change the type to "binary", uploaded_file returns bytes.
+# NOTE: This means we can handle multiple files by tweaking this expected type.
+def handle_file_input(file_path: str | None, uploaded_file: gr.File | None):
+    if not uploaded_file and not file_path:
+        return "Please enter a file path or upload a file."
+    
+    if uploaded_file:
+        filepath = Path(uploaded_file.name)
+    else:
+        filepath = Path(file_path.strip())
+        
+    if not filepath.exists():
+        logging.error(f"Error: The specified file path does not exist: {filepath}")
+        return "Error: The uploaded filepath does not exist."
+    
+    if not is_allowed_filetype(filepath.suffix):
+        return f"Error: The uploaded filetype {filepath.suffix} is not supported."
+    
+    # Obtain the bytes
+    with open(filepath, "rb") as file:
+        file_contents = file.read()
+        
+    
+    upload_result = upload_file_to_vectara(file_contents, filepath.name)
+    
+    return f"Uploaded document: {upload_result['id']}"
+
+# ---------------------------
+# Gradio UI (Blocks API)
+# ---------------------------
+
+def get_gradio_blocks():
+    with gr.Blocks(title="Knowledge Graph Agent Interface") as demo:
+        gr.Markdown("## 🧠 Knowledge Graph Agent Interface\nBuilt with Gradio + MCP Support for LLM Tool Integration")
+
+        with gr.Tab("🗣️ Natural Language Mode"):
+            gr.Markdown("Input natural language requests for system actions.")
+            with gr.Row():
+                user_query = gr.Textbox(label="Type your query")
+            query_btn = gr.Button("Process Request")
+            query_out = gr.Textbox(label="System Response")
+            query_btn.click(fn=natural_language_handler, inputs=user_query, outputs=query_out)
+
+        with gr.Tab("⚙️ Agentic Chat"):
+            gr.Markdown("Agentic Question and Answer1")
+            chatbot = gr.Chatbot(label="KG Agent", height=500, show_label=True, container=True, type="messages",
+                bubble_full_width=False,
+                value=[
+                    {"role": "assistant", "content": "👋 Hello! I'm the KG Agent, your intelligent assistant for serving KG. How can I help you today?"}
+                ])
+            user_input = gr.Textbox(placeholder="Type your question...", label="Message", lines=2, scale=4, show_label=False, value="Inspect the vector database. Tell me how many documents are in the database.")
+            #clear_button = gr.Button("🗑️ Clear Chat", size="sm")
+            send_btn = gr.Button("Send", variant="primary", scale=1)
+
+            # Wire up the button (and hitting Enter) to call `agent_chat`
+            send_btn.click(
+                fn=agent_chat,
+                inputs=[user_input, chatbot],
+                outputs=[chatbot, user_input],
+                show_progress=True
+            )
+            user_input.submit(
+                fn=agent_chat,
+                inputs=[user_input, chatbot],
+                outputs=[chatbot, user_input],
+                #outputs=[self.chatbot, self.message_input, self.context_display, self.suggestions_display],
+            )
+
+        with gr.Tab("🗣️Input File"):
+            gr.Markdown("Input file")
+            with gr.Row():
+                file_path_input = gr.Textbox(label="Enter File Path")
+                file_upload_input = gr.File(label="Upload a File", type="filepath")
+            submit_btn = gr.Button("Submit")
+            output = gr.Textbox(label="Result")
+            submit_btn.click(fn=handle_file_input, inputs=[file_path_input, file_upload_input], outputs=output)
+
+    return demo
+
+# ---------------------------
+# Launch as MCP Server
+# ---------------------------
+if __name__ == "__main__":
+    from dotenv import load_dotenv
+    load_dotenv()
+    demo = get_gradio_blocks()
+    demo.launch(mcp_server=True)

yt_rag/__init__.py

@@ -0,0 +1,5 @@
+"""
+RAG (Retrieval Augmented Generation) module for YouTwo
+"""
+
+# from .rag import retrieve_chunks, is_allowed_filetype, upload_file_to_vectara

rag.py → yt_rag/rag.py

@@ -4,7 +4,8 @@ import os
 from pathlib import Path
 import requests
 from pprint import pprint
-from schemas import UploadResult
+from src.schemas import UploadResult
+
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -34,6 +35,7 @@ def load_environment_variables():
         raise IndexingError("Vectara API key not set. Please set the VECTARA_API_KEY environment variable.")
 
 
+
 def is_allowed_filetype(suffix: str):
     # Commonmark / Markdown (md extension).
     # PDF/A (pdf).
@@ -45,7 +47,7 @@ def is_allowed_filetype(suffix: str):
     # LXML files (.lxml).
     # RTF files (.rtf).
     # ePUB files (.epub).
-    return suffix in [".pdf", ".odt", ".doc", ".docx", ".ppt", ".pptx", ".txt", ".html", ".lxml", ".rtf", ".epub"]
+    return suffix in [".md", ".pdf", ".odt", ".doc", ".docx", ".ppt", ".pptx", ".txt", ".html", ".lxml", ".rtf", ".epub"]
 
 def save_response_to_file(response_json: dict, filename: str):
     """
@@ -135,7 +137,7 @@ def process_upload_response(response_json: dict) -> UploadResult:
         storage_usage=response_json["storage_usage"]
     )
 # See https://docs.vectara.com/docs/rest-api/query-corpus
-def retrieve_chunks(query: str, limit: int = 10) -> tuple[list[str], str]:
+def retrieve_chunks(query: str, limit: int = 10, filter_by_id: str = None) -> tuple[list[str], str]:
     """
     Retrieves relevant chunks and a generated summary from the Vectara corpus based on the query.
 
@@ -156,18 +158,19 @@ def retrieve_chunks(query: str, limit: int = 10) -> tuple[list[str], str]:
         "x-api-key": api_key,
         "Content-Type": "application/json"
     }
+    metadata_filter = f"doc.id='{filter_by_id}'" if filter_by_id else None
+    if metadata_filter:
+        search = {
+            "metadata_filter": metadata_filter,
+            "limit": limit,
+        }
+    else:
+        search = {
+            "limit": limit,
+        }
     payload = {
         "query": query,
-        "search": {
-            "limit": limit,  # Number of search results to retrieve
-            # "reranker": {
-            #     "type": "customer_reranker",
-            #     "reranker_name": "Rerank_Multilingual_v1",
-            #     "limit": 0,
-            #     "cutoff": 0,
-            #     "include_context": True
-            # }
-        },
+        "search": search,
         "generation": {
             "generation_preset_name": "mockingbird-2.0", # Using Mockingbird for RAG
             "max_used_search_results": 5,
@@ -212,6 +215,128 @@ def retrieve_chunks(query: str, limit: int = 10) -> tuple[list[str], str]:
     except Exception as e:
         raise VectaraAPIError(f"An unexpected error occurred during Vectara query: {e}") from e
 
+def fetch_documents_from_corpus(limit: int = 10, metadata_filter: str = None, page_key: str = None) -> dict:
+    """
+    Fetches documents from a specific Vectara corpus.
+    
+    Args:
+        limit (int, optional): Maximum number of documents to return. Must be between 1 and 100. Defaults to 10.
+        metadata_filter (str, optional): Filter documents by metadata. Uses expression similar to query metadata filter.
+        page_key (str, optional): Key used to retrieve the next page of documents after the limit has been reached.
+        request_timeout (int, optional): Time in seconds the API will attempt to complete the request before timing out.
+        request_timeout_millis (int, optional): Time in milliseconds the API will attempt to complete the request.
+    
+    Returns:
+        dict: The response from the Vectara API containing the requested documents.
+        
+    Raises:
+        VectaraAPIError: If there's an error with the Vectara API request.
+    """
+    import os
+    import requests
+    CORPUS_KEY = "YouTwo"
+    request_timeout = 20
+    request_timeout_millis = 60000
+
+
+    # Validate inputs
+    if limit is not None and (limit < 1 or limit > 100):
+        raise ValueError("Limit must be between 1 and 100")
+    
+    if len(CORPUS_KEY) > 50 or not all(c.isalnum() or c in ['_', '=', '-'] for c in CORPUS_KEY):
+        raise ValueError("corpus_key must be <= 50 characters and match regex [a-zA-Z0-9_\\=\\-]+$")
+    
+    # Prepare request
+    vectara_api_key = os.getenv("VECTARA_API_KEY")
+    
+    if not vectara_api_key:
+        raise VectaraAPIError("Vectara API key not found in environment variables")
+    
+    url = f"https://api.vectara.io/v2/corpora/{CORPUS_KEY}/documents"
+    
+    headers = {
+        "Accept": "application/json",
+        "x-api-key": vectara_api_key
+    }
+    
+    payload = {}
+    
+    # Build query params
+    params = {}
+    if limit is not None:
+        params["limit"] = limit
+    if metadata_filter is not None:
+        params["metadata_filter"] = metadata_filter
+    if page_key is not None:
+        params["page_key"] = page_key
+    try:
+        response = requests.get(url, headers=headers, params=params)
+        response.raise_for_status()
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        raise VectaraAPIError(f"Error fetching documents from Vectara corpus: {e}") from e
+    except Exception as e:
+        raise VectaraAPIError(f"An unexpected error occurred while fetching documents: {e}") from e
+
+def fetch_document_by_id(document_id: str) -> dict:
+    """
+    Retrieves the content and metadata of a specific document by its ID.
+    
+    Args:
+        document_id (str): The document ID to retrieve. Must be percent encoded.
+        
+    Returns:
+        dict: The document data including content and metadata.
+        
+    Raises:
+        VectaraAPIError: If there's an error with the Vectara API request.
+    """
+    import os
+    import requests
+    from urllib.parse import quote
+    
+    CORPUS_KEY = "YouTwo"
+    request_timeout = 20
+    request_timeout_millis = 60000
+    
+    # Validate corpus key
+    if len(CORPUS_KEY) > 50 or not all(c.isalnum() or c in ['_', '=', '-'] for c in CORPUS_KEY):
+        raise ValueError("corpus_key must be <= 50 characters and match regex [a-zA-Z0-9_\\=\\-]+$")
+    
+    # Prepare request
+    vectara_api_key = os.getenv("VECTARA_API_KEY")
+    
+    if not vectara_api_key:
+        raise VectaraAPIError("Vectara API key not found in environment variables")
+    
+    # Ensure document_id is percent encoded
+    encoded_document_id = quote(document_id)
+    
+    url = f"https://api.vectara.io/v2/corpora/{CORPUS_KEY}/documents/{encoded_document_id}"
+    
+    headers = {
+        "Accept": "application/json",
+        "x-api-key": vectara_api_key
+    }
+    
+    payload = {}
+    
+    # Set timeout parameters if needed
+    params = {}
+    if request_timeout is not None:
+        headers["Request-Timeout"] = str(request_timeout)
+    if request_timeout_millis is not None:
+        headers["Request-Timeout-Millis"] = str(request_timeout_millis)
+        
+    try:
+        response = requests.get(url, headers=headers, params=params, data=payload)
+        response.raise_for_status()
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        raise VectaraAPIError(f"Error fetching document from Vectara: {e}") from e
+    except Exception as e:
+        raise VectaraAPIError(f"An unexpected error occurred while fetching document: {e}") from e
+
 
 # This is still a placeholder
 def generate_llm_response(chat_state: list[dict], retrieved_chunks: list[str], summary: str) -> str:
@@ -250,6 +375,7 @@ def test_file_upload():
     except Exception as e:
         raise IndexingError(f"Error occurred while uploading PDF: {e}")
 
+
 if __name__ == "__main__":
     from dotenv import load_dotenv
     load_dotenv()