Upload 9 files

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+code_faiss.index filter=lfs diff=lfs merge=lfs -text

app.py

@@ -0,0 +1,170 @@
+# app.py
+import gradio as gr
+import requests
+import json
+import os
+from retrive_docs import load_faiss_index_and_metadata, retrieve_relevant_chunks ,print_results
+
+# --- CONFIGURATION ---
+INDEX_PATH = "code_faiss.index"
+METADATA_PATH = "code_metadata.json"
+CHUNKS_JSON_PATH = "code_chunks.json"
+EMBEDDING_MODEL_NAME = "Qwen/Qwen3-Embedding-0.6B"
+# TOP_K has been removed from here and is now a user input
+
+# --- SYSTEM PROMPT ---
+# This prompt is crucial for guiding the LLM's behavior.
+SYSTEM_PROMPT = """
+You are an expert software developer and technical analyst. Your task is to help a user understand a codebase and debug potential issues.
+
+You have been provided with a user's question and a set of the most relevant code chunks retrieved from the codebase based on their query.
+
+Your mission is to synthesize this information and provide a clear, accurate, and helpful response.
+
+Follow these instructions carefully:
+1.  **Analyze the Goal:** First, understand the user's primary goal. Are they reporting a bug, asking for an explanation, or trying to understand how something works?
+2.  **Base Your Answer on Provided Context:** Your primary source of truth is the retrieved code chunks. Ground your entire analysis in the code provided. Do not invent functionality or assume the existence of code that is not present in the context.
+3.  **Directly Address the Query:** Directly answer the user's question. If the context contains a definitive answer (e.g., a warning message about a known bug), state it clearly and quote the relevant code.
+4.  **Synthesize and Hypothesize:** If the answer is not immediately obvious, synthesize information from multiple chunks. Form a hypothesis about the cause of the bug or the functionality in question, explaining your reasoning by referencing specific lines of code.
+5.  **Provide Actionable Recommendations:** Conclude with clear, actionable advice. This could be a suggested code change, a command to run, or a recommendation to avoid a specific feature based on the evidence in the code.
+6.  **Acknowledge Limitations:** If the provided code chunks are insufficient to fully answer the question, state this clearly. Explain what additional information would be needed.
+7.  **Structure Your Response:** Format your response using Markdown for readability. Use code blocks for code snippets and bold text to highlight key findings.
+8. **show output reference at the end:** to keep a trust show the source where you get the information from, like if it included in line .. or code ...  if available only in the context. 
+"""
+
+# --- LOAD DATA ON STARTUP ---
+print("--- Initializing Application ---")
+# Check if all required files exist before launching the UI
+if not all(os.path.exists(p) for p in [INDEX_PATH, METADATA_PATH, CHUNKS_JSON_PATH]):
+    print("ERROR: One or more required data files are missing.")
+    print("Please make sure 'code_faiss.index', 'code_metadata.json', and 'code_chunks.json' are in the same directory.")
+    # Gradio doesn't have a clean way to exit, so we'll show an error in the UI
+    index, metadata, chunks_dict = None, None, None
+else:
+    index, metadata, chunks_dict = load_faiss_index_and_metadata(
+        index_path=INDEX_PATH,
+        metadata_path=METADATA_PATH,
+        chunks_json_path=CHUNKS_JSON_PATH
+    )
+print("--- Initialization Complete ---")
+
+
+def get_expert_analysis(api_key, api_url, llm_model_name, top_k, user_query):
+    """
+    The main function that orchestrates the RAG pipeline.
+    """
+    if not all([api_key, api_url, llm_model_name, user_query]):
+        return "Error: API Key, API URL, Model Name, and Question are all required."
+
+    if index is None:
+        return "Error: FAISS index and data could not be loaded. Please check the console for errors and restart."
+
+    # 1. RETRIEVAL: Get relevant code chunks
+    print("\n--- Starting Retrieval ---")
+    retrieved_results = retrieve_relevant_chunks(
+        query=user_query,
+        model_name=EMBEDDING_MODEL_NAME,
+        index=index,
+        metadata=metadata,
+        chunks_dict=chunks_dict,
+        top_k=top_k # Use the value from the UI
+    )
+
+    if not retrieved_results:
+        return "Could not find any relevant code chunks for your query. Please try rephrasing it."
+
+    context_str = print_results(retrieved_results)
+    
+    print("--- Starting Generation ---")
+    final_user_prompt = f"""
+{context_str}
+
+--- User's Question ---
+{user_query}
+
+--- Analysis and Answer ---
+Based on the provided code context, here is the analysis of your question:
+"""
+
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "model": llm_model_name,
+        "messages": [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": final_user_prompt}
+        ]
+    }
+
+    try:
+        print(f"Sending request to LLM: {llm_model_name} at {api_url}")
+        response = requests.post(api_url, headers=headers, data=json.dumps(payload))
+        response.raise_for_status()
+        
+        response_json = response.json()
+        llm_answer = response_json['choices'][0]['message']['content']
+        print("--- Generation Complete ---")
+        
+        full_response = f"## Expert Analysis\n\n{llm_answer}\n\n---\n\n### Retrieved Context\n\nThis analysis was based on the following retrieved code chunks:\n\n{context_str}"
+        return full_response
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error calling LLM API: {e}")
+        return f"Error: Failed to connect to the LLM API. Please check your API URL, API key, and network connection.\n\nDetails: {e}"
+    except (KeyError, IndexError) as e:
+        print(f"Error parsing LLM response: {e}")
+        return f"Error: Received an unexpected response from the LLM API. Please check the model name and try again.\n\nResponse: {response.text}"
+
+
+# --- GRADIO UI ---
+with gr.Blocks(theme=gr.themes.Soft(), title="RAG Code Assistant") as demo:
+    gr.Markdown("# RAG-Powered Code Assistant")
+    gr.Markdown("This tool uses a local code database (FAISS) and a Large Language Model (LLM) to answer questions about your codebase.")
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            api_key_input = gr.Textbox(
+                label="API Key",
+                type="password",
+                placeholder="Enter your API key here"
+            )
+            api_url_input = gr.Textbox(
+                label="API Endpoint URL",
+                value="https://openrouter.ai/api/v1/chat/completions",
+                placeholder="Enter the chat completions endpoint URL"
+            )
+            llm_model_input = gr.Textbox(
+                label="LLM Model Name",
+                value="moonshotai/kimi-k2:free",
+                placeholder="e.g., moonshotai/kimi-k2:free"
+            )
+            # New Dropdown for Top-K selection
+            top_k_input = gr.Dropdown(
+                label="Number of Chunks to Retrieve (Top K)",
+                choices=[5, 6, 7, 8, 9, 10],
+                value=10,
+            )
+            user_query_input = gr.Textbox(
+                label="Your Question / Bug Report",
+                lines=8,
+                placeholder="e.g., 'When I use cache=True, my RAM usage explodes. Why?'"
+            )
+            submit_button = gr.Button("Get Analysis", variant="primary")
+
+        with gr.Column(scale=2):
+            gr.Markdown("## Analysis Result")
+            output_text = gr.Markdown()
+
+    # Update the inputs list for the click event
+    submit_button.click(
+        fn=get_expert_analysis,
+        inputs=[api_key_input, api_url_input, llm_model_input, top_k_input, user_query_input],
+        outputs=output_text
+    )
+
+
+
+if __name__ == "__main__":
+    demo.launch(share=True)

