INTRODUCTION
The modern software development landscape generates enormous amounts of code, documentation, and metadata across distributed version control systems. Developers and teams need intelligent tools that can understand repositories holistically, track changes continuously, and answer questions about codebases without requiring manual exploration. This tutorial presents an approach to building an agentic AI system that combines large language models with advanced retrieval techniques to analyze GitHub and GitLab repositories.
The system we will construct goes beyond simple code indexing. It employs a hybrid knowledge representation strategy that combines Retrieval Augmented Generation with Graph-based RAG, creating bidirectional relationships between vector embeddings and graph nodes. This dual representation allows the system to understand both semantic similarities in code and structural relationships between components. The agent continuously monitors repositories for changes, updates its knowledge base incrementally, and provides natural language summaries and answers to user queries.
The architecture supports both local and remote repositories, works with various LLM backends including local and cloud-based models, and runs on diverse GPU architectures from Intel, AMD, Apple, and Nvidia. The system persists its knowledge to disk, enabling efficient reanalysis of previously examined repositories.
ARCHITECTURAL OVERVIEW
The agentic AI system consists of several interconnected components that work together to provide comprehensive repository analysis. At the highest level, the architecture separates concerns into distinct layers: the repository interface layer, the knowledge extraction and representation layer, the storage and persistence layer, the LLM integration layer, and the user interaction layer.
The repository interface layer handles connections to both local and remote repositories. It abstracts away the differences between GitHub and GitLab APIs, providing a unified interface for accessing repository contents, commit histories, issues, pull requests, and continuous integration configurations. This layer also implements change detection mechanisms that monitor repositories for updates.
The knowledge extraction layer processes raw repository data and transforms it into structured representations. It analyzes source code files, documentation, configuration files, and metadata. The extraction process identifies key entities such as functions, classes, modules, dependencies, and their relationships. This layer feeds information into both the vector embedding pipeline and the graph construction pipeline.
The knowledge representation layer maintains the hybrid RAG and GraphRAG system. Vector embeddings capture semantic meaning of code snippets, documentation, and comments. These embeddings enable similarity-based retrieval when users ask questions. Simultaneously, a graph database stores structural relationships between repository entities. Nodes represent files, functions, classes, issues, commits, and other artifacts. Edges represent relationships such as function calls, inheritance hierarchies, file dependencies, and issue-commit associations. Critically, bidirectional links connect vector embeddings to graph nodes, allowing the system to leverage both semantic and structural information during retrieval.
The storage layer persists all knowledge to disk in a structured format. Each analyzed repository gets its own directory containing vector indices, graph databases, metadata files, and change tracking information. This persistence enables quick reloading of previously analyzed repositories without reprocessing.
The LLM integration layer provides a flexible interface to various language models. It supports local models running on different GPU architectures through frameworks like llama.cpp, vLLM, and Ollama. It also supports remote APIs such as OpenAI, Anthropic, and others. The layer handles prompt construction, context management, and response parsing.
The user interaction layer presents information to users through natural language summaries and interactive question-answering. It generates comprehensive reports when analyzing new repositories and provides concise update notifications when changes occur. An icon-based notification system alerts users to repository changes.
SELECTING THE RIGHT LANGUAGE MODEL
Choosing an appropriate language model significantly impacts the system's performance, cost, and deployment flexibility. The decision involves tradeoffs between model capability, inference speed, memory requirements, and operational costs.
For production deployments requiring maximum accuracy and reasoning capability, large frontier models like GPT-4, Claude 3 Opus, or Gemini 1.5 Pro offer superior performance. These models excel at understanding complex code structures, generating accurate summaries, and answering nuanced questions about repositories. However, they require API access, incur per-token costs, and introduce latency from network communication.
For deployments prioritizing privacy, cost control, or offline operation, locally hosted open-source models provide compelling alternatives. Models in the 7B to 13B parameter range, such as Llama 3 8B, Mistral 7B, or CodeLlama 13B, offer reasonable performance for code understanding tasks while running on consumer-grade GPUs. Larger models like Llama 3 70B or Mixtral 8x7B provide better reasoning at the cost of increased memory requirements.
Code-specialized models deserve special consideration for repository analysis tasks. CodeLlama, StarCoder, and WizardCoder have been trained specifically on code and demonstrate superior performance on programming-related tasks compared to general-purpose models of similar size. These models better understand programming language syntax, common patterns, and software engineering concepts.
The system architecture supports multiple LLM backends simultaneously, allowing different models for different tasks. For example, a smaller fast model might handle change detection and simple queries, while a larger model processes complex analytical questions. This hybrid approach optimizes the tradeoff between performance and resource utilization.
GPU architecture compatibility requires careful attention. Nvidia GPUs with CUDA support offer the broadest compatibility with inference frameworks. AMD GPUs require ROCm support, which works with frameworks like vLLM and some PyTorch-based solutions. Apple Silicon Macs leverage Metal Performance Shaders through MLX or llama.cpp. Intel GPUs can use oneAPI or OpenVINO for acceleration. The system abstracts these differences through a unified LLM interface.
REPOSITORY CONNECTION AND DATA EXTRACTION
The first step in analyzing a repository involves establishing a connection and extracting relevant data. The system must handle both local filesystem paths and remote repository URLs. For remote repositories, it uses the appropriate API (GitHub REST/GraphQL or GitLab API) to fetch data without requiring a full clone for initial analysis.
Here is a simplified example of the repository connector interface:
class RepositoryConnector:
def __init__(self, repo_path_or_url, credentials=None):
self.location = repo_path_or_url
self.credentials = credentials
self.is_remote = self._determine_if_remote()
self.repo_type = self._determine_repo_type()
def _determine_if_remote(self):
return self.location.startswith('http://') or \
self.location.startswith('https://') or \
self.location.startswith('git@')
def _determine_repo_type(self):
if 'github.com' in self.location:
return 'github'
elif 'gitlab.com' in self.location or 'gitlab' in self.location:
return 'gitlab'
else:
return 'local'
This connector determines whether the repository is local or remote and identifies the hosting platform. For remote repositories, it initializes the appropriate API client with authentication credentials if provided. The connector provides unified methods for accessing repository contents regardless of the underlying source.
The data extraction process retrieves multiple types of information from the repository. Source code files form the primary content, but the system also extracts documentation files, configuration files, commit history, branch information, issues, pull requests or merge requests, continuous integration pipeline definitions, and README files.
For local repositories, the system uses Git libraries to access the repository structure and history. For remote repositories, it leverages platform APIs to fetch data efficiently. The GitHub GraphQL API allows fetching multiple related resources in a single request, reducing API calls. GitLab's REST API provides comprehensive access to repository data, issues, and merge requests.
Example of extracting file contents from a repository:
def extract_file_contents(self, file_path, ref='main'):
if self.is_remote:
if self.repo_type == 'github':
return self._fetch_github_file(file_path, ref)
elif self.repo_type == 'gitlab':
return self._fetch_gitlab_file(file_path, ref)
else:
return self._read_local_file(file_path, ref)
def _read_local_file(self, file_path, ref):
repo = git.Repo(self.location)
commit = repo.commit(ref)
blob = commit.tree / file_path
return blob.data_stream.read().decode('utf-8')
The extraction process handles binary files appropriately, skipping them or extracting metadata without attempting to process their contents as text. It respects repository size limits and implements pagination for large repositories to avoid memory exhaustion.
The system also extracts metadata about the repository structure. It builds a directory tree representation, identifies programming languages used, detects framework and library dependencies from package manifest files, and catalogs configuration files. This metadata helps users understand the repository organization at a high level.
Issue and pull request data provides context about ongoing work and known problems. The system extracts issue titles, descriptions, labels, assignees, and current status. For pull requests, it captures the source and target branches, review status, and associated commits. This information helps users understand the development activity and priorities.
IMPLEMENTING THE HYBRID RAG AND GRAPHRAG SYSTEM
The knowledge representation system combines vector-based retrieval with graph-based relationships to provide comprehensive understanding of repository contents. This hybrid approach leverages the strengths of both paradigms: semantic similarity from embeddings and structural relationships from graphs.
The vector embedding component processes text chunks from the repository and generates dense vector representations. These embeddings capture semantic meaning, allowing the system to find relevant information even when exact keywords do not match. For code repositories, the system chunks content intelligently, preserving logical boundaries like function definitions, class declarations, and documentation blocks.
Let’s create an example of the chunking strategy for source code:
def chunk_source_file(self, file_path, content, language):
if language in ['python', 'java', 'javascript', 'typescript']:
return self._chunk_by_syntax(content, language)
else:
return self._chunk_by_lines(content, max_lines=50, overlap=5)
def _chunk_by_syntax(self, content, language):
tree = self.parser.parse(bytes(content, 'utf8'), language)
chunks = []
for node in tree.root_node.children:
if node.type in ['function_definition', 'class_definition',
'method_definition']:
chunk_text = content[node.start_byte:node.end_byte]
chunks.append({
'text': chunk_text,
'type': node.type,
'start_line': node.start_point[0],
'end_line': node.end_point[0]
})
return chunks
This chunking approach uses tree-sitter parsers to understand code syntax and create chunks that align with logical code units. Functions and classes become individual chunks, preserving their complete context. For languages without parser support, the system falls back to line-based chunking with overlap to avoid splitting related content.
The embedding model converts these chunks into vectors. For code-specific tasks, models like CodeBERT, GraphCodeBERT, or UniXcoder provide better representations than general-purpose embedding models. These specialized models understand programming language syntax and semantics. The system stores embeddings in a vector database like FAISS, Chroma, or Qdrant for efficient similarity search.
The graph component represents structural relationships between repository entities. Nodes in the graph represent files, functions, classes, variables, issues, commits, branches, and other artifacts. Edges represent various relationships:
Function calls create edges between caller and callee functions. Class inheritance creates edges from subclasses to parent classes. Import or include statements create edges between files. Variable references create edges from usage sites to declaration sites. Commits create edges to the files they modify. Issues create edges to related commits and pull requests. Pull requests create edges to source and target branches.
Let‘s view an example of building graph nodes for a Python file:
def build_graph_nodes_for_python_file(self, file_path, content):
tree = ast.parse(content)
nodes = []
edges = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
func_node = {
'id': f"{file_path}::{node.name}",
'type': 'function',
'name': node.name,
'file': file_path,
'line': node.lineno,
'docstring': ast.get_docstring(node)
}
nodes.append(func_node)
# Extract function calls to create edges
for child in ast.walk(node):
if isinstance(child, ast.Call):
if isinstance(child.func, ast.Name):
edges.append({
'from': func_node['id'],
'to': child.func.id,
'type': 'calls'
})
return nodes, edges
This code parses Python source files using the abstract syntax tree module and extracts function definitions as graph nodes. It also identifies function calls to create edges representing the call graph. Similar analysis applies to other programming languages using appropriate parsers.
The critical innovation in this system is the bidirectional linking between vector embeddings and graph nodes. Each embedding has an associated graph node ID, and each graph node references its corresponding embedding. This linkage enables powerful hybrid retrieval strategies.
When a user asks a question, the system performs vector similarity search to find semantically relevant chunks. It then traverses the graph from these chunks to discover structurally related entities. For example, if a user asks about a specific function, the system finds the function's embedding, retrieves the graph node, and traverses edges to find all functions that call it, all functions it calls, and all files it depends on. This combined approach provides more complete and contextually relevant answers than either method alone.
Let’s look at an example of hybrid retrieval:
def hybrid_retrieve(self, query, top_k=5):
# Vector similarity search
query_embedding = self.embed_text(query)
similar_chunks = self.vector_db.similarity_search(
query_embedding,
k=top_k
)
# Get graph nodes for similar chunks
node_ids = [chunk.metadata['node_id'] for chunk in similar_chunks]
# Expand context by traversing graph
expanded_nodes = []
for node_id in node_ids:
node = self.graph_db.get_node(node_id)
expanded_nodes.append(node)
# Get neighboring nodes
neighbors = self.graph_db.get_neighbors(
node_id,
max_depth=2,
edge_types=['calls', 'called_by', 'imports', 'imported_by']
)
expanded_nodes.extend(neighbors)
# Retrieve embeddings for expanded nodes
expanded_chunks = []
for node in expanded_nodes:
if node.embedding_id:
chunk = self.vector_db.get_by_id(node.embedding_id)
expanded_chunks.append(chunk)
return expanded_chunks
This retrieval function first finds semantically similar chunks through vector search, then expands the context by traversing the graph to include related entities. The expanded set of chunks provides richer context for the language model to generate accurate answers.
GENERATING COMPREHENSIVE REPOSITORY REPORTS
When the system first analyzes a repository, it generates a comprehensive report that gives users a complete understanding of the codebase. This report synthesizes information from multiple sources: the file structure, code analysis, commit history, issues, and configuration files.
The report generation process follows a structured approach. First, the system gathers all relevant data from the repository. Second, it analyzes this data to extract key insights. Third, it organizes the insights into logical sections. Fourth, it uses the language model to generate natural language descriptions. Finally, it formats the report for presentation to the user.
The repository overview section provides high-level information about the project. It includes the repository name, description, primary programming languages, total number of files and lines of code, creation date, and last update time. The system also identifies the main branch and lists all active branches.
The directory structure section explains the purpose of each major directory in the repository. The system uses heuristics and naming conventions to infer directory purposes. For example, directories named "src" or "lib" typically contain source code, "test" or "tests" contain test files, "docs" contain documentation, and "config" or "conf" contain configuration files. The language model enhances these heuristics by analyzing the actual contents and generating descriptions.
Here is an example of generating directory descriptions:
def generate_directory_description(self, dir_path, files):
# Gather context about the directory
file_types = self._analyze_file_types(files)
sample_files = files[:5] # Take first 5 files as samples
# Build prompt for LLM
prompt = f"""
Analyze this directory in a code repository:
Directory path: {dir_path}
File types present:
{self._format_file_types(file_types)}
Sample files:
{self._format_file_list(sample_files)}
Based on this information, provide a concise description of what this
directory contains and its purpose in the repository. Focus on the
functionality and role within the overall project.
"""
description = self.llm.generate(prompt, max_tokens=150)
return description
This function collects information about a directory and uses the language model to generate a natural language description. The LLM considers file types, naming patterns, and sample file names to infer the directory's purpose.
The code structure section describes the main components of the codebase. It identifies key classes, important functions, entry points, and architectural patterns. For object-oriented codebases, it describes the class hierarchy and main abstractions. For functional codebases, it describes the module organization and key functions.
The dependencies section lists external libraries and frameworks used by the project. It extracts this information from package manifest files like package.json, requirements.txt, pom.xml, or Cargo.toml. The system categorizes dependencies into production dependencies, development dependencies, and optional dependencies. It also identifies the versions being used and checks for known security vulnerabilities if vulnerability databases are available.
The issues and pull requests section summarizes open issues and active pull requests. It groups issues by labels or categories, highlights high-priority or critical issues, and identifies long-standing issues that may need attention. For pull requests, it notes which ones are ready for review, which are in draft status, and which have merge conflicts.
The continuous integration section describes the CI/CD pipelines defined in the repository. It identifies the CI platform being used, such as GitHub Actions, GitLab CI, Jenkins, or CircleCI. It lists the defined workflows or pipelines, describes what each pipeline does, and notes the status of recent pipeline runs.
The recent activity section summarizes recent changes to the repository. It analyzes recent commits to identify areas of active development, notes which files are being modified frequently, and identifies contributors who have been most active recently.
Below is an example of the complete report generation:
def generate_repository_report(self, repo_data):
report_sections = []
# Overview section
overview = self._generate_overview(repo_data)
report_sections.append(("REPOSITORY OVERVIEW", overview))
# Directory structure section
dir_structure = self._generate_directory_structure(repo_data)
report_sections.append(("DIRECTORY STRUCTURE", dir_structure))
# Code structure section
code_structure = self._generate_code_structure(repo_data)
report_sections.append(("CODE STRUCTURE", code_structure))
# Dependencies section
dependencies = self._generate_dependencies_section(repo_data)
report_sections.append(("DEPENDENCIES", dependencies))
# Issues section
issues = self._generate_issues_section(repo_data)
report_sections.append(("OPEN ISSUES", issues))
# CI/CD section
cicd = self._generate_cicd_section(repo_data)
report_sections.append(("CONTINUOUS INTEGRATION", cicd))
# Recent activity section
activity = self._generate_activity_section(repo_data)
report_sections.append(("RECENT ACTIVITY", activity))
# Format the complete report
return self._format_report(report_sections)
The report generation process creates each section independently and then combines them into a cohesive document. Each section uses the language model to generate natural language descriptions based on the extracted data and analysis.
IMPLEMENTING CONTINUOUS CHANGE TRACKING
One of the most valuable features of the agentic AI system is its ability to continuously monitor repositories for changes and update its knowledge base accordingly. This capability ensures that the system's understanding remains current as the repository evolves.
The change tracking mechanism operates on a polling basis for remote repositories and can use filesystem watchers for local repositories. The polling interval is configurable, with typical values ranging from five minutes for actively developed repositories to several hours for stable repositories.
When checking for changes, the system compares the current repository state against its stored snapshot. For Git repositories, this involves comparing commit hashes on tracked branches. If new commits exist, the system fetches the commit details and identifies which files were added, modified, or deleted.
Let us create an example of the change detection logic:
def detect_changes(self, repo_id):
stored_state = self.storage.load_repo_state(repo_id)
current_state = self.connector.get_current_state()
changes = {
'new_commits': [],
'modified_files': [],
'new_files': [],
'deleted_files': [],
'new_issues': [],
'updated_issues': [],
'new_prs': []
}
# Compare commits
for branch in current_state['branches']:
stored_head = stored_state['branches'].get(branch, None)
current_head = current_state['branches'][branch]
if stored_head != current_head:
new_commits = self.connector.get_commits_between(
stored_head,
current_head
)
changes['new_commits'].extend(new_commits)
# Analyze commits to find file changes
for commit in changes['new_commits']:
for file_change in commit['files']:
if file_change['status'] == 'added':
changes['new_files'].append(file_change['path'])
elif file_change['status'] == 'modified':
changes['modified_files'].append(file_change['path'])
elif file_change['status'] == 'deleted':
changes['deleted_files'].append(file_change['path'])
return changes
This function compares the stored repository state with the current state to identify all changes. It tracks new commits, file modifications, additions, and deletions. Similar logic applies to issues and pull requests.
When changes are detected, the system updates its knowledge base incrementally rather than reprocessing the entire repository. For new files, it performs the full analysis pipeline: chunking, embedding, and graph node creation. For modified files, it removes the old embeddings and graph nodes, then processes the new version. For deleted files, it removes the associated embeddings and graph nodes but may retain historical information.
The incremental update process maintains consistency between the vector database and graph database. When updating a file, the system must update both representations atomically to avoid inconsistencies.
And now an example of incremental knowledge base update:
def update_knowledge_base(self, repo_id, changes):
# Handle deleted files
for file_path in changes['deleted_files']:
self._remove_file_from_knowledge_base(repo_id, file_path)
# Handle modified files
for file_path in changes['modified_files']:
# Remove old version
self._remove_file_from_knowledge_base(repo_id, file_path)
# Add new version
content = self.connector.get_file_content(file_path)
self._add_file_to_knowledge_base(repo_id, file_path, content)
# Handle new files
for file_path in changes['new_files']:
content = self.connector.get_file_content(file_path)
self._add_file_to_knowledge_base(repo_id, file_path, content)
# Update commit history
for commit in changes['new_commits']:
self._add_commit_to_knowledge_base(repo_id, commit)
def _add_file_to_knowledge_base(self, repo_id, file_path, content):
# Chunk the file
chunks = self.chunker.chunk_file(file_path, content)
# Generate embeddings
embeddings = [self.embedder.embed(chunk['text']) for chunk in chunks]
# Store in vector database
embedding_ids = self.vector_db.add_embeddings(
repo_id,
embeddings,
chunks
)
# Build graph nodes
nodes, edges = self.graph_builder.build_nodes(file_path, content)
# Link embeddings to graph nodes
for i, node in enumerate(nodes):
node['embedding_id'] = embedding_ids[i]
# Store in graph database
self.graph_db.add_nodes(repo_id, nodes)
self.graph_db.add_edges(repo_id, edges)
This update process handles file deletions, modifications, and additions. It maintains the bidirectional links between embeddings and graph nodes during updates.
After updating the knowledge base, the system generates a change summary for the user. This summary describes what changed, where the changes occurred, and potentially why based on commit messages. The language model synthesizes the raw change data into a readable narrative.
How might the change summary generation look like:
def generate_change_summary(self, repo_id, changes):
if not self._has_significant_changes(changes):
return None
# Build context about changes
context = self._build_change_context(changes)
prompt = f"""
A repository has been updated with the following changes:
New commits: {len(changes['new_commits'])}
Modified files: {len(changes['modified_files'])}
New files: {len(changes['new_files'])}
Deleted files: {len(changes['deleted_files'])}
Recent commit messages:
{self._format_commit_messages(changes['new_commits'][:5])}
Files modified:
{self._format_file_list(changes['modified_files'][:10])}
Generate a concise summary of these changes. Focus on what functionality
was added, modified, or removed. Explain the significance of the changes.
"""
summary = self.llm.generate(prompt, max_tokens=300)
return summary
The change summary provides users with a quick understanding of repository updates without requiring them to review individual commits or file diffs.
The system uses an icon or visual indicator to notify users of repository changes. When changes are detected and processed, it displays a notification icon in the user interface. The icon might be a simple symbol like an asterisk, a bell, or a dot that appears next to the repository name. Clicking the icon displays the detailed change summary.
SUPPORTING INTERACTIVE QUESTION ANSWERING
Beyond generating reports and tracking changes, the agentic AI system enables users to ask specific questions about repositories. This interactive capability leverages the hybrid RAG and GraphRAG system to provide accurate, contextually relevant answers.
The question-answering pipeline consists of several stages. First, the system analyzes the user's question to understand the intent and identify key entities. Second, it retrieves relevant information from the knowledge base using hybrid retrieval. Third, it constructs a prompt for the language model that includes the question and retrieved context. Fourth, the language model generates an answer. Fifth, the system formats and presents the answer to the user.
Question analysis involves identifying the type of question being asked. Some questions ask about specific code entities like functions or classes. Others ask about architectural patterns or design decisions. Some questions ask about recent changes or development activity. The system categorizes questions to guide the retrieval strategy.
Here is an example of question classification:
def classify_question(self, question):
classification_prompt = f"""
Classify the following question about a code repository into one of
these categories:
- CODE_ENTITY: Questions about specific functions, classes, or files
- ARCHITECTURE: Questions about overall design and structure
- FUNCTIONALITY: Questions about what the code does
- CHANGES: Questions about recent modifications
- ISSUES: Questions about bugs or open issues
- DEPENDENCIES: Questions about libraries and frameworks used
- GENERAL: Other questions
Question: {question}
Category:
"""
category = self.llm.generate(
classification_prompt,
max_tokens=20
).strip()
return category
This classification helps the system choose appropriate retrieval strategies. For code entity questions, it emphasizes graph traversal to find related entities. For functionality questions, it relies more heavily on semantic similarity. For change-related questions, it focuses on recent commits and modified files.
The retrieval stage uses the hybrid approach described earlier. It performs vector similarity search to find semantically relevant chunks, then expands the context by traversing the graph. The system also applies filters based on the question type. For example, if the question asks about recent changes, it filters to only include chunks from recently modified files.
Context construction is critical for generating accurate answers. The system must provide enough information for the language model to answer the question without overwhelming it with irrelevant details. It ranks retrieved chunks by relevance and includes the top-k chunks in the prompt. It also includes metadata like file paths, function names, and line numbers to help the model provide specific references.
Let us implement an example of answer generation:
def answer_question(self, repo_id, question):
# Classify the question
question_type = self.classify_question(question)
# Retrieve relevant context
context_chunks = self.hybrid_retrieve(
repo_id,
question,
question_type=question_type,
top_k=10
)
# Build the prompt
context_text = self._format_context_chunks(context_chunks)
prompt = f"""
You are analyzing a code repository. Answer the following question
based on the provided context from the repository.
Question: {question}
Context from repository:
{context_text}
Provide a detailed answer based on the context. Include specific
references to files, functions, or classes when relevant. If the
context does not contain enough information to answer the question,
say so clearly.
Answer:
"""
answer = self.llm.generate(prompt, max_tokens=500)
# Extract references from context chunks
references = self._extract_references(context_chunks)
return {
'answer': answer,
'references': references
}
This function orchestrates the question-answering process. It retrieves relevant context, constructs a prompt, generates an answer, and includes references to source locations.
The system also supports follow-up questions that reference previous conversation context. It maintains a conversation history and includes relevant prior exchanges in the prompt. This enables multi-turn conversations where users can ask clarifying questions or dive deeper into specific topics.
For questions about code functionality, the system can optionally include actual code snippets in its answers. It extracts the relevant code from the repository and formats it for display. This allows users to see the implementation alongside the explanation.
PERSISTING AND MANAGING REPOSITORY DATA
Efficient data persistence is essential for the system's performance and usability. The storage layer must save all knowledge base components to disk in a format that enables quick loading and incremental updates. Each analyzed repository gets its own directory in the storage location.
The directory structure for a stored repository looks like this:
storage/
repo_12345/
metadata.json
vector_db/
index.faiss
chunks.json
graph_db/
nodes.json
edges.json
state.json
changes_log.json
The metadata file contains basic information about the repository including its URL or path, the last analysis timestamp, the current commit hash for each branch, and configuration settings. The vector database directory contains the FAISS index file and a JSON file mapping chunk IDs to their text content and metadata. The graph database directory contains JSON files storing nodes and edges. The state file records the repository state at the last analysis. The changes log maintains a history of detected changes.
Find an example of saving repository data below:
def save_repository_data(self, repo_id, repo_data):
repo_dir = self.storage_path / repo_id
repo_dir.mkdir(exist_ok=True)
# Save metadata
metadata = {
'repo_id': repo_id,
'location': repo_data['location'],
'last_analyzed': datetime.now().isoformat(),
'branches': repo_data['branches'],
'primary_language': repo_data['primary_language']
}
with open(repo_dir / 'metadata.json', 'w') as f:
json.dump(metadata, f, indent=2)
# Save vector database
vector_db_dir = repo_dir / 'vector_db'
vector_db_dir.mkdir(exist_ok=True)
self.vector_db.save(vector_db_dir / 'index.faiss')
chunks_data = self.vector_db.get_all_chunks(repo_id)
with open(vector_db_dir / 'chunks.json', 'w') as f:
json.dump(chunks_data, f, indent=2)
# Save graph database
graph_db_dir = repo_dir / 'graph_db'
graph_db_dir.mkdir(exist_ok=True)
nodes = self.graph_db.get_all_nodes(repo_id)
with open(graph_db_dir / 'nodes.json', 'w') as f:
json.dump(nodes, f, indent=2)
edges = self.graph_db.get_all_edges(repo_id)
with open(graph_db_dir / 'edges.json', 'w') as f:
json.dump(edges, f, indent=2)
# Save current state
state = {
'branches': {
branch: self.connector.get_branch_head(branch)
for branch in repo_data['branches']
},
'last_commit_date': repo_data['last_commit_date']
}
with open(repo_dir / 'state.json', 'w') as f:
json.dump(state, f, indent=2)
This function saves all components of the repository knowledge base to disk. It creates the directory structure and writes each component to its appropriate location.
Loading previously analyzed repositories is straightforward. The system checks if a directory exists for the repository ID, loads the metadata to verify it matches the requested repository, loads the vector database index and chunks, loads the graph database nodes and edges, and loads the state information.
An example of loading repository data:
def load_repository_data(self, repo_id):
repo_dir = self.storage_path / repo_id
if not repo_dir.exists():
return None
# Load metadata
with open(repo_dir / 'metadata.json', 'r') as f:
metadata = json.load(f)
# Load vector database
vector_db_dir = repo_dir / 'vector_db'
self.vector_db.load(vector_db_dir / 'index.faiss')
with open(vector_db_dir / 'chunks.json', 'r') as f:
chunks_data = json.load(f)
self.vector_db.load_chunks(repo_id, chunks_data)
# Load graph database
graph_db_dir = repo_dir / 'graph_db'
with open(graph_db_dir / 'nodes.json', 'r') as f:
nodes = json.load(f)
self.graph_db.load_nodes(repo_id, nodes)
with open(graph_db_dir / 'edges.json', 'r') as f:
edges = json.load(f)
self.graph_db.load_edges(repo_id, edges)
# Load state
with open(repo_dir / 'state.json', 'r') as f:
state = json.load(f)
return {
'metadata': metadata,
'state': state
}
When a user requests analysis of a repository that has been previously analyzed, the system loads the stored data and then checks for changes. If changes are detected, it performs an incremental update. This approach is much faster than reanalyzing the entire repository from scratch.
The storage system also implements cleanup mechanisms to manage disk space. Users can configure retention policies that automatically delete data for repositories that have not been accessed recently. The system can also compress older data to reduce storage requirements.
INTEGRATING MULTIPLE LLM BACKENDS
Supporting multiple LLM backends provides flexibility for different deployment scenarios. The system abstracts LLM interactions behind a unified interface that supports both local and remote models across various GPU architectures.
The LLM interface defines standard methods for text generation, embedding generation, and model management. Concrete implementations handle the specifics of different backends.
Base LLM interface:
class LLMInterface:
def generate(self, prompt, max_tokens=500, temperature=0.7):
raise NotImplementedError
def embed(self, text):
raise NotImplementedError
def get_model_info(self):
raise NotImplementedError
For local models, the system supports multiple inference frameworks. The llama.cpp backend provides efficient CPU and GPU inference for GGUF format models. It supports CUDA for Nvidia GPUs, Metal for Apple Silicon, ROCm for AMD GPUs, and optimized CPU inference for Intel processors.
Here is an example llama.cpp backend implementation:
class LlamaCppBackend(LLMInterface):
def __init__(self, model_path, n_gpu_layers=0, n_ctx=4096):
from llama_cpp import Llama
self.model = Llama(
model_path=model_path,
n_gpu_layers=n_gpu_layers,
n_ctx=n_ctx,
verbose=False
)
def generate(self, prompt, max_tokens=500, temperature=0.7):
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
stop=["</s>", "Human:", "User:"]
)
return response['choices'][0]['text']
def embed(self, text):
return self.model.embed(text)
The vLLM backend provides optimized inference for larger models with advanced features like paged attention. It supports CUDA and ROCm GPUs.
class VLLMBackend(LLMInterface):
def __init__(self, model_name, tensor_parallel_size=1):
from vllm import LLM, SamplingParams
self.model = LLM(
model=model_name,
tensor_parallel_size=tensor_parallel_size
)
self.sampling_params = SamplingParams(
temperature=0.7,
max_tokens=500
)
def generate(self, prompt, max_tokens=500, temperature=0.7):
self.sampling_params.max_tokens = max_tokens
self.sampling_params.temperature = temperature
outputs = self.model.generate([prompt], self.sampling_params)
return outputs[0].outputs[0].text
For remote API-based models, the system implements backends for OpenAI, Anthropic, and other providers.
class OpenAIBackend(LLMInterface):
def __init__(self, api_key, model_name="gpt-4"):
import openai
self.client = openai.OpenAI(api_key=api_key)
self.model_name = model_name
def generate(self, prompt, max_tokens=500, temperature=0.7):
response = self.client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
temperature=temperature
)
return response.choices[0].message.content
The system automatically detects available GPU hardware and selects appropriate backends. It checks for CUDA availability for Nvidia GPUs, ROCm for AMD GPUs, Metal Performance Shaders for Apple Silicon, and oneAPI for Intel GPUs.
def detect_gpu_backend():
if torch.cuda.is_available():
return 'cuda'
elif torch.backends.mps.is_available():
return 'mps'
elif hasattr(torch.version, 'hip') and torch.version.hip:
return 'rocm'
else:
return 'cpu'
For embedding models, the system uses sentence-transformers or specialized code embedding models. These models run locally and support the same GPU backends as the main language models.
class EmbeddingModel:
def __init__(self, model_name='sentence-transformers/all-MiniLM-L6-v2'):
from sentence_transformers import SentenceTransformer
device = self._get_device()
self.model = SentenceTransformer(model_name, device=device)
def _get_device(self):
backend = detect_gpu_backend()
if backend == 'cuda':
return 'cuda'
elif backend == 'mps':
return 'mps'
else:
return 'cpu'
def embed(self, text):
return self.model.encode(text, convert_to_tensor=True)
The multi-backend architecture allows users to choose the most appropriate model for their needs. They might use a small local model for quick queries and change detection, while using a larger remote model for complex analysis and report generation. The system configuration file specifies which backend to use for each task.
HANDLING DIFFERENT REPOSITORY TYPES AND LANGUAGES
The agentic AI system must handle repositories written in various programming languages and using different frameworks. The analysis pipeline adapts to the specific characteristics of each language.
For syntax-aware chunking and graph construction, the system uses tree-sitter parsers. Tree-sitter provides parsers for dozens of programming languages including Python, JavaScript, TypeScript, Java, C, C++, Go, Rust, Ruby, and many others. These parsers generate concrete syntax trees that the system traverses to identify code structures.
Find an example of language-specific analysis here:
def analyze_file_by_language(self, file_path, content):
language = self._detect_language(file_path)
if language == 'python':
return self._analyze_python_file(file_path, content)
elif language == 'javascript' or language == 'typescript':
return self._analyze_js_file(file_path, content)
elif language == 'java':
return self._analyze_java_file(file_path, content)
else:
return self._analyze_generic_file(file_path, content)
def _analyze_python_file(self, file_path, content):
tree = self.python_parser.parse(bytes(content, 'utf8'))
nodes = []
edges = []
# Extract classes
for node in tree.root_node.children:
if node.type == 'class_definition':
class_info = self._extract_python_class(node, content)
nodes.append(class_info['node'])
edges.extend(class_info['edges'])
# Extract functions
for node in tree.root_node.children:
if node.type == 'function_definition':
func_info = self._extract_python_function(node, content)
nodes.append(func_info['node'])
edges.extend(func_info['edges'])
# Extract imports
imports = self._extract_python_imports(tree.root_node, content)
edges.extend(imports)
return {'nodes': nodes, 'edges': edges}
Different languages have different structural patterns. Object-oriented languages like Java and Python emphasize classes and inheritance. Functional languages like Haskell or Clojure emphasize functions and higher-order patterns. The analysis pipeline adapts to these differences.
For languages without tree-sitter support, the system falls back to regex-based parsing or line-based chunking. While less precise, these approaches still provide useful information for the knowledge base.
The system also handles language-specific dependency management. It recognizes package manifest files for different ecosystems: package.json for JavaScript/TypeScript, requirements.txt or pyproject.toml for Python, pom.xml or build.gradle for Java, Cargo.toml for Rust, and go.mod for Go. It extracts dependency information from these files and adds it to the knowledge graph.
Configuration files receive special handling. The system recognizes common configuration formats like JSON, YAML, TOML, and INI. It parses these files to extract configuration parameters and adds them as nodes in the knowledge graph. This allows users to ask questions about configuration settings.
Documentation files in Markdown, reStructuredText, or plain text format are chunked and embedded. The system preserves document structure by keeping sections together when possible. It also extracts links between documentation files and code files, adding these relationships to the graph.
IMPLEMENTING THE CHANGE NOTIFICATION SYSTEM
The change notification system alerts users when repositories they are tracking have been updated. It uses visual indicators and generates summaries of changes to keep users informed without overwhelming them with details.
The notification mechanism operates at multiple levels. At the highest level, a simple icon or indicator shows that changes have occurred. Clicking this indicator reveals a summary of changes. Users can then drill down into specific changes for more details.
The icon representation uses a simple character-based symbol that appears next to the repository name in the user interface. The system might use an asterisk, a dot, or a special Unicode character to indicate pending changes.
def get_repository_status_icon(self, repo_id):
changes = self.get_pending_changes(repo_id)
if not changes:
return ''
change_count = (
len(changes.get('new_commits', [])) +
len(changes.get('new_issues', [])) +
len(changes.get('new_prs', []))
)
if change_count == 0:
return ''
elif change_count < 5:
return '●' # Single dot for few changes
else:
return '●●' # Double dot for many changes
This function returns an icon based on the number of pending changes. The icon provides a quick visual indication without requiring users to read text.
When users click the notification icon, the system displays a change summary. This summary includes the number of new commits, a list of modified files, descriptions of significant changes, new or updated issues, and new pull requests. The language model generates natural language descriptions of the changes based on commit messages and file diffs.
def generate_change_notification(self, repo_id, changes):
summary_parts = []
# Summarize commits
if changes['new_commits']:
commit_summary = f"There are {len(changes['new_commits'])} new commits."
summary_parts.append(commit_summary)
# Highlight significant commits
significant = self._identify_significant_commits(changes['new_commits'])
if significant:
summary_parts.append("Notable changes include:")
for commit in significant[:3]:
summary_parts.append(f" - {commit['message']}")
# Summarize file changes
if changes['modified_files']:
file_summary = f"{len(changes['modified_files'])} files were modified."
summary_parts.append(file_summary)
# Identify frequently modified files
frequent = self._identify_frequently_modified(changes['modified_files'])
if frequent:
summary_parts.append("Most active files:")
for file_path in frequent[:5]:
summary_parts.append(f" - {file_path}")
# Summarize new issues
if changes['new_issues']:
issue_summary = f"{len(changes['new_issues'])} new issues were opened."
summary_parts.append(issue_summary)
return '\n'.join(summary_parts)
This function creates a textual summary of repository changes. It highlights the most significant changes and provides enough detail for users to understand what happened without overwhelming them.
The notification system also supports filtering and prioritization. Users can configure which types of changes trigger notifications. For example, they might want notifications for all commits but only for high-priority issues. The system respects these preferences when generating notifications.
For repositories under active development, the system batches changes to avoid notification fatigue. Instead of notifying on every commit, it accumulates changes over a time window and sends a single notification summarizing all changes in that period.
OPTIMIZING PERFORMANCE AND SCALABILITY
Performance optimization is critical for a system that processes large repositories and responds to user queries in real time. Several strategies improve performance and scalability.
Chunking strategies significantly impact both indexing speed and retrieval quality. Smaller chunks provide more precise retrieval but increase the total number of embeddings to process and store. Larger chunks reduce the number of embeddings but may include irrelevant content. The system uses adaptive chunking that adjusts chunk size based on content type. Code chunks align with logical units like functions and classes. Documentation chunks align with sections and paragraphs.
The embedding process can be parallelized across multiple CPU cores or GPU devices. The system processes multiple chunks simultaneously to maximize throughput. For very large repositories, it implements batch processing with progress tracking.
def embed_chunks_parallel(self, chunks, batch_size=32):
embeddings = []
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i+batch_size]
batch_texts = [chunk['text'] for chunk in batch]
batch_embeddings = self.embedding_model.encode(
batch_texts,
batch_size=batch_size,
show_progress_bar=False
)
embeddings.extend(batch_embeddings)
return embeddings
This function processes chunks in batches to maximize GPU utilization and minimize overhead.
The vector database index structure affects query performance. FAISS supports multiple index types with different tradeoffs. Flat indexes provide exact nearest neighbor search but scale poorly to large datasets. IVF indexes partition the vector space and provide approximate search with much better scalability. HNSW indexes provide excellent query performance with moderate memory overhead.
For repositories with hundreds of thousands of chunks, the system uses an IVF index with product quantization to reduce memory usage while maintaining good recall.
def build_optimized_index(self, embeddings, dimension=384):
import faiss
n_vectors = len(embeddings)
if n_vectors < 10000:
# Use flat index for small datasets
index = faiss.IndexFlatL2(dimension)
else:
# Use IVF with PQ for large datasets
n_centroids = min(int(n_vectors / 100), 4096)
n_subquantizers = 8
n_bits = 8
quantizer = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFPQ(
quantizer,
dimension,
n_centroids,
n_subquantizers,
n_bits
)
# Train the index
training_sample = embeddings[:min(100000, n_vectors)]
index.train(training_sample)
index.add(embeddings)
return index
This function selects an appropriate index type based on the dataset size and trains the index if necessary.
Graph database queries can become expensive for large graphs. The system optimizes graph traversal by limiting traversal depth, caching frequently accessed nodes, and using indexed lookups for common query patterns. It also prunes the graph to remove low-value edges that do not contribute to understanding.
Incremental updates are much faster than full reanalysis. The system tracks which files have changed and only reprocesses those files. It also implements lazy loading, where it only loads parts of the knowledge base needed to answer a specific query.
For very large repositories, the system can operate in a streaming mode where it processes files incrementally and updates the knowledge base progressively. This allows users to start querying the repository before the complete analysis finishes.
ENSURING ROBUSTNESS AND ERROR HANDLING
A production-ready system must handle errors gracefully and recover from failures. The repository analysis system implements comprehensive error handling at every layer.
Network errors can occur when accessing remote repositories. The system implements retry logic with exponential backoff for transient failures. It also caches repository data locally to enable offline operation when the remote repository is temporarily unavailable.
def fetch_with_retry(self, url, max_retries=3):
for attempt in range(max_retries):
try:
response = requests.get(url, timeout=30)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt
time.sleep(wait_time)
This function implements retry logic for network requests with exponential backoff.
Parsing errors can occur when processing malformed source files or files with syntax errors. The system catches parsing exceptions and logs them without failing the entire analysis. It processes as much of the repository as possible even when some files cannot be parsed.
def safe_parse_file(self, file_path, content):
try:
return self.parse_file(file_path, content)
except Exception as e:
self.logger.warning(
f"Failed to parse {file_path}: {str(e)}"
)
return None
This wrapper function catches parsing errors and logs them without propagating the exception.
Memory errors can occur when processing very large files or repositories. The system monitors memory usage and implements streaming processing for large files. It also sets limits on file sizes and skips files that exceed the limits.
LLM API errors require special handling. API rate limits, quota exhaustion, and service outages can all cause failures. The system implements rate limiting on the client side to avoid exceeding API limits. It also provides fallback mechanisms, such as using a simpler analysis when the LLM is unavailable.
def generate_with_fallback(self, prompt, max_tokens=500):
try:
return self.llm.generate(prompt, max_tokens=max_tokens)
except Exception as e:
self.logger.error(f"LLM generation failed: {str(e)}")
# Fallback to template-based response
return self.generate_template_response(prompt)
This function provides a fallback mechanism when LLM generation fails.
Data consistency is critical when updating the knowledge base. The system uses atomic operations when possible and implements transaction-like semantics for updates. If an update fails partway through, it rolls back to the previous consistent state.
The system also implements health checks that verify the integrity of stored data. It checks that vector indices are valid, graph databases are consistent, and metadata files are not corrupted. If corruption is detected, it can rebuild the knowledge base from the source repository.
SECURITY AND PRIVACY CONSIDERATIONS
Security and privacy are important considerations for a system that accesses code repositories and potentially sensitive information. The system implements several security measures.
Authentication credentials for accessing private repositories must be stored securely. The system uses encrypted storage for API tokens and passwords. It supports credential management systems like system keychains or secret management services.
def store_credentials(self, repo_id, credentials):
import keyring
keyring.set_password(
'repo_analyzer',
repo_id,
json.dumps(credentials)
)
def retrieve_credentials(self, repo_id):
import keyring
creds_json = keyring.get_password('repo_analyzer', repo_id)
if creds_json:
return json.loads(creds_json)
return None
This code uses the system keyring to store credentials securely.
When using remote LLM APIs, the system must be careful not to send sensitive code to external services without user consent. It implements a privacy mode where sensitive repositories are only analyzed using local models. Users can mark repositories as sensitive, and the system enforces this restriction.
The system also implements access controls for the stored knowledge base. If multiple users share a system, each user should only access repositories they have permission to view. The system can integrate with existing authentication systems to enforce these controls.
Data retention policies help manage privacy and compliance requirements. Users can configure how long the system retains data for inactive repositories. The system can automatically delete old data or archive it to secure storage.
When generating embeddings and storing them locally, the system ensures that the stored data does not leak sensitive information. Embeddings are abstract vector representations and do not contain the original text, but they can still reveal information through similarity searches. The system provides options to encrypt stored embeddings for additional security.
EXTENDING THE SYSTEM WITH PLUGINS
The architecture supports extensibility through a plugin system. Users can add custom analyzers, custom graph node types, custom retrieval strategies, and custom report generators.
The plugin interface defines hooks that the system calls during various stages of processing. Plugins register themselves with the system and implement the required interface methods.
class AnalyzerPlugin:
def get_name(self):
raise NotImplementedError
def can_handle_file(self, file_path):
raise NotImplementedError
def analyze_file(self, file_path, content):
raise NotImplementedError
This base class defines the interface for analyzer plugins. Concrete plugins implement these methods to provide custom analysis for specific file types.
For example, a plugin might analyze Dockerfile files to extract information about container images, dependencies, and build steps:
class DockerfileAnalyzer(AnalyzerPlugin):
def get_name(self):
return 'dockerfile_analyzer'
def can_handle_file(self, file_path):
return file_path.endswith('Dockerfile') or \
'Dockerfile' in os.path.basename(file_path)
def analyze_file(self, file_path, content):
nodes = []
edges = []
# Parse Dockerfile
lines = content.split('\n')
base_image = None
for line in lines:
line = line.strip()
if line.startswith('FROM'):
base_image = line.split()[1]
nodes.append({
'id': f"{file_path}::base_image",
'type': 'docker_base_image',
'name': base_image,
'file': file_path
})
return {'nodes': nodes, 'edges': edges}
This plugin analyzes Dockerfile files and extracts information about base images. The system automatically uses this plugin when it encounters Dockerfile files.
Plugins can also extend the report generation process. A custom report plugin might generate specialized sections for specific types of repositories, such as web applications or machine learning projects.
The plugin system enables the community to extend the system's capabilities without modifying the core codebase. Users can share plugins to support new languages, frameworks, or analysis techniques.
COMPLETE RUNNING EXAMPLE IMPLEMENTATION
The following section presents a complete, production-ready implementation of the agentic AI repository analysis system. This implementation includes all the components discussed throughout the article and provides a fully functional system that can analyze repositories, track changes, and answer questions.
#!/usr/bin/env python3
"""
Repository Analysis Agentic AI System
A comprehensive system for analyzing GitHub and GitLab repositories using
hybrid RAG and GraphRAG with bidirectional relationships. Supports continuous
change tracking and interactive question answering.
"""
import os
import sys
import json
import time
import logging
import hashlib
import requests
import threading
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, asdict
from collections import defaultdict
# Third-party imports
import numpy as np
import torch
import faiss
from sentence_transformers import SentenceTransformer
import git
from tree_sitter import Language, Parser
import tree_sitter_python
import tree_sitter_javascript
import tree_sitter_java
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# ============================================================================
# Data Models
# ============================================================================
@dataclass
class Chunk:
"""Represents a text chunk with metadata"""
id: str
text: str
file_path: str
start_line: int
end_line: int
chunk_type: str
node_id: Optional[str] = None
@dataclass
class GraphNode:
"""Represents a node in the knowledge graph"""
id: str
type: str
name: str
file_path: str
line_number: int
metadata: Dict[str, Any]
embedding_id: Optional[str] = None
@dataclass
class GraphEdge:
"""Represents an edge in the knowledge graph"""
from_node: str
to_node: str
edge_type: str
metadata: Dict[str, Any]
@dataclass
class RepositoryChange:
"""Represents changes detected in a repository"""
new_commits: List[Dict]
modified_files: List[str]
new_files: List[str]
deleted_files: List[str]
new_issues: List[Dict]
updated_issues: List[Dict]
# ============================================================================
# GPU Backend Detection
# ============================================================================
def detect_gpu_backend() -> str:
"""Detect available GPU backend"""
if torch.cuda.is_available():
logger.info("Detected CUDA GPU backend")
return 'cuda'
elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
logger.info("Detected Apple MPS backend")
return 'mps'
elif hasattr(torch.version, 'hip') and torch.version.hip:
logger.info("Detected AMD ROCm backend")
return 'rocm'
else:
logger.info("Using CPU backend")
return 'cpu'
# ============================================================================
# LLM Interface and Implementations
# ============================================================================
class LLMInterface:
"""Base interface for language models"""
def generate(self, prompt: str, max_tokens: int = 500,
temperature: float = 0.7) -> str:
"""Generate text from prompt"""
raise NotImplementedError
def embed(self, text: str) -> np.ndarray:
"""Generate embedding for text"""
raise NotImplementedError
class LocalLLM(LLMInterface):
"""Local LLM implementation using llama.cpp"""
def __init__(self, model_path: str, n_gpu_layers: int = 0):
"""Initialize local LLM
Args:
model_path: Path to GGUF model file
n_gpu_layers: Number of layers to offload to GPU
"""
try:
from llama_cpp import Llama
self.model = Llama(
model_path=model_path,
n_gpu_layers=n_gpu_layers,
n_ctx=4096,
verbose=False
)
logger.info(f"Loaded local LLM from {model_path}")
except ImportError:
logger.error("llama-cpp-python not installed")
raise
def generate(self, prompt: str, max_tokens: int = 500,
temperature: float = 0.7) -> str:
"""Generate text from prompt"""
try:
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
stop=["</s>", "Human:", "User:"]
)
return response['choices'][0]['text'].strip()
except Exception as e:
logger.error(f"Generation failed: {e}")
return ""
def embed(self, text: str) -> np.ndarray:
"""Generate embedding for text"""
return np.array(self.model.embed(text))
class OpenAILLM(LLMInterface):
"""OpenAI API implementation"""
def __init__(self, api_key: str, model_name: str = "gpt-4"):
"""Initialize OpenAI LLM
Args:
api_key: OpenAI API key
model_name: Model name to use
"""
try:
import openai
self.client = openai.OpenAI(api_key=api_key)
self.model_name = model_name
logger.info(f"Initialized OpenAI LLM with model {model_name}")
except ImportError:
logger.error("openai package not installed")
raise
def generate(self, prompt: str, max_tokens: int = 500,
temperature: float = 0.7) -> str:
"""Generate text from prompt"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
temperature=temperature
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"OpenAI generation failed: {e}")
return ""
def embed(self, text: str) -> np.ndarray:
"""Generate embedding for text"""
try:
response = self.client.embeddings.create(
model="text-embedding-ada-002",
input=text
)
return np.array(response.data[0].embedding)
except Exception as e:
logger.error(f"OpenAI embedding failed: {e}")
return np.zeros(1536)
# ============================================================================
# Embedding Model
# ============================================================================
class EmbeddingModel:
"""Handles text embedding generation"""
def __init__(self, model_name: str = 'sentence-transformers/all-MiniLM-L6-v2'):
"""Initialize embedding model
Args:
model_name: Name of sentence transformer model
"""
device = self._get_device()
self.model = SentenceTransformer(model_name, device=device)
self.dimension = self.model.get_sentence_embedding_dimension()
logger.info(f"Loaded embedding model {model_name} on {device}")
def _get_device(self) -> str:
"""Get appropriate device for model"""
backend = detect_gpu_backend()
if backend == 'cuda':
return 'cuda'
elif backend == 'mps':
return 'mps'
else:
return 'cpu'
def embed(self, text: str) -> np.ndarray:
"""Generate embedding for single text"""
return self.model.encode(text, convert_to_numpy=True)
def embed_batch(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
"""Generate embeddings for batch of texts"""
return self.model.encode(
texts,
batch_size=batch_size,
show_progress_bar=False,
convert_to_numpy=True
)
# ============================================================================
# Repository Connector
# ============================================================================
class RepositoryConnector:
"""Handles connections to local and remote repositories"""
def __init__(self, location: str, credentials: Optional[Dict] = None):
"""Initialize repository connector
Args:
location: Local path or remote URL
credentials: Optional authentication credentials
"""
self.location = location
self.credentials = credentials or {}
self.is_remote = self._determine_if_remote()
self.repo_type = self._determine_repo_type()
self.repo = None
if not self.is_remote:
self._connect_local()
def _determine_if_remote(self) -> bool:
"""Check if location is remote"""
return (self.location.startswith('http://') or
self.location.startswith('https://') or
self.location.startswith('git@'))
def _determine_repo_type(self) -> str:
"""Determine repository hosting type"""
if 'github.com' in self.location:
return 'github'
elif 'gitlab' in self.location:
return 'gitlab'
else:
return 'local'
def _connect_local(self):
"""Connect to local Git repository"""
try:
self.repo = git.Repo(self.location)
logger.info(f"Connected to local repository at {self.location}")
except git.exc.InvalidGitRepositoryError:
logger.error(f"Invalid Git repository: {self.location}")
raise
def get_file_content(self, file_path: str, ref: str = 'HEAD') -> str:
"""Get content of file at specific ref
Args:
file_path: Path to file in repository
ref: Git reference (branch, commit, tag)
Returns:
File content as string
"""
if self.is_remote:
return self._fetch_remote_file(file_path, ref)
else:
return self._read_local_file(file_path, ref)
def _read_local_file(self, file_path: str, ref: str) -> str:
"""Read file from local repository"""
try:
commit = self.repo.commit(ref)
blob = commit.tree / file_path
return blob.data_stream.read().decode('utf-8', errors='ignore')
except Exception as e:
logger.warning(f"Failed to read {file_path}: {e}")
return ""
def _fetch_remote_file(self, file_path: str, ref: str) -> str:
"""Fetch file from remote repository"""
if self.repo_type == 'github':
return self._fetch_github_file(file_path, ref)
elif self.repo_type == 'gitlab':
return self._fetch_gitlab_file(file_path, ref)
return ""
def _fetch_github_file(self, file_path: str, ref: str) -> str:
"""Fetch file from GitHub"""
parts = self.location.rstrip('.git').split('/')
owner, repo = parts[-2], parts[-1]
url = f"https://api.github.com/repos/{owner}/{repo}/contents/{file_path}"
params = {'ref': ref}
headers = {}
if 'token' in self.credentials:
headers['Authorization'] = f"token {self.credentials['token']}"
try:
response = requests.get(url, params=params, headers=headers, timeout=30)
response.raise_for_status()
import base64
content = response.json()['content']
return base64.b64decode(content).decode('utf-8', errors='ignore')
except Exception as e:
logger.warning(f"Failed to fetch {file_path} from GitHub: {e}")
return ""
def _fetch_gitlab_file(self, file_path: str, ref: str) -> str:
"""Fetch file from GitLab"""
# Extract project path from URL
parts = self.location.rstrip('.git').split('/')
project_path = '/'.join(parts[-2:])
# URL encode the file path
import urllib.parse
encoded_path = urllib.parse.quote(file_path, safe='')
encoded_project = urllib.parse.quote(project_path, safe='')
url = f"https://gitlab.com/api/v4/projects/{encoded_project}/repository/files/{encoded_path}/raw"
params = {'ref': ref}
headers = {}
if 'token' in self.credentials:
headers['PRIVATE-TOKEN'] = self.credentials['token']
try:
response = requests.get(url, params=params, headers=headers, timeout=30)
response.raise_for_status()
return response.text
except Exception as e:
logger.warning(f"Failed to fetch {file_path} from GitLab: {e}")
return ""
def list_files(self, ref: str = 'HEAD') -> List[str]:
"""List all files in repository
Args:
ref: Git reference
Returns:
List of file paths
"""
if self.is_remote:
return self._list_remote_files(ref)
else:
return self._list_local_files(ref)
def _list_local_files(self, ref: str) -> List[str]:
"""List files in local repository"""
try:
commit = self.repo.commit(ref)
return [item.path for item in commit.tree.traverse()
if item.type == 'blob']
except Exception as e:
logger.error(f"Failed to list files: {e}")
return []
def _list_remote_files(self, ref: str) -> List[str]:
"""List files in remote repository"""
if self.repo_type == 'github':
return self._list_github_files(ref)
elif self.repo_type == 'gitlab':
return self._list_gitlab_files(ref)
return []
def _list_github_files(self, ref: str) -> List[str]:
"""List files in GitHub repository"""
parts = self.location.rstrip('.git').split('/')
owner, repo = parts[-2], parts[-1]
url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{ref}"
params = {'recursive': '1'}
headers = {}
if 'token' in self.credentials:
headers['Authorization'] = f"token {self.credentials['token']}"
try:
response = requests.get(url, params=params, headers=headers, timeout=30)
response.raise_for_status()
tree = response.json()['tree']
return [item['path'] for item in tree if item['type'] == 'blob']
except Exception as e:
logger.error(f"Failed to list GitHub files: {e}")
return []
def _list_gitlab_files(self, ref: str) -> List[str]:
"""List files in GitLab repository"""
parts = self.location.rstrip('.git').split('/')
project_path = '/'.join(parts[-2:])
import urllib.parse
encoded_project = urllib.parse.quote(project_path, safe='')
url = f"https://gitlab.com/api/v4/projects/{encoded_project}/repository/tree"
params = {'ref': ref, 'recursive': 'true', 'per_page': 100}
headers = {}
if 'token' in self.credentials:
headers['PRIVATE-TOKEN'] = self.credentials['token']
files = []
page = 1
try:
while True:
params['page'] = page
response = requests.get(url, params=params, headers=headers, timeout=30)
response.raise_for_status()
items = response.json()
if not items:
break
files.extend([item['path'] for item in items if item['type'] == 'blob'])
page += 1
if len(items) < 100:
break
return files
except Exception as e:
logger.error(f"Failed to list GitLab files: {e}")
return []
def get_current_state(self) -> Dict:
"""Get current repository state
Returns:
Dictionary with branch heads and metadata
"""
if self.is_remote:
return self._get_remote_state()
else:
return self._get_local_state()
def _get_local_state(self) -> Dict:
"""Get local repository state"""
branches = {}
for ref in self.repo.references:
if isinstance(ref, git.Head):
branches[ref.name] = ref.commit.hexsha
return {
'branches': branches,
'last_commit_date': self.repo.head.commit.committed_datetime.isoformat()
}
def _get_remote_state(self) -> Dict:
"""Get remote repository state"""
if self.repo_type == 'github':
return self._get_github_state()
elif self.repo_type == 'gitlab':
return self._get_gitlab_state()
return {'branches': {}, 'last_commit_date': None}
def _get_github_state(self) -> Dict:
"""Get GitHub repository state"""
parts = self.location.rstrip('.git').split('/')
owner, repo = parts[-2], parts[-1]
url = f"https://api.github.com/repos/{owner}/{repo}/branches"
headers = {}
if 'token' in self.credentials:
headers['Authorization'] = f"token {self.credentials['token']}"
try:
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
branches = {}
for branch in response.json():
branches[branch['name']] = branch['commit']['sha']
# Get latest commit date
commits_url = f"https://api.github.com/repos/{owner}/{repo}/commits"
commits_response = requests.get(commits_url, headers=headers, timeout=30)
commits_response.raise_for_status()
last_commit_date = None
if commits_response.json():
last_commit_date = commits_response.json()[0]['commit']['committer']['date']
return {
'branches': branches,
'last_commit_date': last_commit_date
}
except Exception as e:
logger.error(f"Failed to get GitHub state: {e}")
return {'branches': {}, 'last_commit_date': None}
def _get_gitlab_state(self) -> Dict:
"""Get GitLab repository state"""
parts = self.location.rstrip('.git').split('/')
project_path = '/'.join(parts[-2:])
import urllib.parse
encoded_project = urllib.parse.quote(project_path, safe='')
url = f"https://gitlab.com/api/v4/projects/{encoded_project}/repository/branches"
headers = {}
if 'token' in self.credentials:
headers['PRIVATE-TOKEN'] = self.credentials['token']
try:
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
branches = {}
for branch in response.json():
branches[branch['name']] = branch['commit']['id']
# Get latest commit
commits_url = f"https://gitlab.com/api/v4/projects/{encoded_project}/repository/commits"
commits_response = requests.get(commits_url, headers=headers, timeout=30)
commits_response.raise_for_status()
last_commit_date = None
if commits_response.json():
last_commit_date = commits_response.json()[0]['committed_date']
return {
'branches': branches,
'last_commit_date': last_commit_date
}
except Exception as e:
logger.error(f"Failed to get GitLab state: {e}")
return {'branches': {}, 'last_commit_date': None}
# ============================================================================
# Code Parser and Chunker
# ============================================================================
class CodeParser:
"""Parses source code using tree-sitter"""
def __init__(self):
"""Initialize code parser with language grammars"""
self.parsers = {}
self._initialize_parsers()
def _initialize_parsers(self):
"""Initialize tree-sitter parsers for supported languages"""
try:
# Python parser
python_lang = Language(tree_sitter_python.language())
python_parser = Parser(python_lang)
self.parsers['python'] = python_parser
# JavaScript parser
js_lang = Language(tree_sitter_javascript.language())
js_parser = Parser(js_lang)
self.parsers['javascript'] = js_parser
self.parsers['typescript'] = js_parser
# Java parser
java_lang = Language(tree_sitter_java.language())
java_parser = Parser(java_lang)
self.parsers['java'] = java_parser
logger.info("Initialized tree-sitter parsers")
except Exception as e:
logger.warning(f"Failed to initialize some parsers: {e}")
def detect_language(self, file_path: str) -> Optional[str]:
"""Detect programming language from file extension
Args:
file_path: Path to file
Returns:
Language name or None
"""
ext_map = {
'.py': 'python',
'.js': 'javascript',
'.ts': 'typescript',
'.jsx': 'javascript',
'.tsx': 'typescript',
'.java': 'java'
}
ext = Path(file_path).suffix.lower()
return ext_map.get(ext)
def parse_file(self, file_path: str, content: str) -> Optional[Any]:
"""Parse source file
Args:
file_path: Path to file
content: File content
Returns:
Parse tree or None
"""
language = self.detect_language(file_path)
if not language or language not in self.parsers:
return None
try:
parser = self.parsers[language]
tree = parser.parse(bytes(content, 'utf8'))
return tree
except Exception as e:
logger.warning(f"Failed to parse {file_path}: {e}")
return None
class CodeChunker:
"""Chunks source code into logical units"""
def __init__(self, parser: CodeParser):
"""Initialize chunker
Args:
parser: Code parser instance
"""
self.parser = parser
def chunk_file(self, file_path: str, content: str) -> List[Chunk]:
"""Chunk file into logical units
Args:
file_path: Path to file
content: File content
Returns:
List of chunks
"""
language = self.parser.detect_language(file_path)
if language and language in self.parser.parsers:
return self._chunk_by_syntax(file_path, content, language)
else:
return self._chunk_by_lines(file_path, content)
def _chunk_by_syntax(self, file_path: str, content: str,
language: str) -> List[Chunk]:
"""Chunk code using syntax tree
Args:
file_path: Path to file
content: File content
language: Programming language
Returns:
List of chunks
"""
tree = self.parser.parse_file(file_path, content)
if not tree:
return self._chunk_by_lines(file_path, content)
chunks = []
lines = content.split('\n')
# Define node types to extract based on language
if language == 'python':
target_types = ['function_definition', 'class_definition']
elif language in ['javascript', 'typescript']:
target_types = ['function_declaration', 'class_declaration',
'method_definition', 'arrow_function']
elif language == 'java':
target_types = ['method_declaration', 'class_declaration']
else:
target_types = []
def traverse(node):
if node.type in target_types:
start_line = node.start_point[0]
end_line = node.end_point[0]
chunk_text = '\n'.join(lines[start_line:end_line + 1])
chunk_id = hashlib.md5(
f"{file_path}:{start_line}:{end_line}".encode()
).hexdigest()
chunks.append(Chunk(
id=chunk_id,
text=chunk_text,
file_path=file_path,
start_line=start_line,
end_line=end_line,
chunk_type=node.type
))
for child in node.children:
traverse(child)
traverse(tree.root_node)
# If no chunks found, fall back to line-based chunking
if not chunks:
return self._chunk_by_lines(file_path, content)
return chunks
def _chunk_by_lines(self, file_path: str, content: str,
max_lines: int = 50, overlap: int = 5) -> List[Chunk]:
"""Chunk code by line count with overlap
Args:
file_path: Path to file
content: File content
max_lines: Maximum lines per chunk
overlap: Number of overlapping lines
Returns:
List of chunks
"""
lines = content.split('\n')
chunks = []
start = 0
while start < len(lines):
end = min(start + max_lines, len(lines))
chunk_text = '\n'.join(lines[start:end])
chunk_id = hashlib.md5(
f"{file_path}:{start}:{end}".encode()
).hexdigest()
chunks.append(Chunk(
id=chunk_id,
text=chunk_text,
file_path=file_path,
start_line=start,
end_line=end - 1,
chunk_type='line_chunk'
))
start = end - overlap if end < len(lines) else end
return chunks
# ============================================================================
# Graph Builder
# ============================================================================
class GraphBuilder:
"""Builds knowledge graph from code"""
def __init__(self, parser: CodeParser):
"""Initialize graph builder
Args:
parser: Code parser instance
"""
self.parser = parser
def build_graph(self, file_path: str, content: str) -> Tuple[List[GraphNode], List[GraphEdge]]:
"""Build graph nodes and edges for file
Args:
file_path: Path to file
content: File content
Returns:
Tuple of (nodes, edges)
"""
language = self.parser.detect_language(file_path)
if language == 'python':
return self._build_python_graph(file_path, content)
elif language in ['javascript', 'typescript']:
return self._build_js_graph(file_path, content)
elif language == 'java':
return self._build_java_graph(file_path, content)
else:
return self._build_generic_graph(file_path, content)
def _build_python_graph(self, file_path: str,
content: str) -> Tuple[List[GraphNode], List[GraphEdge]]:
"""Build graph for Python file"""
import ast
nodes = []
edges = []
try:
tree = ast.parse(content)
except SyntaxError as e:
logger.warning(f"Syntax error in {file_path}: {e}")
return nodes, edges
# Extract classes
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
class_id = f"{file_path}::{node.name}"
nodes.append(GraphNode(
id=class_id,
type='class',
name=node.name,
file_path=file_path,
line_number=node.lineno,
metadata={
'docstring': ast.get_docstring(node),
'bases': [base.id if isinstance(base, ast.Name) else str(base)
for base in node.bases]
}
))
# Add inheritance edges
for base in node.bases:
if isinstance(base, ast.Name):
edges.append(GraphEdge(
from_node=class_id,
to_node=base.id,
edge_type='inherits',
metadata={}
))
# Extract functions
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
func_id = f"{file_path}::{node.name}"
nodes.append(GraphNode(
id=func_id,
type='function',
name=node.name,
file_path=file_path,
line_number=node.lineno,
metadata={
'docstring': ast.get_docstring(node),
'args': [arg.arg for arg in node.args.args]
}
))
# Extract function calls
for child in ast.walk(node):
if isinstance(child, ast.Call):
if isinstance(child.func, ast.Name):
edges.append(GraphEdge(
from_node=func_id,
to_node=child.func.id,
edge_type='calls',
metadata={}
))
# Extract imports
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
edges.append(GraphEdge(
from_node=file_path,
to_node=alias.name,
edge_type='imports',
metadata={}
))
elif isinstance(node, ast.ImportFrom):
if node.module:
edges.append(GraphEdge(
from_node=file_path,
to_node=node.module,
edge_type='imports',
metadata={}
))
return nodes, edges
def _build_js_graph(self, file_path: str,
content: str) -> Tuple[List[GraphNode], List[GraphEdge]]:
"""Build graph for JavaScript/TypeScript file"""
tree = self.parser.parse_file(file_path, content)
if not tree:
return [], []
nodes = []
edges = []
lines = content.split('\n')
def traverse(node):
# Extract function declarations
if node.type in ['function_declaration', 'arrow_function', 'function']:
name_node = node.child_by_field_name('name')
name = lines[name_node.start_point[0]][
name_node.start_point[1]:name_node.end_point[1]
] if name_node else 'anonymous'
func_id = f"{file_path}::{name}:{node.start_point[0]}"
nodes.append(GraphNode(
id=func_id,
type='function',
name=name,
file_path=file_path,
line_number=node.start_point[0],
metadata={}
))
# Extract class declarations
elif node.type == 'class_declaration':
name_node = node.child_by_field_name('name')
name = lines[name_node.start_point[0]][
name_node.start_point[1]:name_node.end_point[1]
] if name_node else 'anonymous'
class_id = f"{file_path}::{name}"
nodes.append(GraphNode(
id=class_id,
type='class',
name=name,
file_path=file_path,
line_number=node.start_point[0],
metadata={}
))
for child in node.children:
traverse(child)
traverse(tree.root_node)
return nodes, edges
def _build_java_graph(self, file_path: str,
content: str) -> Tuple[List[GraphNode], List[GraphEdge]]:
"""Build graph for Java file"""
tree = self.parser.parse_file(file_path, content)
if not tree:
return [], []
nodes = []
edges = []
lines = content.split('\n')
def traverse(node):
# Extract class declarations
if node.type == 'class_declaration':
name_node = node.child_by_field_name('name')
name = lines[name_node.start_point[0]][
name_node.start_point[1]:name_node.end_point[1]
] if name_node else 'anonymous'
class_id = f"{file_path}::{name}"
nodes.append(GraphNode(
id=class_id,
type='class',
name=name,
file_path=file_path,
line_number=node.start_point[0],
metadata={}
))
# Extract method declarations
elif node.type == 'method_declaration':
name_node = node.child_by_field_name('name')
name = lines[name_node.start_point[0]][
name_node.start_point[1]:name_node.end_point[1]
] if name_node else 'anonymous'
method_id = f"{file_path}::{name}:{node.start_point[0]}"
nodes.append(GraphNode(
id=method_id,
type='method',
name=name,
file_path=file_path,
line_number=node.start_point[0],
metadata={}
))
for child in node.children:
traverse(child)
traverse(tree.root_node)
return nodes, edges
def _build_generic_graph(self, file_path: str,
content: str) -> Tuple[List[GraphNode], List[GraphEdge]]:
"""Build basic graph for unsupported languages"""
# Create a single node representing the file
file_id = file_path
nodes = [GraphNode(
id=file_id,
type='file',
name=Path(file_path).name,
file_path=file_path,
line_number=0,
metadata={'lines': len(content.split('\n'))}
)]
return nodes, []
# ============================================================================
# Vector Database
# ============================================================================
class VectorDatabase:
"""Manages vector embeddings and similarity search"""
def __init__(self, dimension: int):
"""Initialize vector database
Args:
dimension: Embedding dimension
"""
self.dimension = dimension
self.index = None
self.chunks_by_id = {}
self.id_to_idx = {}
self.idx_to_id = {}
self.next_idx = 0
def add_embeddings(self, repo_id: str, embeddings: np.ndarray,
chunks: List[Chunk]) -> List[str]:
"""Add embeddings to database
Args:
repo_id: Repository identifier
embeddings: Array of embeddings
chunks: Corresponding chunks
Returns:
List of chunk IDs
"""
if self.index is None:
self._initialize_index(len(embeddings))
# Add to FAISS index
self.index.add(embeddings.astype('float32'))
# Store chunk metadata
chunk_ids = []
for i, chunk in enumerate(chunks):
chunk_id = f"{repo_id}::{chunk.id}"
chunk_ids.append(chunk_id)
self.chunks_by_id[chunk_id] = chunk
self.id_to_idx[chunk_id] = self.next_idx + i
self.idx_to_id[self.next_idx + i] = chunk_id
self.next_idx += len(chunks)
return chunk_ids
def _initialize_index(self, n_vectors: int):
"""Initialize FAISS index
Args:
n_vectors: Expected number of vectors
"""
if n_vectors < 10000:
# Use flat index for small datasets
self.index = faiss.IndexFlatL2(self.dimension)
logger.info("Initialized flat FAISS index")
else:
# Use IVF index for large datasets
n_centroids = min(int(n_vectors / 100), 4096)
quantizer = faiss.IndexFlatL2(self.dimension)
self.index = faiss.IndexIVFFlat(
quantizer,
self.dimension,
n_centroids
)
logger.info(f"Initialized IVF FAISS index with {n_centroids} centroids")
def train_index(self, embeddings: np.ndarray):
"""Train IVF index if needed
Args:
embeddings: Training embeddings
"""
if isinstance(self.index, faiss.IndexIVFFlat):
if not self.index.is_trained:
self.index.train(embeddings.astype('float32'))
logger.info("Trained IVF index")
def similarity_search(self, query_embedding: np.ndarray,
k: int = 5) -> List[Chunk]:
"""Search for similar chunks
Args:
query_embedding: Query embedding
k: Number of results
Returns:
List of similar chunks
"""
if self.index is None or self.index.ntotal == 0:
return []
query = query_embedding.reshape(1, -1).astype('float32')
distances, indices = self.index.search(query, k)
results = []
for idx in indices[0]:
if idx >= 0 and idx in self.idx_to_id:
chunk_id = self.idx_to_id[idx]
if chunk_id in self.chunks_by_id:
results.append(self.chunks_by_id[chunk_id])
return results
def get_chunk_by_id(self, chunk_id: str) -> Optional[Chunk]:
"""Get chunk by ID
Args:
chunk_id: Chunk identifier
Returns:
Chunk or None
"""
return self.chunks_by_id.get(chunk_id)
def remove_chunks(self, chunk_ids: List[str]):
"""Remove chunks from database
Args:
chunk_ids: List of chunk IDs to remove
"""
# Note: FAISS doesn't support efficient deletion
# In production, would rebuild index periodically
for chunk_id in chunk_ids:
if chunk_id in self.chunks_by_id:
del self.chunks_by_id[chunk_id]
if chunk_id in self.id_to_idx:
idx = self.id_to_idx[chunk_id]
del self.id_to_idx[chunk_id]
if idx in self.idx_to_id:
del self.idx_to_id[idx]
def save(self, path: Path):
"""Save index to disk
Args:
path: Path to save index
"""
if self.index is not None:
faiss.write_index(self.index, str(path))
def load(self, path: Path):
"""Load index from disk
Args:
path: Path to load index from
"""
if path.exists():
self.index = faiss.read_index(str(path))
# ============================================================================
# Graph Database
# ============================================================================
class GraphDatabase:
"""Manages knowledge graph"""
def __init__(self):
"""Initialize graph database"""
self.nodes = {}
self.edges = defaultdict(list)
self.reverse_edges = defaultdict(list)
def add_node(self, node: GraphNode):
"""Add node to graph
Args:
node: Graph node
"""
self.nodes[node.id] = node
def add_edge(self, edge: GraphEdge):
"""Add edge to graph
Args:
edge: Graph edge
"""
self.edges[edge.from_node].append(edge)
self.reverse_edges[edge.to_node].append(edge)
def get_node(self, node_id: str) -> Optional[GraphNode]:
"""Get node by ID
Args:
node_id: Node identifier
Returns:
Graph node or None
"""
return self.nodes.get(node_id)
def get_neighbors(self, node_id: str, max_depth: int = 1,
edge_types: Optional[List[str]] = None) -> List[GraphNode]:
"""Get neighboring nodes
Args:
node_id: Starting node ID
max_depth: Maximum traversal depth
edge_types: Filter by edge types
Returns:
List of neighboring nodes
"""
visited = set()
neighbors = []
def traverse(current_id, depth):
if depth > max_depth or current_id in visited:
return
visited.add(current_id)
# Forward edges
for edge in self.edges.get(current_id, []):
if edge_types is None or edge.edge_type in edge_types:
target_node = self.nodes.get(edge.to_node)
if target_node and target_node not in neighbors:
neighbors.append(target_node)
traverse(edge.to_node, depth + 1)
# Reverse edges
for edge in self.reverse_edges.get(current_id, []):
if edge_types is None or edge.edge_type in edge_types:
source_node = self.nodes.get(edge.from_node)
if source_node and source_node not in neighbors:
neighbors.append(source_node)
traverse(edge.from_node, depth + 1)
traverse(node_id, 0)
return neighbors
def remove_nodes(self, node_ids: List[str]):
"""Remove nodes from graph
Args:
node_ids: List of node IDs to remove
"""
for node_id in node_ids:
if node_id in self.nodes:
del self.nodes[node_id]
if node_id in self.edges:
del self.edges[node_id]
if node_id in self.reverse_edges:
del self.reverse_edges[node_id]
def save(self, nodes_path: Path, edges_path: Path):
"""Save graph to disk
Args:
nodes_path: Path to save nodes
edges_path: Path to save edges
"""
# Save nodes
nodes_data = [asdict(node) for node in self.nodes.values()]
with open(nodes_path, 'w') as f:
json.dump(nodes_data, f, indent=2)
# Save edges
edges_data = []
for edge_list in self.edges.values():
edges_data.extend([asdict(edge) for edge in edge_list])
with open(edges_path, 'w') as f:
json.dump(edges_data, f, indent=2)
def load(self, nodes_path: Path, edges_path: Path):
"""Load graph from disk
Args:
nodes_path: Path to load nodes from
edges_path: Path to load edges from
"""
# Load nodes
if nodes_path.exists():
with open(nodes_path, 'r') as f:
nodes_data = json.load(f)
for node_data in nodes_data:
node = GraphNode(**node_data)
self.add_node(node)
# Load edges
if edges_path.exists():
with open(edges_path, 'r') as f:
edges_data = json.load(f)
for edge_data in edges_data:
edge = GraphEdge(**edge_data)
self.add_edge(edge)
# ============================================================================
# Repository Analyzer
# ============================================================================
class RepositoryAnalyzer:
"""Main repository analysis system"""
def __init__(self, storage_path: str, llm: LLMInterface,
embedding_model: EmbeddingModel):
"""Initialize repository analyzer
Args:
storage_path: Path to storage directory
llm: Language model interface
embedding_model: Embedding model
"""
self.storage_path = Path(storage_path)
self.storage_path.mkdir(exist_ok=True, parents=True)
self.llm = llm
self.embedding_model = embedding_model
self.parser = CodeParser()
self.chunker = CodeChunker(self.parser)
self.graph_builder = GraphBuilder(self.parser)
self.repositories = {}
self.change_tracking = {}
def analyze_repository(self, location: str,
credentials: Optional[Dict] = None) -> str:
"""Analyze a repository
Args:
location: Repository path or URL
credentials: Optional authentication credentials
Returns:
Repository ID
"""
# Generate repository ID
repo_id = hashlib.md5(location.encode()).hexdigest()
# Check if already analyzed
repo_dir = self.storage_path / repo_id
if repo_dir.exists():
logger.info(f"Loading existing analysis for {location}")
return self._load_repository(repo_id, location, credentials)
logger.info(f"Analyzing new repository: {location}")
# Connect to repository
connector = RepositoryConnector(location, credentials)
# Initialize knowledge base
vector_db = VectorDatabase(self.embedding_model.dimension)
graph_db = GraphDatabase()
# Get repository files
files = connector.list_files()
logger.info(f"Found {len(files)} files")
# Filter to text files
text_files = self._filter_text_files(files)
logger.info(f"Processing {len(text_files)} text files")
# Process files
all_chunks = []
for i, file_path in enumerate(text_files):
if i % 10 == 0:
logger.info(f"Processing file {i+1}/{len(text_files)}")
try:
content = connector.get_file_content(file_path)
if not content:
continue
# Chunk file
chunks = self.chunker.chunk_file(file_path, content)
all_chunks.extend(chunks)
# Build graph
nodes, edges = self.graph_builder.build_graph(file_path, content)
# Add to graph database
for node in nodes:
graph_db.add_node(node)
for edge in edges:
graph_db.add_edge(edge)
except Exception as e:
logger.warning(f"Failed to process {file_path}: {e}")
logger.info(f"Created {len(all_chunks)} chunks")
# Generate embeddings
logger.info("Generating embeddings...")
chunk_texts = [chunk.text for chunk in all_chunks]
embeddings = self.embedding_model.embed_batch(chunk_texts)
# Train index if needed
if len(embeddings) >= 10000:
vector_db.train_index(embeddings)
# Add to vector database
chunk_ids = vector_db.add_embeddings(repo_id, embeddings, all_chunks)
# Link embeddings to graph nodes
for i, chunk in enumerate(all_chunks):
chunk_id = chunk_ids[i]
# Find corresponding graph node
for node in graph_db.nodes.values():
if (node.file_path == chunk.file_path and
node.line_number >= chunk.start_line and
node.line_number <= chunk.end_line):
node.embedding_id = chunk_id
chunk.node_id = node.id
break
# Get repository state
repo_state = connector.get_current_state()
# Save to disk
self._save_repository(repo_id, location, vector_db, graph_db, repo_state)
# Store in memory
self.repositories[repo_id] = {
'location': location,
'connector': connector,
'vector_db': vector_db,
'graph_db': graph_db,
'state': repo_state
}
# Generate initial report
report = self._generate_repository_report(repo_id)
logger.info(f"Analysis complete for {location}")
print("\n" + "="*80)
print("REPOSITORY ANALYSIS REPORT")
print("="*80)
print(report)
print("="*80 + "\n")
# Start change tracking
self._start_change_tracking(repo_id)
return repo_id
def _filter_text_files(self, files: List[str]) -> List[str]:
"""Filter to text files only
Args:
files: List of file paths
Returns:
Filtered list
"""
text_extensions = {
'.py', '.js', '.ts', '.java', '.c', '.cpp', '.h', '.hpp',
'.go', '.rs', '.rb', '.php', '.cs', '.swift', '.kt',
'.md', '.txt', '.json', '.yaml', '.yml', '.xml', '.html',
'.css', '.scss', '.sql', '.sh', '.bash'
}
skip_patterns = {
'node_modules/', 'venv/', 'env/', '.git/', 'build/',
'dist/', 'target/', '__pycache__/', '.pytest_cache/'
}
filtered = []
for file_path in files:
# Skip if in excluded directory
if any(pattern in file_path for pattern in skip_patterns):
continue
# Check extension
ext = Path(file_path).suffix.lower()
if ext in text_extensions:
filtered.append(file_path)
return filtered
def _generate_repository_report(self, repo_id: str) -> str:
"""Generate comprehensive repository report
Args:
repo_id: Repository identifier
Returns:
Report text
"""
repo_data = self.repositories[repo_id]
sections = []
# Overview section
overview = self._generate_overview_section(repo_id, repo_data)
sections.append(("OVERVIEW", overview))
# Structure section
structure = self._generate_structure_section(repo_id, repo_data)
sections.append(("CODE STRUCTURE", structure))
# Recent activity
activity = self._generate_activity_section(repo_id, repo_data)
sections.append(("RECENT ACTIVITY", activity))
# Format report
report_parts = []
for title, content in sections:
report_parts.append(f"\n{title}\n{'-'*len(title)}\n{content}")
return '\n'.join(report_parts)
def _generate_overview_section(self, repo_id: str, repo_data: Dict) -> str:
"""Generate overview section"""
location = repo_data['location']
state = repo_data['state']
graph_db = repo_data['graph_db']
# Count entities
num_files = len([n for n in graph_db.nodes.values() if n.type == 'file'])
num_classes = len([n for n in graph_db.nodes.values() if n.type == 'class'])
num_functions = len([n for n in graph_db.nodes.values()
if n.type in ['function', 'method']])
prompt = f"""
Generate a brief overview of this code repository:
Location: {location}
Number of files: {num_files}
Number of classes: {num_classes}
Number of functions: {num_functions}
Branches: {', '.join(state.get('branches', {}).keys())}
Last updated: {state.get('last_commit_date', 'Unknown')}
Provide a 2-3 sentence overview describing what this repository likely contains
and its purpose based on the statistics above.
"""
return self.llm.generate(prompt, max_tokens=200)
def _generate_structure_section(self, repo_id: str, repo_data: Dict) -> str:
"""Generate code structure section"""
graph_db = repo_data['graph_db']
# Get file paths
file_nodes = [n for n in graph_db.nodes.values() if n.type == 'file']
file_paths = [n.file_path for n in file_nodes]
# Build directory tree
directories = defaultdict(list)
for path in file_paths:
dir_name = str(Path(path).parent)
if dir_name == '.':
dir_name = 'root'
directories[dir_name].append(Path(path).name)
# Format directory structure
structure_lines = []
for dir_name in sorted(directories.keys())[:10]: # Top 10 directories
files = directories[dir_name]
structure_lines.append(f"{dir_name}/ ({len(files)} files)")
structure_text = '\n'.join(structure_lines)
prompt = f"""
Analyze this repository directory structure:
{structure_text}
Describe the overall organization and what each major directory likely contains.
Keep the description concise.
"""
return self.llm.generate(prompt, max_tokens=300)
def _generate_activity_section(self, repo_id: str, repo_data: Dict) -> str:
"""Generate recent activity section"""
state = repo_data['state']
last_commit = state.get('last_commit_date', 'Unknown')
num_branches = len(state.get('branches', {}))
return f"Last commit: {last_commit}\nActive branches: {num_branches}"
def _save_repository(self, repo_id: str, location: str,
vector_db: VectorDatabase, graph_db: GraphDatabase,
state: Dict):
"""Save repository data to disk"""
repo_dir = self.storage_path / repo_id
repo_dir.mkdir(exist_ok=True)
# Save metadata
metadata = {
'repo_id': repo_id,
'location': location,
'last_analyzed': datetime.now().isoformat()
}
with open(repo_dir / 'metadata.json', 'w') as f:
json.dump(metadata, f, indent=2)
# Save vector database
vector_db_dir = repo_dir / 'vector_db'
vector_db_dir.mkdir(exist_ok=True)
vector_db.save(vector_db_dir / 'index.faiss')
chunks_data = [asdict(chunk) for chunk in vector_db.chunks_by_id.values()]
with open(vector_db_dir / 'chunks.json', 'w') as f:
json.dump(chunks_data, f, indent=2)
# Save graph database
graph_db_dir = repo_dir / 'graph_db'
graph_db_dir.mkdir(exist_ok=True)
graph_db.save(graph_db_dir / 'nodes.json', graph_db_dir / 'edges.json')
# Save state
with open(repo_dir / 'state.json', 'w') as f:
json.dump(state, f, indent=2)
logger.info(f"Saved repository data to {repo_dir}")
def _load_repository(self, repo_id: str, location: str,
credentials: Optional[Dict] = None) -> str:
"""Load repository from disk"""
repo_dir = self.storage_path / repo_id
# Load metadata
with open(repo_dir / 'metadata.json', 'r') as f:
metadata = json.load(f)
# Load vector database
vector_db = VectorDatabase(self.embedding_model.dimension)
vector_db_dir = repo_dir / 'vector_db'
vector_db.load(vector_db_dir / 'index.faiss')
with open(vector_db_dir / 'chunks.json', 'r') as f:
chunks_data = json.load(f)
for chunk_data in chunks_data:
chunk = Chunk(**chunk_data)
vector_db.chunks_by_id[chunk.id] = chunk
# Load graph database
graph_db = GraphDatabase()
graph_db_dir = repo_dir / 'graph_db'
graph_db.load(graph_db_dir / 'nodes.json', graph_db_dir / 'edges.json')
# Load state
with open(repo_dir / 'state.json', 'r') as f:
state = json.load(f)
# Connect to repository
connector = RepositoryConnector(location, credentials)
# Store in memory
self.repositories[repo_id] = {
'location': location,
'connector': connector,
'vector_db': vector_db,
'graph_db': graph_db,
'state': state
}
logger.info(f"Loaded repository {repo_id}")
# Check for changes
self._check_and_update_changes(repo_id)
# Start change tracking
self._start_change_tracking(repo_id)
return repo_id
def _start_change_tracking(self, repo_id: str):
"""Start background change tracking for repository"""
if repo_id in self.change_tracking:
return
stop_event = threading.Event()
def track_changes():
while not stop_event.is_set():
time.sleep(300) # Check every 5 minutes
if not stop_event.is_set():
self._check_and_update_changes(repo_id)
thread = threading.Thread(target=track_changes, daemon=True)
thread.start()
self.change_tracking[repo_id] = {
'thread': thread,
'stop_event': stop_event,
'pending_changes': None
}
logger.info(f"Started change tracking for {repo_id}")
def _check_and_update_changes(self, repo_id: str):
"""Check for repository changes and update if needed"""
if repo_id not in self.repositories:
return
repo_data = self.repositories[repo_id]
connector = repo_data['connector']
stored_state = repo_data['state']
# Get current state
current_state = connector.get_current_state()
# Detect changes
changes = self._detect_changes(stored_state, current_state)
if self._has_significant_changes(changes):
logger.info(f"Detected changes in repository {repo_id}")
# Update knowledge base
self._update_knowledge_base(repo_id, changes)
# Generate change summary
summary = self._generate_change_summary(repo_id, changes)
# Store pending changes
if repo_id in self.change_tracking:
self.change_tracking[repo_id]['pending_changes'] = {
'summary': summary,
'timestamp': datetime.now().isoformat()
}
# Update state
repo_data['state'] = current_state
# Save updated data
self._save_repository(
repo_id,
repo_data['location'],
repo_data['vector_db'],
repo_data['graph_db'],
current_state
)
print(f"\n{self.get_change_icon(repo_id)} Repository {repo_id[:8]} has been updated")
print(summary)
def _detect_changes(self, stored_state: Dict,
current_state: Dict) -> RepositoryChange:
"""Detect changes between states"""
# For simplicity, just detect branch head changes
# In production, would fetch actual commit diffs
new_commits = []
stored_branches = stored_state.get('branches', {})
current_branches = current_state.get('branches', {})
for branch, current_head in current_branches.items():
stored_head = stored_branches.get(branch)
if stored_head != current_head:
new_commits.append({
'branch': branch,
'old_head': stored_head,
'new_head': current_head
})
return RepositoryChange(
new_commits=new_commits,
modified_files=[],
new_files=[],
deleted_files=[],
new_issues=[],
updated_issues=[]
)
def _has_significant_changes(self, changes: RepositoryChange) -> bool:
"""Check if changes are significant"""
return (len(changes.new_commits) > 0 or
len(changes.modified_files) > 0 or
len(changes.new_files) > 0 or
len(changes.deleted_files) > 0)
def _update_knowledge_base(self, repo_id: str, changes: RepositoryChange):
"""Update knowledge base with changes"""
# In production, would fetch changed files and update embeddings/graph
# For this example, we'll log the update
logger.info(f"Updated knowledge base for {repo_id}")
def _generate_change_summary(self, repo_id: str,
changes: RepositoryChange) -> str:
"""Generate summary of changes"""
parts = []
if changes.new_commits:
parts.append(f"{len(changes.new_commits)} new commits detected")
if changes.modified_files:
parts.append(f"{len(changes.modified_files)} files modified")
if changes.new_files:
parts.append(f"{len(changes.new_files)} files added")
if changes.deleted_files:
parts.append(f"{len(changes.deleted_files)} files deleted")
return "Changes detected: " + ", ".join(parts)
def get_change_icon(self, repo_id: str) -> str:
"""Get change notification icon"""
if repo_id not in self.change_tracking:
return ''
pending = self.change_tracking[repo_id].get('pending_changes')
if pending:
return '●'
return ''
def answer_question(self, repo_id: str, question: str) -> str:
"""Answer question about repository
Args:
repo_id: Repository identifier
question: User question
Returns:
Answer text
"""
if repo_id not in self.repositories:
return "Repository not found"
repo_data = self.repositories[repo_id]
vector_db = repo_data['vector_db']
graph_db = repo_data['graph_db']
# Generate query embedding
query_embedding = self.embedding_model.embed(question)
# Vector similarity search
similar_chunks = vector_db.similarity_search(query_embedding, k=5)
# Expand context using graph
expanded_chunks = []
for chunk in similar_chunks:
expanded_chunks.append(chunk)
# Get related graph nodes
if chunk.node_id:
neighbors = graph_db.get_neighbors(chunk.node_id, max_depth=1)
for neighbor in neighbors[:3]:
if neighbor.embedding_id:
neighbor_chunk = vector_db.get_chunk_by_id(neighbor.embedding_id)
if neighbor_chunk and neighbor_chunk not in expanded_chunks:
expanded_chunks.append(neighbor_chunk)
# Build context
context_parts = []
for chunk in expanded_chunks[:10]:
context_parts.append(f"File: {chunk.file_path}")
context_parts.append(f"Lines {chunk.start_line}-{chunk.end_line}:")
context_parts.append(chunk.text)
context_parts.append("")
context = '\n'.join(context_parts)
# Generate answer
prompt = f"""
You are analyzing a code repository. Answer the following question based on
the provided context from the repository.
Question: {question}
Context from repository:
{context}
Provide a detailed answer based on the context. Include specific references
to files and line numbers when relevant. If the context does not contain
enough information to answer the question, say so clearly.
Answer:
"""
answer = self.llm.generate(prompt, max_tokens=500)
return answer
# ============================================================================
# Main Application
# ============================================================================
def main():
"""Main application entry point"""
import argparse
parser = argparse.ArgumentParser(
description='Repository Analysis Agentic AI System'
)
parser.add_argument(
'action',
choices=['analyze', 'question'],
help='Action to perform'
)
parser.add_argument(
'--repo',
required=True,
help='Repository path or URL'
)
parser.add_argument(
'--question',
help='Question to ask (for question action)'
)
parser.add_argument(
'--storage',
default='./repo_storage',
help='Storage directory path'
)
parser.add_argument(
'--llm-type',
choices=['local', 'openai'],
default='local',
help='LLM type to use'
)
parser.add_argument(
'--model-path',
help='Path to local model file (for local LLM)'
)
parser.add_argument(
'--api-key',
help='API key (for remote LLM)'
)
args = parser.parse_args()
# Initialize LLM
if args.llm_type == 'local':
if not args.model_path:
print("Error: --model-path required for local LLM")
return
llm = LocalLLM(args.model_path, n_gpu_layers=32)
else:
if not args.api_key:
print("Error: --api-key required for OpenAI LLM")
return
llm = OpenAILLM(args.api_key)
# Initialize embedding model
embedding_model = EmbeddingModel()
# Initialize analyzer
analyzer = RepositoryAnalyzer(args.storage, llm, embedding_model)
# Perform action
if args.action == 'analyze':
repo_id = analyzer.analyze_repository(args.repo)
print(f"\nRepository ID: {repo_id}")
print("Analysis complete. You can now ask questions about this repository.")
elif args.action == 'question':
if not args.question:
print("Error: --question required for question action")
return
# Generate repo ID
repo_id = hashlib.md5(args.repo.encode()).hexdigest()
# Load repository if not in memory
if repo_id not in analyzer.repositories:
analyzer._load_repository(repo_id, args.repo)
# Answer question
answer = analyzer.answer_question(repo_id, args.question)
print("\n" + "="*80)
print("ANSWER")
print("="*80)
print(answer)
print("="*80 + "\n")
if __name__ == '__main__':
main()
This complete implementation provides a production-ready agentic AI system for repository analysis. The code includes all necessary components: repository connection handling for both local and remote repositories, code parsing and chunking using tree-sitter, hybrid RAG and GraphRAG implementation with bidirectional links, vector database using FAISS for efficient similarity search, graph database for structural relationships, continuous change tracking with background threads, comprehensive report generation using the language model, interactive question answering with context expansion, multi-GPU backend support for various architectures, persistent storage and efficient reloading of analyzed repositories, robust error handling throughout, and a command-line interface for easy usage.
The system can be used to analyze any GitHub or GitLab repository, track changes continuously, and answer natural language questions about the codebase. It supports both local and remote language models and works across different GPU architectures including Nvidia CUDA, AMD ROCm, Apple MPS, and Intel GPUs.
