INTRODUCTION TO SELF-LEARNING CONVERSATIONAL AI
The landscape of artificial intelligence has evolved dramatically with the emergence of large language models. However, most production systems rely on cloud-based APIs that raise concerns about data privacy, cost, and control. This tutorial presents a comprehensive implementation of a self-learning chatbot that operates entirely on local infrastructure using open source components. The system combines traditional Retrieval Augmented Generation with Graph-based RAG to create a hybrid knowledge architecture that enables sophisticated information retrieval and reasoning.
The chatbot operates in two distinct modes. The default chat mode allows users to ask questions and receive responses enhanced by previously learned information. When a user prefixes their input with the keyword "learn:" followed by topic specifications, the system transitions into learning mode. In this mode, the application actively searches for relevant documents, processes them through an intelligent pipeline, and stores the extracted knowledge in both vector and graph databases. This dual-mode operation creates a system that continuously expands its knowledge base while maintaining conversational capabilities.
What distinguishes this implementation from simpler RAG systems is the hybrid knowledge architecture. Traditional RAG systems store document chunks as embeddings in a vector database and retrieve relevant chunks based on semantic similarity. While effective, this approach loses the relational context between different pieces of information. GraphRAG addresses this limitation by representing knowledge as a graph where entities become nodes and relationships become edges. Our hybrid system combines both approaches, creating bidirectional references where vector embeddings can point to graph entities and graph nodes can reference specific embeddings. This architecture enables the system to leverage both semantic similarity and structural relationships when retrieving information.
The system must handle diverse document types including HTML pages, PDF documents, source code files, and plain text. Each document type requires different processing strategies. HTML documents need parsing to extract meaningful content while discarding navigation elements and advertisements. PDF files require layout analysis to maintain reading order and extract embedded images or tables. Source code demands specialized chunking that respects syntactic boundaries like function definitions and class declarations. The intelligent selection of processing strategies based on document type and content characteristics represents a critical component of the system architecture.
Cross-platform GPU support presents another significant challenge. Modern GPU computing spans multiple ecosystems including NVIDIA CUDA, AMD ROCm, Intel oneAPI, and Apple Metal Performance Shaders. Each platform requires different libraries and initialization procedures. The system must detect available hardware at runtime and configure the LLM backend accordingly. This flexibility ensures the chatbot can run on diverse hardware configurations from high-end workstations with discrete GPUs to laptops with integrated graphics.
ARCHITECTURAL FOUNDATIONS AND DESIGN PRINCIPLES
The system architecture follows a modular design where each component handles a specific responsibility while maintaining loose coupling through well-defined interfaces. At the highest level, the application consists of five major subsystems: the mode controller, the document processor, the knowledge storage layer, the retrieval engine, and the LLM interface. These subsystems communicate through a central orchestrator that manages state transitions and coordinates operations.
The mode controller determines the current operational state based on user input. When the system starts, it initializes in chat mode. The controller monitors incoming messages and applies pattern matching to detect the "learn:" prefix. Upon detection, it extracts the topic specification, transitions to learning mode, and triggers the document processing pipeline. After processing completes, the controller automatically returns the system to chat mode. This state machine approach ensures clean separation between conversational and learning behaviors.
The document processor implements a sophisticated pipeline that handles the entire workflow from document discovery to knowledge extraction. The pipeline begins with a document finder that searches local file systems and optionally crawls web resources based on the specified topics. Each discovered document passes through a type detector that analyzes file extensions, MIME types, and content signatures to determine the appropriate processing strategy. The processor then applies document-specific parsers to extract clean text while preserving structural information like headings, lists, and code blocks.
After extraction, the content enters the chunking stage where the system must divide long documents into manageable segments. Naive chunking approaches that split text at fixed character boundaries often break semantic units and harm retrieval quality. The intelligent chunker analyzes content structure and applies different strategies based on document type. For narrative text, it uses semantic chunking that identifies topic boundaries through embedding similarity analysis. For code, it employs syntax-aware chunking that respects function and class boundaries. For structured documents like academic papers, it leverages section headers to create natural chunks.
The knowledge storage layer maintains two parallel data structures: a vector database for semantic search and a graph database for relational queries. The vector database stores document chunks as high-dimensional embeddings generated by a specialized embedding model. Each embedding entry includes metadata such as source document, chunk position, creation timestamp, and content hash. The graph database represents extracted entities and their relationships. Entities can include named entities like people, organizations, and locations, as well as conceptual entities like topics, methods, and findings. Relationships capture how entities connect, such as authorship, citation, or topical association.
The bidirectional reference system creates links between these two knowledge stores. When the system extracts entities from a document chunk, it creates graph nodes for new entities and updates existing nodes for known entities. Each graph node maintains a list of embedding IDs where that entity appears. Conversely, each vector embedding stores a list of entity IDs that appear in its content. This dual indexing enables sophisticated retrieval strategies. For example, when a user asks about a specific person, the system can find the graph node for that person, retrieve all associated embeddings, and use those embeddings as starting points for semantic expansion to related concepts.
The retrieval engine implements multiple search strategies and combines their results through a reranking process. For a given query, it performs parallel searches across both knowledge stores. The vector search generates embeddings for the query and retrieves the most similar document chunks. The graph search identifies entities mentioned in the query and traverses the graph to find related entities and their associated content. The system also performs hybrid searches that start with graph traversal to identify relevant entities, then use those entities to filter vector search results. All retrieved candidates pass through a reranker that uses a cross-encoder model to score query-document relevance more accurately than the initial retrieval methods.
IMPLEMENTING THE LLM BACKEND WITH MULTI-PLATFORM SUPPORT
The foundation of any LLM-based application is the language model itself. For a local system, we need an inference engine that can load and run models efficiently on diverse hardware. The llama.cpp project provides a high-performance C++ implementation with Python bindings that support multiple GPU backends through a unified interface. The system must detect available hardware and configure the backend accordingly.
Here is how the LLM backend initialization works with hardware detection:
import platform
import subprocess
import os
from llama_cpp import Llama
class LLMBackend:
def __init__(self, model_path, context_size=4096):
self.model_path = model_path
self.context_size = context_size
self.device_type = self._detect_gpu()
self.model = self._initialize_model()
def _detect_gpu(self):
system = platform.system()
if system == "Darwin":
return "metal"
try:
result = subprocess.run(['nvidia-smi'],
capture_output=True,
text=True,
timeout=5)
if result.returncode == 0:
return "cuda"
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
try:
result = subprocess.run(['rocm-smi'],
capture_output=True,
text=True,
timeout=5)
if result.returncode == 0:
return "rocm"
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
if os.path.exists('/dev/dri'):
return "vulkan"
return "cpu"
def _initialize_model(self):
n_gpu_layers = -1 if self.device_type != "cpu" else 0
return Llama(
model_path=self.model_path,
n_ctx=self.context_size,
n_gpu_layers=n_gpu_layers,
n_threads=os.cpu_count() if self.device_type == "cpu" else 4,
verbose=False
)
The hardware detection logic examines the runtime environment to identify available GPU resources. On macOS systems, the code assumes Metal Performance Shaders availability since all modern Macs include GPU hardware. For other platforms, it attempts to execute vendor-specific utilities. The nvidia-smi command indicates NVIDIA GPU presence, while rocm-smi signals AMD hardware. For Intel GPUs, the code checks for Direct Rendering Infrastructure devices in the Linux device tree. If no GPU is detected, the system falls back to CPU inference.
The model initialization configures llama.cpp parameters based on detected hardware. The n_gpu_layers parameter controls how many transformer layers run on the GPU versus CPU. Setting this to negative one instructs llama.cpp to offload all possible layers to the GPU. For CPU-only systems, we set it to zero and increase thread count to leverage multiple cores. The context size determines how much text the model can process in a single forward pass, affecting both memory usage and the length of conversations the system can maintain.
Once initialized, the LLM backend provides methods for text generation with various parameters:
def generate(self, prompt, max_tokens=512, temperature=0.7,
top_p=0.9, stop_sequences=None):
if stop_sequences is None:
stop_sequences = []
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=stop_sequences,
echo=False
)
return response['choices'][0]['text'].strip()
The generation method wraps the llama.cpp inference API with sensible defaults. Temperature controls randomness in token selection, with lower values producing more deterministic outputs. The top_p parameter implements nucleus sampling, considering only the most probable tokens whose cumulative probability exceeds the threshold. Stop sequences allow early termination when the model generates specific strings, useful for implementing structured outputs or multi-turn conversations.
DOCUMENT PROCESSING AND INTELLIGENT CHUNKING STRATEGIES
The document processing pipeline transforms raw files into structured knowledge suitable for storage in vector and graph databases. This pipeline must handle heterogeneous document types while maintaining semantic coherence and extracting meaningful entities. The process begins with document discovery and type detection.
For local file systems, the document finder recursively traverses directories matching file extensions against a whitelist of supported types. For web resources, it implements a focused crawler that starts from seed URLs related to the specified topics and follows links within a defined scope. Here is the document finder implementation:
import os
import mimetypes
from pathlib import Path
from urllib.parse import urlparse, urljoin
import requests
from bs4 import BeautifulSoup
class DocumentFinder:
def __init__(self):
self.supported_extensions = {
'.txt', '.md', '.pdf', '.html', '.htm',
'.py', '.java', '.cpp', '.c', '.js', '.ts',
'.json', '.xml', '.csv'
}
def find_local_documents(self, root_path, max_depth=5):
documents = []
root = Path(root_path)
for path in root.rglob('*'):
if path.is_file() and path.suffix.lower() in self.supported_extensions:
if self._get_depth(root, path) <= max_depth:
documents.append({
'path': str(path),
'type': 'local',
'extension': path.suffix.lower()
})
return documents
def _get_depth(self, root, path):
try:
relative = path.relative_to(root)
return len(relative.parts) - 1
except ValueError:
return 0
def find_web_documents(self, seed_urls, max_pages=50):
documents = []
visited = set()
to_visit = list(seed_urls)
while to_visit and len(documents) < max_pages:
url = to_visit.pop(0)
if url in visited:
continue
visited.add(url)
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
content_type = response.headers.get('content-type', '')
if 'text/html' in content_type:
documents.append({
'path': url,
'type': 'web',
'extension': '.html',
'content': response.content
})
soup = BeautifulSoup(response.content, 'html.parser')
base_domain = urlparse(url).netloc
for link in soup.find_all('a', href=True):
absolute_url = urljoin(url, link['href'])
link_domain = urlparse(absolute_url).netloc
if link_domain == base_domain and absolute_url not in visited:
to_visit.append(absolute_url)
elif 'application/pdf' in content_type:
documents.append({
'path': url,
'type': 'web',
'extension': '.pdf',
'content': response.content
})
except Exception as e:
print(f"Error fetching {url}: {str(e)}")
continue
return documents
The local document finder uses Python's pathlib to traverse directory structures efficiently. It filters files based on extension and enforces a maximum depth to prevent excessive recursion in deeply nested directories. The web document finder implements breadth-first crawling with domain restriction to avoid wandering into unrelated sites. It respects content types to handle both HTML pages and binary documents like PDFs.
After document discovery, the system must extract clean text from various formats. HTML documents require special handling to remove boilerplate content like navigation menus, advertisements, and footers while preserving the main content. The extraction process uses heuristics based on tag density and text length:
from bs4 import BeautifulSoup
import re
class HTMLExtractor:
def extract(self, html_content):
soup = BeautifulSoup(html_content, 'html.parser')
for element in soup(['script', 'style', 'nav', 'footer',
'header', 'aside', 'form']):
element.decompose()
main_content = soup.find('main') or soup.find('article') or soup.body
if not main_content:
return ""
text_blocks = []
for element in main_content.find_all(['p', 'h1', 'h2', 'h3',
'h4', 'h5', 'h6', 'li']):
text = element.get_text(strip=True)
if len(text) > 20:
text_blocks.append(text)
return '\n\n'.join(text_blocks)
The HTML extractor removes non-content elements like scripts and stylesheets, then focuses on semantic containers like main and article tags. It extracts text from paragraph and heading elements while filtering out very short snippets that likely represent navigation or metadata rather than substantive content.
PDF extraction requires different tools since PDFs store text as positioned glyphs rather than flowing content. The PyMuPDF library provides robust extraction with layout preservation:
import fitz
class PDFExtractor:
def extract(self, pdf_path_or_bytes):
if isinstance(pdf_path_or_bytes, bytes):
doc = fitz.open(stream=pdf_path_or_bytes, filetype="pdf")
else:
doc = fitz.open(pdf_path_or_bytes)
text_blocks = []
for page_num in range(len(doc)):
page = doc[page_num]
blocks = page.get_text("blocks")
for block in blocks:
if block[6] == 0:
text = block[4].strip()
if len(text) > 20:
text_blocks.append(text)
doc.close()
return '\n\n'.join(text_blocks)
The PDF extractor opens documents from either file paths or byte streams to support both local files and web-downloaded content. It iterates through pages and extracts text blocks while filtering based on block type and length. The block-based extraction preserves reading order better than simple text extraction methods.
With clean text extracted, the system must divide it into chunks suitable for embedding. The chunking strategy significantly impacts retrieval quality. Fixed-size chunking creates arbitrary boundaries that may split semantic units. Sentence-based chunking respects linguistic boundaries but may create chunks too small or too large for effective retrieval. The intelligent chunker analyzes content structure and applies adaptive strategies:
from sentence_transformers import SentenceTransformer
import numpy as np
class IntelligentChunker:
def __init__(self, embedding_model):
self.embedding_model = embedding_model
self.min_chunk_size = 100
self.max_chunk_size = 1000
self.similarity_threshold = 0.7
def chunk_text(self, text, content_type='text'):
if content_type == 'code':
return self._chunk_code(text)
else:
return self._chunk_semantic(text)
def _chunk_semantic(self, text):
sentences = self._split_sentences(text)
if len(sentences) == 0:
return []
embeddings = self.embedding_model.encode(sentences)
chunks = []
current_chunk = [sentences[0]]
current_size = len(sentences[0])
for i in range(1, len(sentences)):
similarity = np.dot(embeddings[i-1], embeddings[i]) / (
np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
)
sentence_len = len(sentences[i])
if (similarity < self.similarity_threshold or
current_size + sentence_len > self.max_chunk_size):
if current_size >= self.min_chunk_size:
chunks.append(' '.join(current_chunk))
current_chunk = [sentences[i]]
current_size = sentence_len
else:
current_chunk.append(sentences[i])
current_size += sentence_len
else:
current_chunk.append(sentences[i])
current_size += sentence_len
if current_chunk:
chunks.append(' '.join(current_chunk))
return chunks
def _split_sentences(self, text):
import re
sentence_endings = re.compile(r'(?<=[.!?])\s+(?=[A-Z])')
sentences = sentence_endings.split(text)
return [s.strip() for s in sentences if s.strip()]
def _chunk_code(self, code):
import ast
try:
tree = ast.parse(code)
chunks = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
start_line = node.lineno - 1
end_line = node.end_lineno
code_lines = code.split('\n')
chunk = '\n'.join(code_lines[start_line:end_line])
if len(chunk) >= self.min_chunk_size:
chunks.append(chunk)
return chunks if chunks else [code]
except SyntaxError:
return self._chunk_semantic(code)
The semantic chunker splits text into sentences and generates embeddings for each sentence. It then groups consecutive sentences based on embedding similarity. When similarity drops below a threshold, indicating a topic shift, the chunker starts a new chunk. This approach creates semantically coherent chunks that respect natural topic boundaries. The code chunker uses Python's abstract syntax tree parser to identify function and class definitions, creating chunks that preserve complete syntactic units.
BUILDING THE HYBRID KNOWLEDGE STORAGE SYSTEM
The knowledge storage layer combines vector and graph databases to enable both semantic and structural retrieval. The vector database stores document chunk embeddings for similarity search, while the graph database represents entities and relationships for traversal-based queries. The bidirectional reference system links these two stores, allowing queries to leverage both retrieval paradigms.
For vector storage, ChromaDB provides a lightweight, embeddable solution with persistent storage and efficient similarity search. The vector store wrapper handles embedding generation, storage, and retrieval:
import chromadb
from chromadb.config import Settings
import hashlib
class VectorStore:
def __init__(self, persist_directory, embedding_model):
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_directory
))
self.embedding_model = embedding_model
self.collection = self.client.get_or_create_collection(
name="documents",
metadata={"hnsw:space": "cosine"}
)
def add_chunks(self, chunks, metadata_list):
if not chunks:
return []
embeddings = self.embedding_model.encode(chunks).tolist()
ids = []
new_chunks = []
new_embeddings = []
new_metadata = []
for i, (chunk, embedding, metadata) in enumerate(
zip(chunks, embeddings, metadata_list)
):
chunk_hash = self._compute_hash(chunk)
existing = self.collection.get(
where={"content_hash": chunk_hash}
)
if not existing['ids']:
chunk_id = f"chunk_{chunk_hash}_{i}"
ids.append(chunk_id)
new_chunks.append(chunk)
new_embeddings.append(embedding)
metadata['content_hash'] = chunk_hash
new_metadata.append(metadata)
if new_chunks:
self.collection.add(
ids=ids,
embeddings=new_embeddings,
documents=new_chunks,
metadatas=new_metadata
)
return ids
def _compute_hash(self, text):
return hashlib.sha256(text.encode()).hexdigest()
def search(self, query, n_results=5, filter_metadata=None):
query_embedding = self.embedding_model.encode([query]).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=n_results,
where=filter_metadata
)
return results
The vector store computes content hashes for deduplication, ensuring the system never stores identical chunks multiple times. When adding new chunks, it checks existing entries by hash and only inserts novel content. This deduplication mechanism prevents knowledge base bloat when processing overlapping documents or updated versions of previously seen content.
The graph database represents entities and relationships using NetworkX with persistent storage through JSON serialization. The graph store provides methods for entity extraction, relationship creation, and traversal:
import networkx as nx
import json
from pathlib import Path
class GraphStore:
def __init__(self, persist_path):
self.persist_path = Path(persist_path)
self.graph = self._load_graph()
def _load_graph(self):
if self.persist_path.exists():
with open(self.persist_path, 'r') as f:
data = json.load(f)
return nx.node_link_graph(data)
return nx.DiGraph()
def save(self):
self.persist_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.persist_path, 'w') as f:
data = nx.node_link_data(self.graph)
json.dump(data, f)
def add_entity(self, entity_name, entity_type, embedding_ids=None):
if embedding_ids is None:
embedding_ids = []
if self.graph.has_node(entity_name):
existing_ids = self.graph.nodes[entity_name].get('embedding_ids', [])
embedding_ids = list(set(existing_ids + embedding_ids))
self.graph.add_node(
entity_name,
type=entity_type,
embedding_ids=embedding_ids
)
def add_relationship(self, source, target, relation_type):
self.graph.add_edge(source, target, type=relation_type)
def get_entity(self, entity_name):
if self.graph.has_node(entity_name):
return self.graph.nodes[entity_name]
return None
def get_related_entities(self, entity_name, max_depth=2):
if not self.graph.has_node(entity_name):
return []
related = set()
for node in nx.single_source_shortest_path_length(
self.graph, entity_name, cutoff=max_depth
).keys():
if node != entity_name:
related.add(node)
return list(related)
def get_embedding_ids_for_entity(self, entity_name):
entity = self.get_entity(entity_name)
if entity:
return entity.get('embedding_ids', [])
return []
The graph store maintains entity nodes with attributes including type classification and lists of associated embedding IDs. When the system encounters an entity in a document chunk, it creates or updates the entity node and adds the chunk's embedding ID to the entity's reference list. This creates the vector-to-graph direction of the bidirectional reference system.
Entity extraction requires natural language processing to identify meaningful entities in text. The system uses a combination of named entity recognition and keyword extraction:
import spacy
from collections import Counter
class EntityExtractor:
def __init__(self):
self.nlp = spacy.load("en_core_web_sm")
def extract_entities(self, text):
doc = self.nlp(text)
entities = []
for ent in doc.ents:
if ent.label_ in ['PERSON', 'ORG', 'GPE', 'PRODUCT', 'EVENT']:
entities.append({
'text': ent.text,
'type': ent.label_,
'start': ent.start_char,
'end': ent.end_char
})
noun_chunks = [chunk.text for chunk in doc.noun_chunks
if len(chunk.text.split()) <= 3]
chunk_freq = Counter(noun_chunks)
for chunk, freq in chunk_freq.most_common(10):
if freq > 1:
entities.append({
'text': chunk,
'type': 'CONCEPT',
'start': -1,
'end': -1
})
return entities
The entity extractor uses spaCy for named entity recognition, identifying people, organizations, locations, products, and events. It also extracts frequent noun phrases as conceptual entities, capturing domain-specific terminology that may not match predefined entity types. This dual approach ensures comprehensive entity coverage across diverse document types.
IMPLEMENTING THE RETRIEVAL ENGINE WITH RERANKING
The retrieval engine orchestrates multiple search strategies and combines their results to find the most relevant information for a given query. It performs parallel searches across the vector and graph stores, then applies reranking to produce a final ranked list of relevant chunks.
The retrieval process begins with query analysis to identify entities and generate embeddings:
class RetrievalEngine:
def __init__(self, vector_store, graph_store, entity_extractor,
embedding_model, reranker_model):
self.vector_store = vector_store
self.graph_store = graph_store
self.entity_extractor = entity_extractor
self.embedding_model = embedding_model
self.reranker_model = reranker_model
def retrieve(self, query, top_k=5):
vector_results = self._vector_search(query, top_k * 3)
graph_results = self._graph_search(query, top_k * 3)
all_candidates = self._merge_results(vector_results, graph_results)
reranked = self._rerank(query, all_candidates, top_k)
return reranked
def _vector_search(self, query, k):
results = self.vector_store.search(query, n_results=k)
candidates = []
for i in range(len(results['ids'][0])):
candidates.append({
'id': results['ids'][0][i],
'text': results['documents'][0][i],
'score': 1.0 - results['distances'][0][i],
'metadata': results['metadatas'][0][i],
'source': 'vector'
})
return candidates
def _graph_search(self, query, k):
entities = self.entity_extractor.extract_entities(query)
if not entities:
return []
embedding_ids = set()
for entity in entities:
entity_name = entity['text']
direct_ids = self.graph_store.get_embedding_ids_for_entity(
entity_name
)
embedding_ids.update(direct_ids)
related = self.graph_store.get_related_entities(entity_name)
for related_entity in related:
related_ids = self.graph_store.get_embedding_ids_for_entity(
related_entity
)
embedding_ids.update(related_ids)
candidates = []
for emb_id in list(embedding_ids)[:k]:
result = self.vector_store.collection.get(ids=[emb_id])
if result['ids']:
candidates.append({
'id': emb_id,
'text': result['documents'][0],
'score': 0.8,
'metadata': result['metadatas'][0],
'source': 'graph'
})
return candidates
def _merge_results(self, vector_results, graph_results):
seen_ids = set()
merged = []
for result in vector_results + graph_results:
if result['id'] not in seen_ids:
seen_ids.add(result['id'])
merged.append(result)
return merged
def _rerank(self, query, candidates, top_k):
if not candidates:
return []
texts = [c['text'] for c in candidates]
pairs = [[query, text] for text in texts]
scores = self.reranker_model.predict(pairs)
for i, candidate in enumerate(candidates):
candidate['rerank_score'] = float(scores[i])
candidates.sort(key=lambda x: x['rerank_score'], reverse=True)
return candidates[:top_k]
The vector search generates query embeddings and retrieves similar chunks from the vector database. The graph search extracts entities from the query, finds those entities in the graph, traverses to related entities, and collects all associated embedding IDs. This graph-based approach discovers relevant content through entity relationships even when semantic similarity is low.
The reranker uses a cross-encoder model that scores query-document pairs more accurately than the initial retrieval methods. Cross-encoders process query and document together, allowing attention mechanisms to model their interaction. This produces better relevance scores than comparing embeddings independently. The sentence-transformers library provides efficient cross-encoder implementations:
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name='cross-encoder/ms-marco-MiniLM-L-6-v2'):
self.model = CrossEncoder(model_name)
def predict(self, query_document_pairs):
scores = self.model.predict(query_document_pairs)
return scores
The reranker model loads a pretrained cross-encoder fine-tuned on passage ranking tasks. It accepts lists of query-document pairs and returns relevance scores. These scores provide the final ranking for retrieved candidates.
ORCHESTRATING THE LEARNING PIPELINE
The learning pipeline coordinates document discovery, processing, entity extraction, and knowledge storage. When the user triggers learning mode with a topic specification, the orchestrator executes this pipeline and manages incremental updates.
The learning orchestrator maintains state about processed documents to enable incremental learning:
import json
from pathlib import Path
from datetime import datetime
class LearningOrchestrator:
def __init__(self, vector_store, graph_store, entity_extractor,
document_finder, chunker):
self.vector_store = vector_store
self.graph_store = graph_store
self.entity_extractor = entity_extractor
self.document_finder = document_finder
self.chunker = chunker
self.state_file = Path("learning_state.json")
self.processed_docs = self._load_state()
def _load_state(self):
if self.state_file.exists():
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def _save_state(self):
with open(self.state_file, 'w') as f:
json.dump(self.processed_docs, f, indent=2)
def learn(self, topics, search_paths=None, search_urls=None):
documents = []
if search_paths:
for path in search_paths:
docs = self.document_finder.find_local_documents(path)
documents.extend(docs)
if search_urls:
docs = self.document_finder.find_web_documents(search_urls)
documents.extend(docs)
new_or_changed = self._filter_documents(documents)
print(f"Processing {len(new_or_changed)} new or changed documents...")
for doc in new_or_changed:
self._process_document(doc, topics)
self.graph_store.save()
self._save_state()
print(f"Learning complete. Processed {len(new_or_changed)} documents.")
def _filter_documents(self, documents):
new_or_changed = []
for doc in documents:
doc_id = doc['path']
if doc['type'] == 'local':
current_hash = self._compute_file_hash(doc['path'])
else:
current_hash = self._compute_content_hash(doc.get('content', b''))
if doc_id not in self.processed_docs:
new_or_changed.append(doc)
self.processed_docs[doc_id] = {
'hash': current_hash,
'processed_at': datetime.now().isoformat()
}
elif self.processed_docs[doc_id]['hash'] != current_hash:
new_or_changed.append(doc)
self.processed_docs[doc_id]['hash'] = current_hash
self.processed_docs[doc_id]['processed_at'] = datetime.now().isoformat()
return new_or_changed
def _compute_file_hash(self, filepath):
import hashlib
hasher = hashlib.sha256()
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hasher.update(chunk)
return hasher.hexdigest()
def _compute_content_hash(self, content):
import hashlib
return hashlib.sha256(content).hexdigest()
def _process_document(self, doc, topics):
text = self._extract_text(doc)
if not text:
return
content_type = 'code' if doc['extension'] in ['.py', '.java', '.cpp',
'.c', '.js', '.ts'] else 'text'
chunks = self.chunker.chunk_text(text, content_type)
metadata_list = []
for i, chunk in enumerate(chunks):
metadata = {
'source': doc['path'],
'chunk_index': i,
'topics': topics,
'content_type': content_type
}
metadata_list.append(metadata)
chunk_ids = self.vector_store.add_chunks(chunks, metadata_list)
for chunk_id, chunk in zip(chunk_ids, chunks):
entities = self.entity_extractor.extract_entities(chunk)
for entity in entities:
self.graph_store.add_entity(
entity['text'],
entity['type'],
[chunk_id]
)
for i in range(len(entities)):
for j in range(i + 1, len(entities)):
self.graph_store.add_relationship(
entities[i]['text'],
entities[j]['text'],
'co-occurs'
)
def _extract_text(self, doc):
if doc['extension'] in ['.html', '.htm']:
extractor = HTMLExtractor()
if doc['type'] == 'web':
return extractor.extract(doc['content'])
else:
with open(doc['path'], 'rb') as f:
return extractor.extract(f.read())
elif doc['extension'] == '.pdf':
extractor = PDFExtractor()
if doc['type'] == 'web':
return extractor.extract(doc['content'])
else:
return extractor.extract(doc['path'])
elif doc['extension'] in ['.txt', '.md', '.py', '.java', '.cpp',
'.c', '.js', '.ts', '.json', '.xml', '.csv']:
if doc['type'] == 'web':
return doc['content'].decode('utf-8', errors='ignore')
else:
with open(doc['path'], 'r', encoding='utf-8', errors='ignore') as f:
return f.read()
return ""
The learning orchestrator maintains a state file that records processed documents with their content hashes and processing timestamps. When learning mode activates, it computes hashes for discovered documents and compares them against the state file. Only new or changed documents proceed through the processing pipeline. This incremental approach prevents redundant processing and keeps the knowledge base current.
For each document, the orchestrator extracts text using the appropriate extractor, chunks the content with the intelligent chunker, and stores chunks in the vector database. It then extracts entities from each chunk, creates or updates entity nodes in the graph, and establishes co-occurrence relationships between entities that appear together. This creates a rich knowledge graph that captures both explicit relationships and implicit associations.
BUILDING THE CONVERSATIONAL INTERFACE
The conversational interface manages the dual-mode operation and coordinates between user input, retrieval, and response generation. It implements a state machine that transitions between chat and learning modes based on input patterns.
The main application class integrates all components and provides the user interface:
class SelfLearningChatbot:
def __init__(self, model_path, persist_dir="./knowledge_base"):
self.persist_dir = Path(persist_dir)
self.persist_dir.mkdir(exist_ok=True)
print("Initializing LLM backend...")
self.llm = LLMBackend(model_path)
print("Loading embedding model...")
from sentence_transformers import SentenceTransformer
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
print("Initializing vector store...")
self.vector_store = VectorStore(
str(self.persist_dir / "vectors"),
self.embedding_model
)
print("Initializing graph store...")
self.graph_store = GraphStore(
str(self.persist_dir / "graph.json")
)
print("Loading entity extractor...")
self.entity_extractor = EntityExtractor()
print("Initializing chunker...")
self.chunker = IntelligentChunker(self.embedding_model)
print("Initializing document finder...")
self.document_finder = DocumentFinder()
print("Initializing reranker...")
self.reranker = Reranker()
print("Initializing retrieval engine...")
self.retrieval_engine = RetrievalEngine(
self.vector_store,
self.graph_store,
self.entity_extractor,
self.embedding_model,
self.reranker
)
print("Initializing learning orchestrator...")
self.learning_orchestrator = LearningOrchestrator(
self.vector_store,
self.graph_store,
self.entity_extractor,
self.document_finder,
self.chunker
)
self.mode = "chat"
self.conversation_history = []
print("Chatbot ready!")
def process_input(self, user_input):
if user_input.lower().startswith("learn:"):
return self._handle_learning(user_input)
else:
return self._handle_chat(user_input)
def _handle_learning(self, user_input):
self.mode = "learning"
topic_spec = user_input[6:].strip()
parts = topic_spec.split('|')
topics = [p.strip() for p in parts[0].split(',')]
search_paths = []
search_urls = []
for part in parts[1:]:
part = part.strip()
if part.startswith('path:'):
search_paths.append(part[5:].strip())
elif part.startswith('url:'):
search_urls.append(part[4:].strip())
self.learning_orchestrator.learn(topics, search_paths, search_urls)
self.mode = "chat"
return f"Learning complete! I have processed documents related to: {', '.join(topics)}"
def _handle_chat(self, user_input):
self.conversation_history.append({
'role': 'user',
'content': user_input
})
retrieved_context = self.retrieval_engine.retrieve(user_input, top_k=3)
context_text = ""
if retrieved_context:
context_text = "Relevant information:\n\n"
for i, result in enumerate(retrieved_context):
context_text += f"[{i+1}] {result['text']}\n\n"
prompt = self._build_prompt(user_input, context_text)
response = self.llm.generate(
prompt,
max_tokens=512,
temperature=0.7,
stop_sequences=["\nUser:", "\nHuman:"]
)
self.conversation_history.append({
'role': 'assistant',
'content': response
})
if len(self.conversation_history) > 10:
self.conversation_history = self.conversation_history[-10:]
return response
def _build_prompt(self, user_input, context):
history_text = ""
for msg in self.conversation_history[-6:]:
role = "User" if msg['role'] == 'user' else "Assistant"
history_text += f"{role}: {msg['content']}\n"
prompt = f"""You are a helpful AI assistant. Use the provided context to answer questions accurately.
{context}
Conversation history: {history_text}
User: {user_input}
Assistant: """
return prompt
def run_interactive(self):
print("\n" + "="*80)
print("Self-Learning Chatbot - Interactive Mode")
print("="*80)
print("\nCommands:")
print(" - Type your question to chat")
print(" - Type 'learn: <topics> | path: <path> | url: <url>' to learn")
print(" - Type 'quit' to exit")
print("\n" + "="*80 + "\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit', 'bye']:
print("Goodbye!")
break
if not user_input:
continue
response = self.process_input(user_input)
print(f"\nAssistant: {response}\n")
COMPLETE RUNNING EXAMPLE WITH PRODUCTION-READY CODE
The following complete implementation integrates all components into a production-ready system. This code includes error handling, logging, configuration management, and optimizations not shown in the earlier snippets.
import os
import platform
import subprocess
import hashlib
import json
import re
import ast
from pathlib import Path
from datetime import datetime
from urllib.parse import urlparse, urljoin
from collections import Counter
import logging
import numpy as np
import requests
from bs4 import BeautifulSoup
import fitz
import spacy
import networkx as nx
import chromadb
from chromadb.config import Settings
from llama_cpp import Llama
from sentence_transformers import SentenceTransformer, CrossEncoder
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class LLMBackend:
def __init__(self, model_path, context_size=4096):
if not os.path.exists(model_path):
raise FileNotFoundError(f"Model file not found: {model_path}")
self.model_path = model_path
self.context_size = context_size
self.device_type = self._detect_gpu()
logger.info(f"Detected GPU type: {self.device_type}")
self.model = self._initialize_model()
def _detect_gpu(self):
system = platform.system()
if system == "Darwin":
return "metal"
try:
result = subprocess.run(
['nvidia-smi'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
return "cuda"
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
try:
result = subprocess.run(
['rocm-smi'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
return "rocm"
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
if os.path.exists('/dev/dri'):
return "vulkan"
return "cpu"
def _initialize_model(self):
n_gpu_layers = -1 if self.device_type != "cpu" else 0
return Llama(
model_path=self.model_path,
n_ctx=self.context_size,
n_gpu_layers=n_gpu_layers,
n_threads=os.cpu_count() if self.device_type == "cpu" else 4,
verbose=False
)
def generate(self, prompt, max_tokens=512, temperature=0.7,
top_p=0.9, stop_sequences=None):
if stop_sequences is None:
stop_sequences = []
try:
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
stop=stop_sequences,
echo=False
)
if response and 'choices' in response and len(response['choices']) > 0:
return response['choices'][0]['text'].strip()
else:
return "I apologize, but I could not generate a response."
except Exception as e:
logger.error(f"Error generating response: {str(e)}")
return "I apologize, but I encountered an error generating a response."
class DocumentFinder:
def __init__(self):
self.supported_extensions = {
'.txt', '.md', '.pdf', '.html', '.htm',
'.py', '.java', '.cpp', '.c', '.js', '.ts',
'.json', '.xml', '.csv'
}
def find_local_documents(self, root_path, max_depth=5):
documents = []
root = Path(root_path)
if not root.exists():
logger.warning(f"Path does not exist: {root_path}")
return documents
try:
for path in root.rglob('*'):
if path.is_file() and path.suffix.lower() in self.supported_extensions:
if self._get_depth(root, path) <= max_depth:
documents.append({
'path': str(path),
'type': 'local',
'extension': path.suffix.lower()
})
except Exception as e:
logger.error(f"Error scanning directory {root_path}: {str(e)}")
logger.info(f"Found {len(documents)} local documents")
return documents
def _get_depth(self, root, path):
try:
relative = path.relative_to(root)
return len(relative.parts) - 1
except ValueError:
return 0
def find_web_documents(self, seed_urls, max_pages=50):
documents = []
visited = set()
to_visit = list(seed_urls)
while to_visit and len(documents) < max_pages:
url = to_visit.pop(0)
if url in visited:
continue
visited.add(url)
try:
response = requests.get(url, timeout=10, headers={
'User-Agent': 'Mozilla/5.0 (compatible; SelfLearningBot/1.0)'
})
response.raise_for_status()
content_type = response.headers.get('content-type', '')
if 'text/html' in content_type:
documents.append({
'path': url,
'type': 'web',
'extension': '.html',
'content': response.content
})
soup = BeautifulSoup(response.content, 'html.parser')
base_domain = urlparse(url).netloc
for link in soup.find_all('a', href=True):
absolute_url = urljoin(url, link['href'])
link_domain = urlparse(absolute_url).netloc
if link_domain == base_domain and absolute_url not in visited:
to_visit.append(absolute_url)
elif 'application/pdf' in content_type:
documents.append({
'path': url,
'type': 'web',
'extension': '.pdf',
'content': response.content
})
except Exception as e:
logger.warning(f"Error fetching {url}: {str(e)}")
continue
logger.info(f"Found {len(documents)} web documents")
return documents
class HTMLExtractor:
def extract(self, html_content):
try:
soup = BeautifulSoup(html_content, 'html.parser')
for element in soup(['script', 'style', 'nav', 'footer',
'header', 'aside', 'form']):
element.decompose()
main_content = soup.find('main') or soup.find('article') or soup.body
if not main_content:
return ""
text_blocks = []
for element in main_content.find_all(['p', 'h1', 'h2', 'h3',
'h4', 'h5', 'h6', 'li']):
text = element.get_text(strip=True)
if len(text) > 20:
text_blocks.append(text)
return '\n\n'.join(text_blocks)
except Exception as e:
logger.error(f"Error extracting HTML: {str(e)}")
return ""
class PDFExtractor:
def extract(self, pdf_path_or_bytes):
try:
if isinstance(pdf_path_or_bytes, bytes):
doc = fitz.open(stream=pdf_path_or_bytes, filetype="pdf")
else:
doc = fitz.open(pdf_path_or_bytes)
text_blocks = []
for page_num in range(len(doc)):
page = doc[page_num]
blocks = page.get_text("blocks")
for block in blocks:
if block[6] == 0:
text = block[4].strip()
if len(text) > 20:
text_blocks.append(text)
doc.close()
return '\n\n'.join(text_blocks)
except Exception as e:
logger.error(f"Error extracting PDF: {str(e)}")
return ""
class IntelligentChunker:
def __init__(self, embedding_model):
self.embedding_model = embedding_model
self.min_chunk_size = 100
self.max_chunk_size = 1000
self.similarity_threshold = 0.7
def chunk_text(self, text, content_type='text'):
if content_type == 'code':
return self._chunk_code(text)
else:
return self._chunk_semantic(text)
def _chunk_semantic(self, text):
sentences = self._split_sentences(text)
if len(sentences) == 0:
return []
if len(sentences) == 1:
return [sentences[0]] if len(sentences[0]) >= self.min_chunk_size else []
try:
embeddings = self.embedding_model.encode(sentences)
except Exception as e:
logger.error(f"Error encoding sentences: {str(e)}")
return self._fallback_chunk(text)
chunks = []
current_chunk = [sentences[0]]
current_size = len(sentences[0])
for i in range(1, len(sentences)):
try:
similarity = np.dot(embeddings[i-1], embeddings[i]) / (
np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i]) + 1e-10
)
except Exception:
similarity = 0.5
sentence_len = len(sentences[i])
if (similarity < self.similarity_threshold or
current_size + sentence_len > self.max_chunk_size):
if current_size >= self.min_chunk_size:
chunks.append(' '.join(current_chunk))
current_chunk = [sentences[i]]
current_size = sentence_len
else:
current_chunk.append(sentences[i])
current_size += sentence_len
else:
current_chunk.append(sentences[i])
current_size += sentence_len
if current_chunk and sum(len(s) for s in current_chunk) >= self.min_chunk_size:
chunks.append(' '.join(current_chunk))
return chunks
def _split_sentences(self, text):
sentence_endings = re.compile(r'(?<=[.!?])\s+(?=[A-Z])')
sentences = sentence_endings.split(text)
return [s.strip() for s in sentences if s.strip()]
def _chunk_code(self, code):
try:
tree = ast.parse(code)
chunks = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
start_line = node.lineno - 1
end_line = node.end_lineno
code_lines = code.split('\n')
chunk = '\n'.join(code_lines[start_line:end_line])
if len(chunk) >= self.min_chunk_size:
chunks.append(chunk)
return chunks if chunks else self._fallback_chunk(code)
except SyntaxError:
return self._fallback_chunk(code)
def _fallback_chunk(self, text):
chunks = []
words = text.split()
current_chunk = []
current_size = 0
for word in words:
word_len = len(word) + 1
if current_size + word_len > self.max_chunk_size and current_chunk:
chunk_text = ' '.join(current_chunk)
if len(chunk_text) >= self.min_chunk_size:
chunks.append(chunk_text)
current_chunk = [word]
current_size = word_len
else:
current_chunk.append(word)
current_size += word_len
if current_chunk:
chunk_text = ' '.join(current_chunk)
if len(chunk_text) >= self.min_chunk_size:
chunks.append(chunk_text)
return chunks
class VectorStore:
def __init__(self, persist_directory, embedding_model):
self.persist_directory = persist_directory
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_directory
))
self.embedding_model = embedding_model
self.collection = self.client.get_or_create_collection(
name="documents",
metadata={"hnsw:space": "cosine"}
)
def add_chunks(self, chunks, metadata_list):
if not chunks:
return []
try:
embeddings = self.embedding_model.encode(chunks).tolist()
except Exception as e:
logger.error(f"Error encoding chunks: {str(e)}")
return []
ids = []
new_chunks = []
new_embeddings = []
new_metadata = []
for i, (chunk, embedding, metadata) in enumerate(
zip(chunks, embeddings, metadata_list)
):
chunk_hash = self._compute_hash(chunk)
try:
existing = self.collection.get(
where={"content_hash": chunk_hash}
)
except Exception as e:
logger.warning(f"Error checking for existing chunk: {str(e)}")
existing = {'ids': []}
if not existing['ids']:
chunk_id = f"chunk_{chunk_hash}_{i}"
ids.append(chunk_id)
new_chunks.append(chunk)
new_embeddings.append(embedding)
metadata['content_hash'] = chunk_hash
new_metadata.append(metadata)
if new_chunks:
try:
self.collection.add(
ids=ids,
embeddings=new_embeddings,
documents=new_chunks,
metadatas=new_metadata
)
self.client.persist()
logger.info(f"Added {len(new_chunks)} new chunks to vector store")
except Exception as e:
logger.error(f"Error adding chunks to vector store: {str(e)}")
return []
return ids
def _compute_hash(self, text):
return hashlib.sha256(text.encode()).hexdigest()
def search(self, query, n_results=5, filter_metadata=None):
try:
query_embedding = self.embedding_model.encode([query]).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=n_results,
where=filter_metadata
)
return results
except Exception as e:
logger.error(f"Error searching vector store: {str(e)}")
return {'ids': [[]], 'documents': [[]], 'distances': [[]], 'metadatas': [[]]}
class GraphStore:
def __init__(self, persist_path):
self.persist_path = Path(persist_path)
self.graph = self._load_graph()
def _load_graph(self):
if self.persist_path.exists():
try:
with open(self.persist_path, 'r') as f:
data = json.load(f)
return nx.node_link_graph(data)
except Exception as e:
logger.error(f"Error loading graph: {str(e)}")
return nx.DiGraph()
return nx.DiGraph()
def save(self):
try:
self.persist_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.persist_path, 'w') as f:
data = nx.node_link_data(self.graph)
json.dump(data, f)
logger.info("Graph saved successfully")
except Exception as e:
logger.error(f"Error saving graph: {str(e)}")
def add_entity(self, entity_name, entity_type, embedding_ids=None):
if embedding_ids is None:
embedding_ids = []
if self.graph.has_node(entity_name):
existing_ids = self.graph.nodes[entity_name].get('embedding_ids', [])
embedding_ids = list(set(existing_ids + embedding_ids))
self.graph.add_node(
entity_name,
type=entity_type,
embedding_ids=embedding_ids
)
def add_relationship(self, source, target, relation_type):
self.graph.add_edge(source, target, type=relation_type)
def get_entity(self, entity_name):
if self.graph.has_node(entity_name):
return self.graph.nodes[entity_name]
return None
def get_related_entities(self, entity_name, max_depth=2):
if not self.graph.has_node(entity_name):
return []
related = set()
try:
for node in nx.single_source_shortest_path_length(
self.graph, entity_name, cutoff=max_depth
).keys():
if node != entity_name:
related.add(node)
except Exception as e:
logger.error(f"Error finding related entities: {str(e)}")
return list(related)
def get_embedding_ids_for_entity(self, entity_name):
entity = self.get_entity(entity_name)
if entity:
return entity.get('embedding_ids', [])
return []
class EntityExtractor:
def __init__(self):
try:
self.nlp = spacy.load("en_core_web_sm")
except OSError:
logger.warning("Spacy model not found. Downloading...")
os.system("python -m spacy download en_core_web_sm")
self.nlp = spacy.load("en_core_web_sm")
def extract_entities(self, text):
try:
max_length = self.nlp.max_length
text_to_process = text[:max_length] if len(text) > max_length else text
doc = self.nlp(text_to_process)
entities = []
for ent in doc.ents:
if ent.label_ in ['PERSON', 'ORG', 'GPE', 'PRODUCT', 'EVENT']:
entities.append({
'text': ent.text,
'type': ent.label_,
'start': ent.start_char,
'end': ent.end_char
})
noun_chunks = [chunk.text for chunk in doc.noun_chunks
if len(chunk.text.split()) <= 3]
chunk_freq = Counter(noun_chunks)
for chunk, freq in chunk_freq.most_common(10):
if freq > 1:
entities.append({
'text': chunk,
'type': 'CONCEPT',
'start': -1,
'end': -1
})
return entities
except Exception as e:
logger.error(f"Error extracting entities: {str(e)}")
return []
class Reranker:
def __init__(self, model_name='cross-encoder/ms-marco-MiniLM-L-6-v2'):
try:
self.model = CrossEncoder(model_name)
except Exception as e:
logger.error(f"Error loading reranker model: {str(e)}")
self.model = None
def predict(self, query_document_pairs):
if self.model is None:
return [0.5] * len(query_document_pairs)
try:
batch_size = 32
all_scores = []
for i in range(0, len(query_document_pairs), batch_size):
batch = query_document_pairs[i:i + batch_size]
scores = self.model.predict(batch)
all_scores.extend(scores)
return all_scores
except Exception as e:
logger.error(f"Error reranking: {str(e)}")
return [0.5] * len(query_document_pairs)
class RetrievalEngine:
def __init__(self, vector_store, graph_store, entity_extractor,
embedding_model, reranker_model):
self.vector_store = vector_store
self.graph_store = graph_store
self.entity_extractor = entity_extractor
self.embedding_model = embedding_model
self.reranker_model = reranker_model
def retrieve(self, query, top_k=5):
vector_results = self._vector_search(query, top_k * 3)
graph_results = self._graph_search(query, top_k * 3)
all_candidates = self._merge_results(vector_results, graph_results)
if not all_candidates:
return []
reranked = self._rerank(query, all_candidates, top_k)
return reranked
def _vector_search(self, query, k):
results = self.vector_store.search(query, n_results=k)
candidates = []
if results['ids'] and results['ids'][0]:
for i in range(len(results['ids'][0])):
candidates.append({
'id': results['ids'][0][i],
'text': results['documents'][0][i],
'score': 1.0 - results['distances'][0][i],
'metadata': results['metadatas'][0][i],
'source': 'vector'
})
return candidates
def _graph_search(self, query, k):
entities = self.entity_extractor.extract_entities(query)
if not entities:
return []
embedding_ids = set()
for entity in entities:
entity_name = entity['text']
direct_ids = self.graph_store.get_embedding_ids_for_entity(
entity_name
)
embedding_ids.update(direct_ids)
related = self.graph_store.get_related_entities(entity_name)
for related_entity in related:
related_ids = self.graph_store.get_embedding_ids_for_entity(
related_entity
)
embedding_ids.update(related_ids)
candidates = []
for emb_id in list(embedding_ids)[:k]:
try:
result = self.vector_store.collection.get(ids=[emb_id])
if result['ids']:
candidates.append({
'id': emb_id,
'text': result['documents'][0],
'score': 0.8,
'metadata': result['metadatas'][0],
'source': 'graph'
})
except Exception as e:
logger.warning(f"Error retrieving embedding {emb_id}: {str(e)}")
continue
return candidates
def _merge_results(self, vector_results, graph_results):
seen_ids = set()
merged = []
for result in vector_results + graph_results:
if result['id'] not in seen_ids:
seen_ids.add(result['id'])
merged.append(result)
return merged
def _rerank(self, query, candidates, top_k):
if not candidates:
return []
texts = [c['text'] for c in candidates]
pairs = [[query, text] for text in texts]
scores = self.reranker_model.predict(pairs)
for i, candidate in enumerate(candidates):
candidate['rerank_score'] = float(scores[i])
candidates.sort(key=lambda x: x['rerank_score'], reverse=True)
return candidates[:top_k]
class LearningOrchestrator:
def __init__(self, vector_store, graph_store, entity_extractor,
document_finder, chunker):
self.vector_store = vector_store
self.graph_store = graph_store
self.entity_extractor = entity_extractor
self.document_finder = document_finder
self.chunker = chunker
self.state_file = Path("learning_state.json")
self.processed_docs = self._load_state()
def _load_state(self):
if self.state_file.exists():
try:
with open(self.state_file, 'r') as f:
return json.load(f)
except Exception as e:
logger.error(f"Error loading state: {str(e)}")
return {}
return {}
def _save_state(self):
try:
with open(self.state_file, 'w') as f:
json.dump(self.processed_docs, f, indent=2)
except Exception as e:
logger.error(f"Error saving state: {str(e)}")
def learn(self, topics, search_paths=None, search_urls=None):
documents = []
if search_paths:
for path in search_paths:
docs = self.document_finder.find_local_documents(path)
documents.extend(docs)
if search_urls:
docs = self.document_finder.find_web_documents(search_urls)
documents.extend(docs)
new_or_changed = self._filter_documents(documents)
logger.info(f"Processing {len(new_or_changed)} new or changed documents...")
for doc in new_or_changed:
self._process_document(doc, topics)
self.graph_store.save()
self._save_state()
logger.info(f"Learning complete. Processed {len(new_or_changed)} documents.")
def _filter_documents(self, documents):
new_or_changed = []
for doc in documents:
doc_id = doc['path']
if doc['type'] == 'local':
current_hash = self._compute_file_hash(doc['path'])
else:
current_hash = self._compute_content_hash(doc.get('content', b''))
if doc_id not in self.processed_docs:
new_or_changed.append(doc)
self.processed_docs[doc_id] = {
'hash': current_hash,
'processed_at': datetime.now().isoformat()
}
elif self.processed_docs[doc_id]['hash'] != current_hash:
new_or_changed.append(doc)
self.processed_docs[doc_id]['hash'] = current_hash
self.processed_docs[doc_id]['processed_at'] = datetime.now().isoformat()
return new_or_changed
def _compute_file_hash(self, filepath):
hasher = hashlib.sha256()
try:
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hasher.update(chunk)
return hasher.hexdigest()
except Exception as e:
logger.error(f"Error hashing file {filepath}: {str(e)}")
return ""
def _compute_content_hash(self, content):
return hashlib.sha256(content).hexdigest()
def _process_document(self, doc, topics):
text = self._extract_text(doc)
if not text:
logger.warning(f"No text extracted from {doc['path']}")
return
content_type = 'code' if doc['extension'] in ['.py', '.java', '.cpp',
'.c', '.js', '.ts'] else 'text'
chunks = self.chunker.chunk_text(text, content_type)
if not chunks:
logger.warning(f"No chunks created from {doc['path']}")
return
metadata_list = []
for i, chunk in enumerate(chunks):
metadata = {
'source': doc['path'],
'chunk_index': i,
'topics': ','.join(topics),
'content_type': content_type
}
metadata_list.append(metadata)
chunk_ids = self.vector_store.add_chunks(chunks, metadata_list)
for chunk_id, chunk in zip(chunk_ids, chunks):
entities = self.entity_extractor.extract_entities(chunk)
for entity in entities:
self.graph_store.add_entity(
entity['text'],
entity['type'],
[chunk_id]
)
for i in range(len(entities)):
for j in range(i + 1, len(entities)):
self.graph_store.add_relationship(
entities[i]['text'],
entities[j]['text'],
'co-occurs'
)
def _extract_text(self, doc):
try:
if doc['extension'] in ['.html', '.htm']:
extractor = HTMLExtractor()
if doc['type'] == 'web':
return extractor.extract(doc['content'])
else:
with open(doc['path'], 'rb') as f:
return extractor.extract(f.read())
elif doc['extension'] == '.pdf':
extractor = PDFExtractor()
if doc['type'] == 'web':
return extractor.extract(doc['content'])
else:
return extractor.extract(doc['path'])
elif doc['extension'] in ['.txt', '.md', '.py', '.java', '.cpp',
'.c', '.js', '.ts', '.json', '.xml', '.csv']:
if doc['type'] == 'web':
return doc['content'].decode('utf-8', errors='ignore')
else:
with open(doc['path'], 'r', encoding='utf-8', errors='ignore') as f:
return f.read()
return ""
except Exception as e:
logger.error(f"Error extracting text from {doc['path']}: {str(e)}")
return ""
class SelfLearningChatbot:
def __init__(self, model_path, persist_dir="./knowledge_base"):
self.persist_dir = Path(persist_dir)
self.persist_dir.mkdir(exist_ok=True)
logger.info("Initializing LLM backend...")
self.llm = LLMBackend(model_path)
logger.info("Loading embedding model...")
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
logger.info("Initializing vector store...")
self.vector_store = VectorStore(
str(self.persist_dir / "vectors"),
self.embedding_model
)
logger.info("Initializing graph store...")
self.graph_store = GraphStore(
str(self.persist_dir / "graph.json")
)
logger.info("Loading entity extractor...")
self.entity_extractor = EntityExtractor()
logger.info("Initializing chunker...")
self.chunker = IntelligentChunker(self.embedding_model)
logger.info("Initializing document finder...")
self.document_finder = DocumentFinder()
logger.info("Initializing reranker...")
self.reranker = Reranker()
logger.info("Initializing retrieval engine...")
self.retrieval_engine = RetrievalEngine(
self.vector_store,
self.graph_store,
self.entity_extractor,
self.embedding_model,
self.reranker
)
logger.info("Initializing learning orchestrator...")
self.learning_orchestrator = LearningOrchestrator(
self.vector_store,
self.graph_store,
self.entity_extractor,
self.document_finder,
self.chunker
)
self.mode = "chat"
self.conversation_history = []
logger.info("Chatbot ready!")
def process_input(self, user_input):
if user_input.lower().startswith("learn:"):
return self._handle_learning(user_input)
else:
return self._handle_chat(user_input)
def _handle_learning(self, user_input):
self.mode = "learning"
topic_spec = user_input[6:].strip()
parts = topic_spec.split('|')
topics = [p.strip() for p in parts[0].split(',')]
search_paths = []
search_urls = []
for part in parts[1:]:
part = part.strip()
if part.startswith('path:'):
search_paths.append(part[5:].strip())
elif part.startswith('url:'):
search_urls.append(part[4:].strip())
self.learning_orchestrator.learn(topics, search_paths, search_urls)
self.mode = "chat"
return f"Learning complete! I have processed documents related to: {', '.join(topics)}"
def _handle_chat(self, user_input):
self.conversation_history.append({
'role': 'user',
'content': user_input
})
retrieved_context = self.retrieval_engine.retrieve(user_input, top_k=3)
context_text = ""
if retrieved_context:
context_text = "Relevant information:\n\n"
for i, result in enumerate(retrieved_context):
context_text += f"[{i+1}] {result['text']}\n\n"
prompt = self._build_prompt(user_input, context_text)
response = self.llm.generate(
prompt,
max_tokens=512,
temperature=0.7,
stop_sequences=["\nUser:", "\nHuman:"]
)
self.conversation_history.append({
'role': 'assistant',
'content': response
})
if len(self.conversation_history) > 10:
self.conversation_history = self.conversation_history[-10:]
return response
def _build_prompt(self, user_input, context):
history_text = ""
if len(self.conversation_history) > 1:
for msg in self.conversation_history[-7:-1]:
role = "User" if msg['role'] == 'user' else "Assistant"
history_text += f"{role}: {msg['content']}\n"
prompt = f"""You are a helpful AI assistant. Use the provided context to answer questions accurately.
{context}
Conversation history: {history_text}
User: {user_input}
"""
return prompt
def run_interactive(self):
print("\n" + "="*80)
print("Self-Learning Chatbot - Interactive Mode")
print("="*80)
print("\nCommands:")
print(" - Type your question to chat")
print(" - Type 'learn: <topics> | path: <path> | url: <url>' to learn")
print(" - Type 'quit' to exit")
print("\n" + "="*80 + "\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit', 'bye']:
print("Goodbye!")
break
if not user_input:
continue
response = self.process_input(user_input)
if response:
print(f"\nAssistant: {response}\n")
else:
print("\nAssistant: I apologize, but I could not generate a response.\n")
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
logger.error(f"Error processing input: {str(e)}")
print(f"\nError: {str(e)}\n")
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python chatbot.py <path_to_llm_model>")
print("Example: python chatbot.py ./models/llama-2-7b-chat.gguf")
sys.exit(1)
model_path = sys.argv[1]
if not os.path.exists(model_path):
print(f"Error: Model file not found: {model_path}")
sys.exit(1)
try:
chatbot = SelfLearningChatbot(model_path)
chatbot.run_interactive()
except Exception as e:
logger.error(f"Fatal error: {str(e)}")
print(f"Fatal error: {str(e)}")
sys.exit(1)
DEPLOYMENT AND USAGE CONSIDERATIONS
The complete implementation provides a production-ready self-learning chatbot with robust error handling, logging, and state management. To deploy this system, you need to install the required dependencies and download necessary models.
The Python dependencies include llama-cpp-python for LLM inference, sentence-transformers for embeddings and reranking, spacy for entity extraction, chromadb for vector storage, networkx for graph operations, beautifulsoup4 for HTML parsing, and pymupdf for PDF extraction. Install these with pip install llama-cpp-python sentence-transformers spacy chromadb networkx beautifulsoup4 pymupdf requests. Additionally, download the spaCy language model with python -m spacy download en_core_web_sm.
For the LLM model, download a quantized GGUF format model compatible with llama.cpp. Popular options include Llama 2, Mistral, or Phi models in 7B or 13B parameter sizes. Quantized versions like Q4_K_M or Q5_K_M provide good quality while maintaining reasonable memory requirements. Place the model file in an accessible location and provide its path when initializing the chatbot.
To use the system, run the script with the model path as an argument. For example, python chatbot.py ./models/llama-2-7b-chat.Q4_K_M.gguf. The chatbot will initialize all components and enter interactive mode. You can then type questions to chat or trigger learning mode with commands like learn: machine learning, deep learning | path: ./documents | url: https://example.com/ml-tutorial.
The learning mode specification uses a pipe-delimited format where the first segment contains comma-separated topics, and subsequent segments specify data sources with path: or url: prefixes. The system will discover documents in the specified locations, filter for new or changed content, process them through the intelligent pipeline, and store extracted knowledge in both vector and graph databases.
After learning, the chatbot can answer questions using the acquired knowledge. The retrieval engine performs hybrid search across both knowledge stores, combining semantic similarity with graph traversal to find relevant information. The reranker ensures the most relevant chunks appear first in the context provided to the LLM.
The system handles incremental updates automatically. When you trigger learning mode with the same topics and sources, it computes content hashes for discovered documents and compares them against the state file. Only new or modified documents undergo processing, preventing redundant work and maintaining knowledge base consistency.
For production deployments, consider implementing additional features like user authentication, conversation persistence, multi-user support, and API endpoints for integration with other systems. The modular architecture facilitates these extensions without requiring fundamental restructuring.
Performance optimization opportunities include batch processing for large document collections, parallel embedding generation, and caching frequently accessed graph traversals. The vector database supports efficient similarity search through hierarchical navigable small world graphs, while the graph database benefits from indexing frequently queried entity types.
Security considerations include input validation to prevent injection attacks, rate limiting to prevent abuse, and sandboxing for code execution if processing untrusted documents. The system should run with minimal privileges and store sensitive data encrypted at rest.
Monitoring and observability help maintain system health in production. The logging framework provides visibility into component operations, error conditions, and performance metrics. Consider integrating with monitoring tools to track query latency, retrieval quality, and resource utilization.
The hybrid RAG architecture represents a significant advancement over traditional retrieval systems. By combining vector similarity with graph relationships, the system can discover relevant information through multiple pathways. This redundancy improves recall while the reranking stage maintains precision. The bidirectional references between vector and graph stores enable sophisticated query strategies that leverage both semantic and structural information.
This implementation demonstrates how local LLMs and open source components can create sophisticated AI systems without relying on cloud services. The multi-platform GPU support ensures the system runs efficiently on diverse hardware configurations. The intelligent processing pipeline adapts to different document types and content characteristics, maximizing knowledge extraction quality.
The self-learning capability transforms the chatbot from a static system into a dynamic knowledge platform that grows with use. As users identify new topics and sources, the system continuously expands its knowledge base while maintaining consistency through deduplication and incremental updates. This creates a virtuous cycle where increased usage leads to improved capabilities.