code_faiss.index

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d0969bcdcff8b67e7a45cae54ce8b7a26f38829fa139e2bcabbb56c0310c469e
+size 3682349

create_chunks.py

@@ -0,0 +1,380 @@
+import ast
+import os
+import json
+from typing import Dict, List, Any, Optional, Union
+from dataclasses import dataclass, asdict
+from pathlib import Path
+import hashlib
+
+@dataclass
+class CodeChunk:
+    """Represents a chunk of code with metadata"""
+    content: str
+    chunk_type: str  # 'function', 'class', 'method', 'import', 'variable', 'comment', 'module'
+    name: str
+    file_path: str
+    start_line: int
+    end_line: int
+    start_col: int
+    end_col: int
+    parent_name: Optional[str] = None
+    docstring: Optional[str] = None
+    parameters: Optional[List[str]] = None
+    return_type: Optional[str] = None
+    decorators: Optional[List[str]] = None
+    complexity_score: Optional[int] = None
+    dependencies: Optional[List[str]] = None
+    chunk_id: Optional[str] = None
+    
+    def __post_init__(self):
+        # Generate unique ID based on content and location
+        content_hash = hashlib.md5(
+            f"{self.file_path}:{self.start_line}:{self.end_line}:{self.content}".encode()
+        ).hexdigest()[:8]
+        self.chunk_id = f"{self.chunk_type}_{self.name}_{content_hash}"
+
+class CodeChunker:
+    """Main class for chunking code using AST"""
+    
+    def __init__(self, supported_extensions: List[str] = None):
+        self.supported_extensions = supported_extensions or ['.py', '.js', '.ts', '.java', '.cpp', '.c', '.h']
+        self.chunks: List[CodeChunk] = []
+        
+    def chunk_file(self, file_path: str) -> List[CodeChunk]:
+        """Chunk a single file and return list of CodeChunk objects"""
+        file_path = Path(file_path)
+        
+        if file_path.suffix not in self.supported_extensions:
+            return []
+            
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                content = f.read()
+        except Exception as e:
+            print(f"Error reading file {file_path}: {e}")
+            return []
+        
+        if file_path.suffix == '.py':
+            return self._chunk_python_file(str(file_path), content)
+        else:
+            # For other languages, use simpler text-based chunking for now
+            return self._chunk_generic_file(str(file_path), content)
+    
+    def _chunk_python_file(self, file_path: str, content: str) -> List[CodeChunk]:
+        """Chunk Python file using AST"""
+        chunks = []
+        lines = content.split('\n')
+        
+        try:
+            tree = ast.parse(content)
+        except SyntaxError as e:
+            print(f"Syntax error in {file_path}: {e}")
+            return []
+        
+        # Track imports at module level
+        imports = []
+        
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.Import, ast.ImportFrom)):
+                imports.extend(self._extract_imports(node))
+        
+        # Process top-level nodes
+        for node in tree.body:
+            chunk = self._process_node(node, file_path, lines, imports)
+            if chunk:
+                chunks.append(chunk)
+        
+        return chunks
+    
+    def _process_node(self, node: ast.AST, file_path: str, lines: List[str], 
+                     imports: List[str], parent_name: str = None) -> Optional[CodeChunk]:
+        """Process an AST node and create a CodeChunk"""
+        
+        if isinstance(node, ast.FunctionDef):
+            return self._create_function_chunk(node, file_path, lines, imports, parent_name)
+        
+        elif isinstance(node, ast.AsyncFunctionDef):
+            return self._create_function_chunk(node, file_path, lines, imports, parent_name, is_async=True)
+        
+        elif isinstance(node, ast.ClassDef):
+            return self._create_class_chunk(node, file_path, lines, imports)
+        
+        elif isinstance(node, ast.Assign):
+            return self._create_variable_chunk(node, file_path, lines, parent_name)
+        
+        elif isinstance(node, (ast.Import, ast.ImportFrom)):
+            return self._create_import_chunk(node, file_path, lines)
+        
+        return None
+    
+    def _create_function_chunk(self, node: Union[ast.FunctionDef, ast.AsyncFunctionDef], 
+                              file_path: str, lines: List[str], imports: List[str], 
+                              parent_name: str = None, is_async: bool = False) -> CodeChunk:
+        """Create a chunk for a function or method"""
+        
+        # Extract function content
+        start_line = node.lineno
+        end_line = node.end_lineno or start_line
+        content = '\n'.join(lines[start_line-1:end_line])
+        
+        # Extract parameters
+        parameters = []
+        for arg in node.args.args:
+            param_str = arg.arg
+            if arg.annotation:
+                param_str += f": {ast.unparse(arg.annotation)}"
+            parameters.append(param_str)
+        
+        # Extract return type
+        return_type = None
+        if node.returns:
+            return_type = ast.unparse(node.returns)
+        
+        # Extract decorators
+        decorators = []
+        for decorator in node.decorator_list:
+            decorators.append(ast.unparse(decorator))
+        
+        # Extract docstring
+        docstring = ast.get_docstring(node)
+        
+        # Calculate complexity (simple metric based on control flow)
+        complexity = self._calculate_complexity(node)
+        
+        chunk_type = "method" if parent_name else "function"
+        if is_async:
+            chunk_type = "async_" + chunk_type
+        
+        return CodeChunk(
+            content=content,
+            chunk_type=chunk_type,
+            name=node.name,
+            file_path=file_path,
+            start_line=start_line,
+            end_line=end_line,
+            start_col=node.col_offset,
+            end_col=node.end_col_offset or 0,
+            parent_name=parent_name,
+            docstring=docstring,
+            parameters=parameters,
+            return_type=return_type,
+            decorators=decorators,
+            complexity_score=complexity,
+            dependencies=imports
+        )
+    
+    def _create_class_chunk(self, node: ast.ClassDef, file_path: str, 
+                           lines: List[str], imports: List[str]) -> CodeChunk:
+        """Create a chunk for a class"""
+        
+        start_line = node.lineno
+        end_line = node.end_lineno or start_line
+        content = '\n'.join(lines[start_line-1:end_line])
+        
+        # Extract base classes
+        base_classes = []
+        for base in node.bases:
+            base_classes.append(ast.unparse(base))
+        
+        # Extract decorators
+        decorators = []
+        for decorator in node.decorator_list:
+            decorators.append(ast.unparse(decorator))
+        
+        # Extract docstring
+        docstring = ast.get_docstring(node)
+        
+        return CodeChunk(
+            content=content,
+            chunk_type="class",
+            name=node.name,
+            file_path=file_path,
+            start_line=start_line,
+            end_line=end_line,
+            start_col=node.col_offset,
+            end_col=node.end_col_offset or 0,
+            docstring=docstring,
+            decorators=decorators,
+            dependencies=imports + base_classes
+        )
+    
+    def _create_variable_chunk(self, node: ast.Assign, file_path: str, 
+                              lines: List[str], parent_name: str = None) -> Optional[CodeChunk]:
+        """Create a chunk for variable assignments"""
+        
+        # Only process simple assignments at module level
+        if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
+            var_name = node.targets[0].id
+            start_line = node.lineno
+            end_line = node.end_lineno or start_line
+            content = '\n'.join(lines[start_line-1:end_line])
+            
+            return CodeChunk(
+                content=content,
+                chunk_type="variable",
+                name=var_name,
+                file_path=file_path,
+                start_line=start_line,
+                end_line=end_line,
+                start_col=node.col_offset,
+                end_col=node.end_col_offset or 0,
+                parent_name=parent_name
+            )
+        
+        return None
+    
+    def _create_import_chunk(self, node: Union[ast.Import, ast.ImportFrom], 
+                            file_path: str, lines: List[str]) -> CodeChunk:
+        """Create a chunk for import statements"""
+        
+        start_line = node.lineno
+        end_line = node.end_lineno or start_line
+        content = '\n'.join(lines[start_line-1:end_line])
+        
+        # Extract imported names
+        imported_names = []
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                imported_names.append(alias.name)
+        else:  # ImportFrom
+            for alias in node.names:
+                imported_names.append(alias.name)
+        
+        return CodeChunk(
+            content=content,
+            chunk_type="import",
+            name=", ".join(imported_names),
+            file_path=file_path,
+            start_line=start_line,
+            end_line=end_line,
+            start_col=node.col_offset,
+            end_col=node.end_col_offset or 0
+        )
+    
+    def _extract_imports(self, node: Union[ast.Import, ast.ImportFrom]) -> List[str]:
+        """Extract import names from import nodes"""
+        imports = []
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                imports.append(alias.name)
+        else:  # ImportFrom
+            module = node.module or ""
+            for alias in node.names:
+                imports.append(f"{module}.{alias.name}" if module else alias.name)
+        return imports
+    
+    def _calculate_complexity(self, node: ast.AST) -> int:
+        """Calculate cyclomatic complexity of a function"""
+        complexity = 1  # Base complexity
+        
+        for child in ast.walk(node):
+            if isinstance(child, (ast.If, ast.While, ast.For, ast.AsyncFor)):
+                complexity += 1
+            elif isinstance(child, ast.ExceptHandler):
+                complexity += 1
+            elif isinstance(child, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)):
+                complexity += 1
+        
+        return complexity
+    
+    def _chunk_generic_file(self, file_path: str, content: str) -> List[CodeChunk]:
+        """Generic chunking for non-Python files"""
+        chunks = []
+        lines = content.split('\n')
+        
+        # Simple function detection for C/C++/Java/JavaScript
+        function_patterns = {
+            '.js': r'function\s+(\w+)',
+            '.ts': r'function\s+(\w+)',
+            '.java': r'(public|private|protected)?\s*(static)?\s*\w+\s+(\w+)\s*\(',
+            '.cpp': r'\w+\s+(\w+)\s*\(',
+            '.c': r'\w+\s+(\w+)\s*\(',
+            '.h': r'\w+\s+(\w+)\s*\('
+        }
+        
+        # This is a simplified implementation - you'd want more sophisticated parsing
+        # for production use
+        
+        return chunks
+    
+    def chunk_directory(self, directory_path: str, recursive: bool = True) -> List[CodeChunk]:
+        """Chunk all supported files in a directory"""
+        all_chunks = []
+        directory_path = Path(directory_path)
+        
+        if recursive:
+            pattern = "**/*"
+        else:
+            pattern = "*"
+        
+        for file_path in directory_path.glob(pattern):
+            if file_path.is_file() and file_path.suffix in self.supported_extensions:
+                chunks = self.chunk_file(str(file_path))
+                all_chunks.extend(chunks)
+        
+        self.chunks = all_chunks
+        return all_chunks
+    
+    def save_chunks(self, output_file: str):
+        """Save chunks to JSON file"""
+        chunks_data = [asdict(chunk) for chunk in self.chunks]
+        
+        with open(output_file, 'w', encoding='utf-8') as f:
+            json.dump(chunks_data, f, indent=2, ensure_ascii=False)
+    
+    def load_chunks(self, input_file: str) -> List[CodeChunk]:
+        """Load chunks from JSON file"""
+        with open(input_file, 'r', encoding='utf-8') as f:
+            chunks_data = json.load(f)
+        
+        self.chunks = [CodeChunk(**chunk_data) for chunk_data in chunks_data]
+        return self.chunks
+    
+    def get_chunks_by_type(self, chunk_type: str) -> List[CodeChunk]:
+        """Filter chunks by type"""
+        return [chunk for chunk in self.chunks if chunk.chunk_type == chunk_type]
+    
+    def get_chunks_by_file(self, file_path: str) -> List[CodeChunk]:
+        """Filter chunks by file path"""
+        return [chunk for chunk in self.chunks if chunk.file_path == file_path]
+    
+    def search_chunks(self, query: str) -> List[CodeChunk]:
+        """Simple text search in chunks"""
+        results = []
+        query_lower = query.lower()
+        
+        for chunk in self.chunks:
+            if (query_lower in chunk.content.lower() or 
+                query_lower in chunk.name.lower() or
+                (chunk.docstring and query_lower in chunk.docstring.lower())):
+                results.append(chunk)
+        
+        return results
+
+# Example usage
+if __name__ == "__main__":
+    # Initialize chunker
+    chunker = CodeChunker()
+    
+    # Example: Chunk a single Python file
+    # chunks = chunker.chunk_file("example.py")
+    
+    # Example: Chunk entire directory
+    chunks = chunker.chunk_directory("ultralytics", recursive=True)
+    
+    # Example: Save chunks to file
+    chunker.save_chunks("code_chunks.json")
+    
+    # Example: Search chunks
+    # results = chunker.search_chunks("database")
+    
+    # Example: Get all functions
+    # functions = chunker.get_chunks_by_type("function")
+    
+    print("Code chunking system initialized!")
+    print("Supported file extensions:", chunker.supported_extensions)
+    print("\nExample usage:")
+    print("1. chunker.chunk_file('path/to/file.py')")
+    print("2. chunker.chunk_directory('path/to/project', recursive=True)")
+    print("3. chunker.save_chunks('output.json')")
+    print("4. chunker.search_chunks('query')")

create_faiss.py

@@ -0,0 +1,127 @@
+import json
+import faiss
+import numpy as np
+from sentence_transformers import SentenceTransformer
+from typing import List, Dict, Any
+
+def create_code_vector_db(json_file_path: str, model_name: str, output_index_path: str, output_metadata_path: str):
+    """
+    Loads code chunks, filters them, generates embeddings, and saves a FAISS index
+    along with corresponding metadata.
+
+    Args:
+        json_file_path (str): Path to the code_chunks.json file.
+        model_name (str): The name of the SentenceTransformer model to use.
+        output_index_path (str): Path to save the FAISS index file.
+        output_metadata_path (str): Path to save the chunk metadata JSON file.
+    """
+    # 1. Load and Filter Chunks
+    print(f"Loading chunks from '{json_file_path}'...")
+    try:
+        with open(json_file_path, 'r', encoding='utf-8') as f:
+            all_chunks = json.load(f)
+    except FileNotFoundError:
+        print(f"Error: The file '{json_file_path}' was not found.")
+        return
+
+    # Filter for chunks that contain meaningful semantic information for a RAG system
+    target_types = {'function', 'class', 'method', 'async_function', 'async_method'}
+    filtered_chunks = [chunk for chunk in all_chunks if chunk.get('chunk_type') in target_types]
+    
+    if not filtered_chunks:
+        print("No chunks of target types found. Exiting.")
+        return
+        
+    print(f"Filtered chunks: Kept {len(filtered_chunks)} out of {len(all_chunks)} total chunks.")
+
+    # 2. Prepare Text for Embedding
+    # Combine code with metadata for richer semantic representation.
+    texts_to_embed = []
+    for chunk in filtered_chunks:
+        # A good practice is to create a descriptive text for each chunk
+        docstring = chunk.get('docstring', '') or "No docstring."
+        name = chunk.get('name', '')
+        chunk_type = chunk.get('chunk_type', '')
+        
+        # Create a descriptive header for the code content
+        header = f"Type: {chunk_type}, Name: {name}\nDocstring: {docstring}\n---\n"
+        prepared_text = header + chunk['content']
+        texts_to_embed.append(prepared_text)
+
+    # 3. Generate Embeddings
+    print(f"Loading SentenceTransformer model: '{model_name}'...")
+    # Using a model well-suited for code is beneficial, but a general one works too.
+    # Consider models like 'microsoft/codebert-base' or 'all-MiniLM-L6-v2' for a start.
+    model = SentenceTransformer(model_name).half()  # Convert the model to half precision for faster inference
+    # model to fp16 for faster inference
+    # model = SentenceTransformer(model_name, device='cpu').half()
+    
+
+
+    
+    print("Generating embeddings for filtered chunks... (This may take a while)")
+    # embeddings = model.encode(texts_to_embed, show_progress_bar=True)
+    # Define a batch size
+    batch_size = 2 # You can adjust this number based on your VRAM
+
+    print("Generating embeddings for filtered chunks... (This may take a while)")
+    embeddings = model.encode(
+        texts_to_embed, 
+        batch_size=batch_size, 
+        show_progress_bar=True
+    )
+
+    # Convert to float32 for FAISS
+    embeddings = np.array(embeddings).astype('float32')
+    dimension = embeddings.shape[1]
+    print(f"Embeddings generated with dimension: {dimension}")
+
+    # 4. Build and Save FAISS Index
+    print("Building FAISS index...")
+    index = faiss.IndexFlatL2(dimension)
+    index.add(embeddings)
+    
+    print(f"Saving FAISS index to '{output_index_path}'...")
+    faiss.write_index(index, output_index_path)
+
+    # 5. Save Metadata for Mapping
+    # We need to save the original chunk info to map FAISS results back to the source code
+    metadata_to_save = [
+        {
+            "chunk_id": chunk.get("chunk_id"),
+            "file_path": chunk.get("file_path"),
+            "start_line": chunk.get("start_line"),
+            "end_line": chunk.get("end_line"),
+            "name": chunk.get("name"),
+            "chunk_type": chunk.get("chunk_type")
+        }
+        for chunk in filtered_chunks
+    ]
+    
+    print(f"Saving metadata mapping to '{output_metadata_path}'...")
+    with open(output_metadata_path, 'w', encoding='utf-8') as f:
+        json.dump(metadata_to_save, f, indent=2)
+        
+    print("\nProcess complete!")
+    print(f"FAISS index and metadata have been successfully saved.")
+
+
+if __name__ == "__main__":
+    # --- CONFIGURATION ---
+    CHUNKS_JSON_PATH = "code_chunks.json"
+    
+    # Recommended model for general purpose, good balance of speed and quality.
+    # For more code-specific tasks, you might explore models like 'microsoft/codebert-base'.
+    MODEL_NAME = "Qwen/Qwen3-Embedding-0.6B"
+    
+    OUTPUT_INDEX_PATH = "code_faiss.index"
+    OUTPUT_METADATA_PATH = "code_metadata.json"
+    
+    # --- EXECUTION ---
+    create_code_vector_db(
+        json_file_path=CHUNKS_JSON_PATH,
+        model_name=MODEL_NAME,
+        output_index_path=OUTPUT_INDEX_PATH,
+        output_metadata_path=OUTPUT_METADATA_PATH
+    )
+

llm_calling.py

@@ -0,0 +1,103 @@
+import retrive_docs
+
+
+import json
+from retrive_docs import load_faiss_index_and_metadata, retrieve_relevant_chunks, print_results
+
+INDEX_PATH = "code_faiss.index"
+METADATA_PATH = "code_metadata.json"
+CHUNKS_JSON_PATH = "code_chunks.json"
+MODEL_NAME = "Qwen/Qwen3-Embedding-0.6B"  # Must match the model used in create_faiss.py
+TOP_K = 5  # Number of results to retrieve
+
+# --- EXECUTION ---
+# Load FAISS index and metadata
+index, metadata, chunks_dict = load_faiss_index_and_metadata(
+    index_path=INDEX_PATH,
+    metadata_path=METADATA_PATH,
+    chunks_json_path=CHUNKS_JSON_PATH
+)
+
+if index is None or metadata is None or chunks_dict is None:
+    print("Failed to load index, metadata, or chunks. Exiting.")
+    exit(1)
+
+# Get user query
+print("\nEnter your query (e.g., 'function to process text data'):")
+# query = input("> ")
+query= '''
+Bug
+when i add (cache=True)in Classification Training , the Ram using is increasing every epoch , until it crash the training , start like from 3 to 6 to 11 to 15 ....... 50 , GB
+but if i don't add it , the ram using work fine , it be like 4 GB and all training is fixed
+
+i work on colab
+!yolo task=classify mode=train cache=True model=yolov8n-cls.pt data='/content/Classification-1' epochs=5  batch=265 imgsz=128 
+
+Environment
+No response
+
+Minimal Reproducible Example
+No response
+
+Additional
+No response'''
+# Retrieve and display results
+results = retrieve_relevant_chunks(
+    query=query,
+    model_name=MODEL_NAME,
+    index=index,
+    metadata=metadata,
+    chunks_dict=chunks_dict,
+    top_k=TOP_K
+    )
+
+
+print(print_results(results))
+#call llm
+# import requests
+# import json
+# import time
+# import os
+
+# sys_prompt = "You ar "
+# # Set API key and API base for the custom API server
+# api_key = os.getenv("API_KEY")  # Replace with your actual API key
+# api_base_url = os.getenv("API_BASE_URL")  # Replace with your API base URL
+
+# # Setup headers for the request
+# headers = {
+#     "Authorization": f"Bearer {api_key}",
+#     "Content-Type": "application/json"
+# }
+
+# # System message and query
+# # sys_msg = "you are a helpful assistant"
+# # query = "what is machine learning?"
+
+# # Prepare the data payload for the POST request
+# data = json.dumps({
+#     "model": "Meta-Llama-3.1-8B-Instruct-AWQ-INT4",
+#     "messages": [
+#         {"role": "system", "content":sys_prompt },
+#         {"role": "user", "content": query}
+#     ],
+#     "temperature": 0.2
+# })
+
+# # Measure request execution time
+# t1 = time.time()
+
+# # Perform the POST request
+# response = requests.post(f"{api_base_url}/chat/completions", headers=headers, data=data)
+# print("Request time:", time.time() - t1)
+
+# # Check the response and handle errors
+# if response.status_code == 200:
+#     # Parse response if request was successful
+#     chat_response = response.json()
+#     print("Chat response:", chat_response['choices'][0]['message']['content'])
+# else:
+#     # Print error information if something went wrong
+#     print("Failed to fetch response:", response.status_code, response.text)
+
+# print("this output based on this query :",query) 

requirements.txt

@@ -0,0 +1,6 @@
+gradio==4.31.5
+requests==2.31.0
+sentence-transformers==2.7.0
+faiss-cpu==1.8.0
+numpy==1.26.4
+torch==2.3.0

retrive_docs.py

@@ -0,0 +1,200 @@
+import json
+import faiss
+import numpy as np
+from sentence_transformers import SentenceTransformer
+from typing import List, Dict, Any
+
+model_name = "Qwen/Qwen3-Embedding-0.6B"
+print(f"Loading SentenceTransformer model: '{model_name}'...")
+device = 'cuda' if faiss.get_num_gpus() > 0 else 'cpu'
+print(f"Using device: {device}")
+model = SentenceTransformer(model_name, device=device)
+if device == 'cuda':
+    model = model.half()  # Use FP16 for GPU
+def load_faiss_index_and_metadata(index_path: str, metadata_path: str, chunks_json_path: str) -> tuple:
+    """
+    Loads the FAISS index, metadata, and original code chunks.
+
+    Args:
+        index_path (str): Path to the FAISS index file.
+        metadata_path (str): Path to the metadata JSON file.
+        chunks_json_path (str): Path to the original code_chunks.json file.
+
+    Returns:
+        tuple: (FAISS index, metadata list, chunks dictionary)
+    """
+    print(f"Loading FAISS index from '{index_path}'...")
+    try:
+        index = faiss.read_index(index_path)
+    except Exception as e:
+        print(f"Error loading FAISS index: {e}")
+        return None, None, None
+
+    print(f"Loading metadata from '{metadata_path}'...")
+    try:
+        with open(metadata_path, 'r', encoding='utf-8') as f:
+            metadata = json.load(f)
+    except FileNotFoundError:
+        print(f"Error: The file '{metadata_path}' was not found.")
+        return None, None, None
+
+    print(f"Loading code chunks from '{chunks_json_path}'...")
+    try:
+        with open(chunks_json_path, 'r', encoding='utf-8') as f:
+            chunks = json.load(f)
+        # Create a dictionary for quick lookup by chunk_id
+        chunks_dict = {chunk['chunk_id']: chunk for chunk in chunks}
+    except FileNotFoundError:
+        print(f"Error: The file '{chunks_json_path}' was not found.")
+        return None, None, None
+
+    return index, metadata, chunks_dict
+
+def retrieve_relevant_chunks(query: str, model_name: str, index: faiss.IndexFlatL2, metadata: List[Dict], chunks_dict: Dict, top_k: int = 5) -> List[Dict]:
+    """
+    Encodes the query and retrieves the top-k most relevant code chunks.
+
+    Args:
+        query (str): The user's input query.
+        model_name (str): The SentenceTransformer model to use.
+        index (faiss.IndexFlatL2): The loaded FAISS index.
+        metadata (List[Dict]): The metadata for the indexed chunks.
+        chunks_dict (Dict): Dictionary mapping chunk_id to chunk details.
+        top_k (int): Number of top results to return.
+
+    Returns:
+        List[Dict]: List of dictionaries containing the retrieved chunks and their metadata.
+    """
+    # Load the model
+    
+
+    # Prepare query text (mimic the chunk format used during indexing)
+    query_text = f"Type: query\nDocstring: {query}\n---\n{query}"
+    query_embedding = model.encode([query_text], show_progress_bar=False).astype('float32')
+
+    # Perform FAISS search
+    print(f"Searching for top {top_k} relevant chunks...")
+    distances, indices = index.search(query_embedding, top_k)
+
+    # Collect results
+    results = []
+    for idx, distance in zip(indices[0], distances[0]):
+        if idx < len(metadata):
+            meta = metadata[idx]
+            chunk_id = meta['chunk_id']
+            chunk = chunks_dict.get(chunk_id, {})
+            results.append({
+                'chunk_id': chunk_id,
+                'file_path': meta.get('file_path', 'Unknown'),
+                'start_line': meta.get('start_line', -1),
+                'end_line': meta.get('end_line', -1),
+                'name': meta.get('name', 'Unknown'),
+                'chunk_type': meta.get('chunk_type', 'Unknown'),
+                'docstring': chunk.get('docstring', 'No docstring.'),
+                'content': chunk.get('content', 'No content available.'),
+                'distance': float(distance)  # Similarity score (L2 distance)
+            })
+        else:
+            print(f"Warning: Index {idx} out of range for metadata.")
+
+    return results
+
+# def print_results(results: List[Dict]):
+#     """
+#     Prints the retrieved results in a readable format.
+
+#     Args:
+#         results (List[Dict]): List of retrieved chunk details.
+#     """
+#     if not results:
+#         print("No relevant chunks found.")
+#         return
+
+#     print("\n=== Retrieved Chunks ===")
+#     returned_text=""
+#     for i, result in enumerate(results, 1):
+#         # print(f"\nResult {i}:")
+#         # print(f"Chunk ID: {result['chunk_id']}")
+#         # print(f"Type: {result['chunk_type']}")
+#         # print(f"Name: {result['name']}")
+#         # print(f"File: {result['file_path']} (Lines {result['start_line']}–{result['end_line']})")
+#         # print(f"Distance: {result['distance']:.4f}")
+#         # print(f"Docstring: {result['docstring']}")
+#         # print("\nCode:")
+#         # print(result['content'])
+#         # print("-" * 80)
+#         returned_text=returned_text + "\n" +"chunk_id: " + "\n"+ f"File: {result['file_path']} (Lines {result['start_line']}–{result['end_line']})" + "\n" + result['chunk_id'] + "\n" +"code: " + result['content'] 
+#     # return in style 
+#     return returned_text
+#     #return { {'results': for result in results  }
+# # In retrive_docs.py
+
+def print_results(results: List[Dict]):
+    """
+    Formats the retrieved results into a Markdown string with GitHub links
+    and syntax highlighting.
+
+    Args:
+        results (List[Dict]): List of retrieved chunk details.
+    """
+    if not results:
+        return "No relevant chunks found."
+
+    GITHUB_BASE_URL = "https://github.com/ultralytics/ultralytics/blob/main/"
+    markdown_output = ""
+
+    for i, result in enumerate(results, 1):
+        file_path = result.get('file_path', 'Unknown')
+        start_line = result.get('start_line', -1)
+        end_line = result.get('end_line', -1)
+
+        # Construct a direct link to the code on GitHub
+        if file_path != 'Unknown' and start_line != -1:
+            github_link = f"{GITHUB_BASE_URL}{file_path}#L{start_line}-L{end_line}"
+            header = f"### {i}. [{file_path}]({github_link}) (Lines {start_line}–{end_line})"
+        else:
+            header = f"### {i}. {file_path} (Lines {start_line}–{end_line})"
+
+        markdown_output += f"{header}\n"
+        markdown_output += f"**Type:** `{result.get('chunk_type', 'N/A')}`  **Name:** `{result.get('name', 'N/A')}`\n\n"
+        markdown_output += "```python\n"
+        markdown_output += result.get('content', '# No content available.') + "\n"
+        markdown_output += "```\n---\n"
+
+    return markdown_output
+
+# if __name__ == "__main__":
+#     # --- CONFIGURATION ---
+#     INDEX_PATH = "code_faiss.index"
+#     METADATA_PATH = "code_metadata.json"
+#     CHUNKS_JSON_PATH = "code_chunks.json"
+#     MODEL_NAME = "Qwen/Qwen3-Embedding-0.6B"  # Must match the model used in create_faiss.py
+#     TOP_K = 5  # Number of results to retrieve
+
+#     # --- EXECUTION ---
+#     # Load FAISS index and metadata
+#     index, metadata, chunks_dict = load_faiss_index_and_metadata(
+#         index_path=INDEX_PATH,
+#         metadata_path=METADATA_PATH,
+#         chunks_json_path=CHUNKS_JSON_PATH
+#     )
+
+#     if index is None or metadata is None or chunks_dict is None:
+#         print("Failed to load index, metadata, or chunks. Exiting.")
+#         exit(1)
+
+#     # Get user query
+#     print("\nEnter your query (e.g., 'function to process text data'):")
+#     query = input("> ")
+
+#     # Retrieve and display results
+#     results = retrieve_relevant_chunks(
+#         query=query,
+#         model_name=MODEL_NAME,
+#         index=index,
+#         metadata=metadata,
+#         chunks_dict=chunks_dict,
+#         top_k=TOP_K
+#     )
+
+#     print_results(results)