Bias, Hallucinations and Limited Context Memory - LLM Challenge Accepted

ADDRESSING LLM CHALLENGES

Large Language Models, or LLMs, have revolutionized how we interact with information and automate complex tasks, offering unprecedented capabilities in natural language understanding and generation. However, these powerful systems are not without their limitations. To effectively harness the full potential of LLMs in real-world applications, it is crucial to understand and systematically address their inherent challenges. This article delves into three primary obstacles: constrained context window sizes, the presence of bias, and the phenomenon of hallucinations, outlining practical strategies and advanced techniques to mitigate these issues.

CHALLENGE 1: CONSTRAINED CONTEXT WINDOW SIZES

One of the most immediate practical limitations of current LLMs is their constrained context window size. This refers to the maximum amount of text, including both the input prompt and the generated response, that an LLM can process and retain at any given time. Exceeding this limit leads to the model forgetting earlier parts of a conversation or document, severely hindering its ability to perform complex, multi-turn interactions or analyze lengthy texts comprehensively.

Retrieval Augmented Generation (RAG) and GraphRAG

To overcome the context window limitation, Retrieval Augmented Generation, or RAG, has emerged as a powerful paradigm. RAG systems augment an LLM's knowledge by first retrieving relevant information from an external, up-to-date knowledge base before generating a response. This process ensures that the LLM has access to specific, grounded facts pertinent to the user's query, rather than relying solely on its potentially outdated or generalized internal training data.

A conceptual Python function illustrating the RAG workflow might look like this:

# This function simulates a RAG process by first retrieving information
# and then using it to augment the LLM's prompt.
def perform_rag_query(user_query, knowledge_base_api, llm_api):
    # Step 1: Retrieve relevant documents or snippets from the knowledge base.
    # In a real system, this would involve vector similarity search or keyword matching.
    print("Retrieving relevant information from the knowledge base...")
    retrieved_info = knowledge_base_api.search(query=user_query)
    
    # Step 2: Construct an augmented prompt for the LLM.
    # This prompt includes both the original user query and the retrieved context.
    augmented_prompt = f"""
    Based on the following information:
    ---
    {retrieved_info}
    ---
    
    Please answer the following question: "{user_query}"
    """
    
    # Step 3: Send the augmented prompt to the LLM for generation.
    print("Sending augmented prompt to LLM...")
    llm_response = llm_api.generate_text(prompt=augmented_prompt)
    
    return llm_response

# Example conceptual usage:
# class MockKnowledgeBase:
#     def search(self, query):
#         if "quantum computing" in query.lower():
#             return "Quantum computing uses quantum-mechanical phenomena..."
#         return "No specific information found."
#
# class MockLLM:
#     def generate_text(self, prompt):
#         return f"LLM generated response based on: {prompt[:50]}..."
#
# kb_api = MockKnowledgeBase()
# llm_api = MockLLM()
#
# response = perform_rag_query("What is quantum computing?", kb_api, llm_api)
# print(response)

Building upon RAG, GraphRAG leverages the structured power of knowledge graphs. In GraphRAG, information is stored not just as raw text but as interconnected nodes and edges representing entities and their relationships. When a query is made, the system can traverse the graph to retrieve not only direct matches but also related concepts and their bidirectional links, providing a richer, more contextually aware set of information to the LLM. This allows for a deeper understanding of complex relationships that might be missed by simple text retrieval.

A conceptual function for querying a knowledge graph within a GraphRAG setup might involve:

# This function simulates querying a knowledge graph to retrieve interconnected
# information relevant to a user's query.
def query_knowledge_graph_for_rag(user_query, graph_db_connector):
    # In a real scenario, this would involve SPARQL queries or graph traversal algorithms.
    print(f"Querying knowledge graph for '{user_query}'...")
    
    # Simulate retrieving nodes and their linked relationships
    # based on the query.
    relevant_nodes = graph_db_connector.find_nodes(keywords=user_query)
    contextual_data = []
    for node in relevant_nodes:
        contextual_data.append(f"Node: {node.name}, Type: {node.type}")
        # Simulate retrieving connected nodes and their relationship types
        for relationship in node.get_relationships():
            contextual_data.append(f"  - {node.name} {relationship.type} {relationship.target_node.name}")
    
    return "\n".join(contextual_data)

# Example conceptual usage:
# class MockGraphDB:
#     def find_nodes(self, keywords):
#         class Node:
#             def __init__(self, name, type):
#                 self.name = name
#                 self.type = type
#                 self.relationships = []
#             def add_relationship(self, rel_type, target_node):
#                 class Relationship:
#                     def __init__(self, type, target_node):
#                         self.type = type
#                         self.target_node = target_node
#                     
#                 self.relationships.append(Relationship(rel_type, target_node))
#             def get_relationships(self):
#                 return self.relationships
#
#         if "technology company" in keywords.lower():
#             company_node = Node("TechCorp Inc.", "Company")
#             ai_node = Node("Artificial Intelligence", "Technology")
#             cloud_node = Node("Cloud Computing", "Technology")
#             company_node.add_relationship("develops", ai_node)
#             company_node.add_relationship("provides", cloud_node)
#             return [company_node]
#         return []
#
# graph_db = MockGraphDB()
# graph_context = query_knowledge_graph_for_rag("technology company products", graph_db)
# print(graph_context)

Multiple Simple Prompts Instead of One Large Complex Prompt

Instead of crafting a single, monolithic prompt for a complex task, breaking the task down into a series of smaller, simpler prompts can be highly effective. Each prompt focuses on a specific sub-task, and the output of one step feeds into the input of the next. This modular approach reduces the cognitive load on the LLM, keeps individual prompt sizes within context limits, and allows for better control and debugging of the overall process. For instance, summarizing a long document could involve a first prompt to extract key entities, a second to identify main themes, and a third to synthesize these into a final summary.

Here is a conceptual illustration of a multi-step prompting approach:

# This function demonstrates breaking down a complex task into sequential,
# simpler LLM calls.
def multi_step_task_processing(initial_data, llm_api):
    # Step 1: Extract key entities from the initial data.
    prompt_step1 = f"Extract the main entities (people, organizations, locations) from the following text: {initial_data}"
    entities = llm_api.generate_text(prompt=prompt_step1)
    print(f"Extracted Entities: {entities}")
    
    # Step 2: Summarize the initial data focusing on these entities.
    prompt_step2 = f"Summarize the following text, specifically mentioning these entities: {entities}\nText: {initial_data}"
    summary = llm_api.generate_text(prompt=prompt_step2)
    print(f"Summary: {summary}")
    
    # Step 3: Generate a follow-up question based on the summary.
    prompt_step3 = f"Based on this summary: '{summary}', suggest one relevant follow-up question."
    follow_up_question = llm_api.generate_text(prompt=prompt_step3)
    print(f"Follow-up Question: {follow_up_question}")
    
    return summary, follow_up_question

# Example conceptual usage:
# class MockLLM:
#     def generate_text(self, prompt):
#         if "main entities" in prompt:
#             return "Dr. Sarah Chen, Global Research Institute, Tokyo"
#         elif "Summarize" in prompt:
#             return "Dr. Sarah Chen from Global Research Institute visited Tokyo to present findings."
#         elif "follow-up question" in prompt:
#             return "What were the key findings presented by Dr. Chen?"
#         return "Mock LLM response."
#
# mock_llm_api = MockLLM()
# long_text = "Dr. Sarah Chen, a leading researcher at the Global Research Institute, traveled to Tokyo last month to present groundbreaking findings on renewable energy systems. The conference was well-attended by international experts."
# multi_step_task_processing(long_text, mock_llm_api)

Using Different (Possibly Fine-tuned) LLMs for Different Purposes

Another strategy involves employing a specialized approach where different LLMs, potentially fine-tuned for specific tasks, handle distinct parts of a workflow. For example, one LLM might be excellent at summarization, another at code generation, and a third at sentiment analysis. By routing tasks to the most appropriate model, one can optimize performance, reduce the risk of context overflow for general-purpose models, and leverage the strengths of each specialized LLM.

A conceptual dispatcher function for routing tasks to specialized LLMs could be structured as follows:

# This function acts as a dispatcher, directing different types of requests
# to specialized LLM instances.
def route_to_specialized_llm(task_type, input_data, llm_registry):
    print(f"Routing task '{task_type}' to appropriate LLM...")
    
    if task_type == "summarization":
        llm_model = llm_registry.get("summarizer_llm")
        if llm_model:
            return llm_model.summarize(input_data)
        else:
            return "Error: Summarizer LLM not found."
    elif task_type == "code_generation":
        llm_model = llm_registry.get("code_gen_llm")
        if llm_model:
            return llm_model.generate_code(input_data)
        else:
            return "Error: Code Generation LLM not found."
    elif task_type == "sentiment_analysis":
        llm_model = llm_registry.get("sentiment_llm")
        if llm_model:
            return llm_model.analyze_sentiment(input_data)
        else:
            return "Error: Sentiment Analysis LLM not found."
    else:
        return "Error: Unknown task type."

# Example conceptual usage:
# class MockSummarizerLLM:
#     def summarize(self, text):
#         return f"Summary of: {text[:30]}..."
#
# class MockCodeGenLLM:
#     def generate_code(self, description):
#         return f"def {description.replace(' ', '_')}(): pass"
#
# class MockSentimentLLM:
#     def analyze_sentiment(self, text):
#         return "Positive" if "excellent" in text else "Neutral"
#
# llm_models = {
#     "summarizer_llm": MockSummarizerLLM(),
#     "code_gen_llm": MockCodeGenLLM(),
#     "sentiment_llm": MockSentimentLLM()
# }
#
# print(route_to_specialized_llm("summarization", "This is a comprehensive research paper on machine learning.", llm_models))
# print(route_to_specialized_llm("code_generation", "create a python function to multiply two numbers", llm_models))

LLMs with Large Context Windows

Recent advancements have led to the development of LLMs with significantly larger context windows, such as Claude, GPT, and Gemini Pro. While these models offer a direct solution to the context limitation, they often come with higher computational costs and may still have practical limits for extremely long inputs. Nonetheless, utilizing these models where feasible can simplify prompt engineering for tasks requiring extensive context.

Sliding Context Windows

For ongoing conversations or processing very long documents chunk by chunk, a sliding context window technique can be employed. This method involves maintaining a fixed-size window of the most recent interactions or document segments. As new information arrives, the oldest information is discarded or summarized to keep the total context within the LLM's limits. This allows for a continuous flow of information without overwhelming the model.

A simple function to manage a sliding context window might operate as follows:

# This function manages a sliding context window for conversational history.
def manage_sliding_context(current_history, new_utterance, max_tokens=1000):
    # Append the new utterance to the history.
    updated_history = current_history + [new_utterance]
    
    # Calculate the approximate token count (very basic estimation here).
    # In a real scenario, a proper tokenizer would be used.
    total_tokens = sum(len(item.split()) for item in updated_history)
    
    # If the history exceeds the max_tokens, remove the oldest items.
    while total_tokens > max_tokens and len(updated_history) > 1:
        print(f"Context window exceeded, removing oldest entry: '{updated_history[0][:20]}...'")
        updated_history.pop(0) # Remove the oldest utterance
        total_tokens = sum(len(item.split()) for item in updated_history) # Recalculate
        
    return updated_history

# Example conceptual usage:
# conversation_history = []
# conversation_history = manage_sliding_context(conversation_history, "User: Hello, how are you?", max_tokens=20)
# conversation_history = manage_sliding_context(conversation_history, "Assistant: I am doing well, thank you!", max_tokens=20)
# conversation_history = manage_sliding_context(conversation_history, "User: What is the weather forecast?", max_tokens=20)
# conversation_history = manage_sliding_context(conversation_history, "Assistant: The weather is sunny today.", max_tokens=20)
# conversation_history = manage_sliding_context(conversation_history, "User: That sounds wonderful!", max_tokens=20)
# print("Final Managed Context:")
# for item in conversation_history:
#     print(item)

Integrating Tools

LLMs can be significantly enhanced by integrating them with external tools. Instead of trying to fit all necessary information into the context window, the LLM can be prompted to decide when to use a tool such as a search engine, a calculator, a database query tool, or an API to retrieve or process information. The tool's output is then fed back into the LLM's context, providing it with the necessary data without consuming excessive context window space for raw information. This approach is fundamental to agents and tool-use frameworks.

A conceptual function illustrating tool integration:

# This function simulates an LLM's ability to call external tools
# and incorporate their results into its reasoning.
def use_tool_for_information(llm_query, available_tools):
    print(f"LLM considering tools for query: '{llm_query}'")
    
    # Simulate LLM deciding which tool to use based on the query.
    if "current stock price" in llm_query.lower():
        tool_name = "stock_checker"
        if tool_name in available_tools:
            print(f"LLM decided to use '{tool_name}' tool.")
            stock_symbol = llm_query.split("for ")[-1].strip("?").upper()
            result = available_tools[tool_name].get_price(stock_symbol)
            return f"The stock price for {stock_symbol} is {result}."
        else:
            return "Stock checker tool not available."
    elif "calculate" in llm_query.lower():
        tool_name = "calculator"
        if tool_name in available_tools:
            print(f"LLM decided to use '{tool_name}' tool.")
            # Simple parsing for demonstration
            expression = llm_query.split("calculate ")[-1].strip("?")
            try:
                result = available_tools[tool_name].evaluate(expression)
                return f"The result of {expression} is {result}."
            except Exception as e:
                return f"Calculation error: {e}"
        else:
            return "Calculator tool not available."
    else:
        return "No suitable tool found for this query."

# Example conceptual usage:
# class MockStockChecker:
#     def get_price(self, symbol):
#         prices = {"AAPL": "175.50 USD", "GOOGL": "142.30 USD", "TSLA": "245.80 USD"}
#         return prices.get(symbol, "N/A")
#
# class MockCalculator:
#     def evaluate(self, expression):
#         # Note: eval() should never be used with untrusted input in production!
#         return eval(expression)
#
# tools = {
#     "stock_checker": MockStockChecker(),
#     "calculator": MockCalculator()
# }
#
# print(use_tool_for_information("What is the current stock price for AAPL?", tools))
# print(use_tool_for_information("Please calculate 15 + 27 * 3", tools))

Hierarchical Summarization

For extremely long documents or conversations, hierarchical summarization can be employed. This technique involves summarizing different sections or turns of a conversation at varying levels of detail. For instance, a long document might first be broken into paragraphs, each paragraph summarized, then these paragraph summaries are summarized into section summaries, and finally, a top-level summary is generated. This allows the LLM to access key information at different granularities without needing to process the entire original text at once.

A conceptual function for hierarchical summarization could involve:

# This function simulates hierarchical summarization of a long document.
def hierarchical_summarize(document_sections, llm_api, summary_level=0):
    if not document_sections:
        return ""
    
    # Base case: if there's only one section or we're at the highest level,
    # just summarize that section.
    if len(document_sections) == 1 or summary_level > 2: # Example max level
        prompt = f"Provide a concise summary of the following text:\n{document_sections[0]}"
        return llm_api.generate_text(prompt=prompt)
    
    # Recursive step: summarize each section, then summarize those summaries.
    intermediate_summaries = []
    for i, section in enumerate(document_sections):
        print(f"Summarizing section {i+1} at level {summary_level}...")
        prompt = f"Summarize the following section:\n{section}"
        section_summary = llm_api.generate_text(prompt=prompt)
        intermediate_summaries.append(section_summary)
    
    # Now, summarize the collection of intermediate summaries.
    combined_summaries = "\n".join(intermediate_summaries)
    print(f"Summarizing intermediate summaries at level {summary_level + 1}...")
    return hierarchical_summarize([combined_summaries], llm_api, summary_level + 1)

# Example conceptual usage:
# class MockLLM:
#     def generate_text(self, prompt):
#         if "section" in prompt.lower():
#             return f"Section summary: {prompt[:50]}..."
#         elif "concise summary" in prompt.lower():
#             return f"Final summary: {prompt[:50]}..."
#         return "Mock summary."
#
# mock_llm_api = MockLLM()
# long_doc_parts = [
#     "The first section discusses the historical development of neural networks, from perceptrons to modern deep learning architectures.",
#     "The second section examines current applications in computer vision, including object detection and image segmentation.",
#     "The final section explores future directions, such as neuromorphic computing and quantum machine learning."
# ]
#
# final_doc_summary = hierarchical_summarize(long_doc_parts, mock_llm_api)
# print("Final Document Summary:")
# print(final_doc_summary)

CHALLENGE 2: BIAS

LLMs, by their very nature, learn from the vast datasets they are trained on, which often reflect societal biases present in the real world. This can lead to models generating outputs that are stereotypical, discriminatory, or unfair, perpetuating harmful narratives. Addressing bias is not just a technical challenge but also an ethical imperative.

Data Preprocessing and Augmentation

Mitigating bias begins with the training data. Techniques include carefully curating datasets to ensure diversity and representation, identifying and removing biased examples, and augmenting data to balance underrepresented groups. For instance, if a dataset disproportionately associates certain professions with one gender, data augmentation can create synthetic examples to balance this representation.

A conceptual function for data cleaning and augmentation might involve:

# This function simulates data preprocessing and augmentation steps
# to mitigate bias in training data.
def preprocess_for_bias_mitigation(raw_data_entries):
    cleaned_data = []
    augmented_data = []
    
    for entry in raw_data_entries:
        # Step 1: Identify and filter explicit biases (e.g., hate speech, stereotypes).
        if not contains_explicit_bias(entry.text):
            cleaned_data.append(entry)
        else:
            print(f"Filtered out biased entry: {entry.text[:50]}...")
            continue # Skip this entry
        
        # Step 2: Augment data to balance representation (conceptual example).
        # If an entry describes a "male engineer", create a "female engineer" variant.
        if "engineer" in entry.text.lower() and "male" in entry.text.lower():
            augmented_entry = create_gender_balanced_variant(entry, "female")
            augmented_data.append(augmented_entry)
        
    return cleaned_data + augmented_data

# Helper function (conceptual) to check for explicit bias
def contains_explicit_bias(text):
    # In a real system, this would involve NLP models or regex for harmful patterns.
    bias_indicators = ["explicitly biased term", "offensive language", "stereotype"]
    return any(indicator in text.lower() for indicator in bias_indicators)

# Helper function (conceptual) to create a gender-balanced variant
class DataEntry:
    def __init__(self, text):
        self.text = text
        
def create_gender_balanced_variant(original_entry, target_gender):
    # This is a highly simplified example. Real augmentation is more complex.
    new_text = original_entry.text.replace("male", target_gender).replace("he", "she").replace("him", "her")
    return DataEntry(new_text)

# Example conceptual usage:
# initial_data = [
#     DataEntry("The male engineer designed the bridge."),
#     DataEntry("She is a brilliant physician."),
#     DataEntry("This contains explicitly biased term and should be removed.")
# ]
#
# processed_data = preprocess_for_bias_mitigation(initial_data)
# print("\nProcessed Data:")
# for item in processed_data:
#     print(item.text)

Fairness-aware Fine-tuning

Beyond data preprocessing, fine-tuning techniques can be adapted to promote fairness. This might involve using fairness-aware loss functions during training, employing adversarial debiasing methods where a discriminator tries to detect bias while the generator tries to hide it, or re-weighting training examples to give more emphasis to underrepresented groups. The goal is to encourage the model to learn less biased representations and generate more equitable outputs.

A conceptual function highlighting fairness-aware fine-tuning:

# This function conceptually outlines a fairness-aware fine-tuning process
# for an LLM, incorporating bias detection and mitigation during training.
def fair_finetune_llm(base_llm_model, fine_tuning_dataset, fairness_metrics, num_epochs):
    print("Starting fairness-aware fine-tuning...")
    
    # Initialize the model for fine-tuning.
    fine_tuned_model = base_llm_model.clone()
    
    for epoch in range(num_epochs):
        print(f"Epoch {epoch+1}...")
        for batch in fine_tuning_dataset:
            # Perform a forward pass and calculate the standard loss.
            predictions = fine_tuned_model.predict(batch.inputs)
            standard_loss = calculate_standard_loss(predictions, batch.labels)
            
            # Calculate fairness-specific loss or bias penalties.
            bias_penalty = calculate_bias_penalty(predictions, batch.sensitive_attributes, fairness_metrics)
            
            # Combine losses for optimization.
            total_loss = standard_loss + bias_penalty
            
            # Perform backpropagation and update model weights.
            fine_tuned_model.optimize(total_loss)
        
        # Periodically evaluate fairness metrics on a validation set.
        current_fairness_score = evaluate_model_fairness(fine_tuned_model, validation_set, fairness_metrics)
        print(f"  Current fairness score: {current_fairness_score}")
    
    print("Fine-tuning complete.")
    return fine_tuned_model

# Helper functions (conceptual)
def calculate_standard_loss(predictions, labels):
    # Placeholder for a typical LLM loss function (e.g., cross-entropy).
    return 0.15 # Example value

def calculate_bias_penalty(predictions, sensitive_attributes, metrics):
    # This would involve comparing prediction outcomes across different
    # sensitive groups (e.g., gender, race) using metrics like
    # demographic parity or equalized odds.
    return 0.08 # Example penalty

def evaluate_model_fairness(model, dataset, metrics):
    # Evaluate the model's performance and bias on a held-out dataset.
    return "Acceptable" # Example fairness evaluation

# Example conceptual usage:
# class MockLLMModel:
#     def clone(self): return MockLLMModel()
#     def predict(self, inputs): return "mock prediction"
#     def optimize(self, loss): pass
#
# class MockDataset:
#     def __iter__(self):
#         class MockBatch:
#             def __init__(self, inputs, labels, sensitive_attributes):
#                 self.inputs = inputs
#                 self.labels = labels
#                 self.sensitive_attributes = sensitive_attributes
#         yield MockBatch("text input", "label", "group_a")
#         yield MockBatch("text input", "label", "group_b")
#
# base_model = MockLLMModel()
# ft_dataset = MockDataset()
# validation_set = None  # Would be defined in real implementation
# fairness_metrics_config = {"demographic_parity": True, "equalized_odds": True}
#
# fine_tuned_llm = fair_finetune_llm(base_model, ft_dataset, fairness_metrics_config, num_epochs=3)

Prompt Engineering for Bias Mitigation

Even without modifying the model itself, careful prompt engineering can significantly reduce biased outputs. By explicitly instructing the LLM to be neutral, inclusive, or to consider diverse perspectives, the model can be guided towards more equitable responses. For instance, a prompt might include phrases like "Please ensure your answer is gender-neutral" or "Consider this from multiple cultural viewpoints."

An example of a bias-mitigating prompt:

# This prompt explicitly instructs the LLM to generate a gender-neutral job description,
# thereby mitigating potential biases in the output.
def create_bias_mitigating_prompt(job_role_description):
    prompt = f"""
    Please write a job description for the following role, ensuring it is
    completely gender-neutral, inclusive, and avoids any stereotypical language.
    Focus solely on skills, responsibilities, and qualifications.
    
    Role Description: {job_role_description}
    
    Job Description:
    """
    return prompt

# Example conceptual usage with a mock LLM call:
# class MockLLM:
#     def generate_text(self, prompt):
#         return "Job description for a Software Developer, emphasizing problem-solving abilities, collaborative skills, and technical expertise in modern programming languages..."
#
# mock_llm_api = MockLLM()
# role = "a software developer who collaborates with cross-functional teams"
# prompt_for_llm = create_bias_mitigating_prompt(role)
# print(f"Generated Prompt:\n{prompt_for_llm}")
# print(f"\nLLM Response:\n{mock_llm_api.generate_text(prompt_for_llm)}")

Filtering and Post-processing

After an LLM generates an output, a secondary filtering or post-processing step can be applied to detect and correct biased content. This might involve using specialized classifiers trained to identify bias, or rule-based systems that flag problematic language. While this is a reactive measure, it serves as a crucial safeguard before content is presented to end-users.

A conceptual function for filtering biased output:

# This function simulates post-processing to filter or flag potentially
# biased content generated by an LLM.
def filter_biased_output(llm_output, bias_detector_model, threshold=0.6):
    print("Analyzing LLM output for bias...")
    
    # Use a bias detection model (e.g., a classifier) to assess the output.
    bias_score = bias_detector_model.predict_bias(llm_output)
    
    if bias_score > threshold:
        print(f"WARNING: Potential bias detected (score: {bias_score:.2f}).")
        # In a real system, one might redact, rewrite, or flag for human review.
        return f"[FLAGGED FOR BIAS REVIEW] {llm_output}"
    else:
        print(f"No significant bias detected (score: {bias_score:.2f}).")
        return llm_output

# Example conceptual usage:
# class MockBiasDetector:
#     def predict_bias(self, text):
#         # Simulate a bias score based on keywords.
#         bias_keywords = ["stereotypical", "discriminatory", "biased"]
#         if any(keyword in text.lower() for keyword in bias_keywords):
#             return 0.85
#         return 0.15
#
# mock_bias_detector = MockBiasDetector()
#
# output1 = "The CEO, a strong leader, guided the organization to success."
# output2 = "This is a stereotypical representation of the profession."
#
# print(f"Output 1 filtered: {filter_biased_output(output1, mock_bias_detector)}")
# print(f"Output 2 filtered: {filter_biased_output(output2, mock_bias_detector)}")

Human-in-the-Loop Validation

Ultimately, human oversight remains indispensable in identifying and correcting LLM bias. Human reviewers can provide nuanced judgments that automated systems often miss. Implementing a human-in-the-loop process, where LLM outputs are reviewed and refined by human experts, especially for sensitive applications, is a robust strategy for ensuring fairness and ethical content generation. This continuous feedback loop also helps in iteratively improving the automated bias detection and mitigation techniques.

CHALLENGE 3: HALLUCINATIONS

Hallucinations refer to the phenomenon where LLMs generate information that is factually incorrect, nonsensical, or contradicts the provided context, yet is presented in a confident and plausible-sounding manner. This is a significant concern, especially in applications requiring high factual accuracy, as it can lead to misinformation and erode trust.

Retrieval Augmented Generation (RAG) Re-emphasized

As discussed for context window limitations, RAG is also a primary defense against hallucinations. By grounding the LLM's responses in verified external data sources, RAG significantly reduces the model's propensity to invent facts. The LLM is prompted to synthesize information from the retrieved documents, making it less likely to generate content not supported by those sources.

Asking for Supporting Sources

A simple yet effective prompt engineering technique is to explicitly instruct the LLM to provide supporting sources or citations for any factual claims it makes. This encourages the model to either retrieve and present evidence or to acknowledge when it cannot find supporting information, thereby making hallucinations more apparent and easier to verify.

An example prompt asking for sources:

# This prompt instructs the LLM to provide supporting sources for its claims,
# helping to mitigate hallucinations.
def create_source_request_prompt(query):
    prompt = f"""
    Please answer the following question. For every factual statement you make,
    please provide a reference to a credible source (e.g., a URL, a document title,
    or a specific section from provided context). If you cannot find a source,
    please state that the information is based on general knowledge and may require
    further verification.
    
    Question: {query}
    """
    return prompt

# Example conceptual usage:
# class MockLLM:
#     def generate_text(self, prompt):
#         if "source" in prompt.lower():
#             return "The speed of light in a vacuum is approximately 299,792 kilometers per second [Source: National Institute of Standards and Technology]."
#         return "The speed of light is very fast."
#
# mock_llm_api = MockLLM()
# query_with_source = create_source_request_prompt("What is the speed of light?")
# print(f"Prompt with source request:\n{query_with_source}")
# print(f"\nLLM Response:\n{mock_llm_api.generate_text(query_with_source)}")

Tree-of-Thought (ToT)

Tree-of-Thought, or ToT, is an advanced prompting strategy that encourages the LLM to explore multiple reasoning paths. Instead of generating a single output directly, the LLM generates several intermediate thoughts or reasoning steps, evaluates their plausibility, and then selectively expands the most promising paths. This branching and pruning process helps the model to self-correct and avoid premature conclusions that might lead to hallucinations.

A conceptual function illustrating the Tree-of-Thought process:

# This function conceptually illustrates the Tree-of-Thought (ToT) process,
# where an LLM explores multiple reasoning paths and evaluates them.
def tree_of_thought_reasoning(initial_problem, llm_api, depth_limit=3):
    print(f"Starting Tree-of-Thought for: {initial_problem}")
    
    # A list to hold potential solutions and their reasoning paths.
    # Each item could be (path_history, current_thought, evaluation_score)
    current_paths = [(initial_problem, "", 0.0)] # Initial state
    
    for depth in range(depth_limit):
        new_paths = []
        print(f"  Exploring at depth {depth+1}...")
        for problem_state, current_thought, score in current_paths:
            # Generate multiple next possible thoughts/steps.
            prompt_for_thoughts = f"""
            Given the problem: "{initial_problem}"
            And current thought path: "{current_thought}"
            Suggest 2-3 distinct next reasoning steps or ideas.
            """
            
            # Simulate LLM generating multiple thoughts.
            # In reality, this would be a structured output from the LLM.
            generated_thoughts = llm_api.generate_thoughts(prompt_for_thoughts)
            
            for thought in generated_thoughts:
                new_thought_path = f"{current_thought} -> {thought}" if current_thought else thought
                
                # Evaluate the plausibility/correctness of the new thought path.
                # This is a critical step for pruning bad paths.
                evaluation_prompt = f"""
                Evaluate the following reasoning step in the context of the problem:
                Problem: "{initial_problem}"
                Path so far: "{new_thought_path}"
                Is this step logical and likely to lead to a correct solution? Rate 0-1.
                """
                # Simulate LLM evaluating the thought.
                evaluation_score = llm_api.evaluate_thought(evaluation_prompt)
                
                if evaluation_score > 0.5: # Only keep promising paths
                    new_paths.append((initial_problem, new_thought_path, evaluation_score))
                    print(f"    Kept path: '{new_thought_path}' (Score: {evaluation_score:.2f})")
                else:
                    print(f"    Pruned path: '{new_thought_path}' (Score: {evaluation_score:.2f})")
        
        if not new_paths:
            print("  No promising paths found, stopping early.")
            break
        current_paths = new_paths
    
    # After exploring, select the best path or synthesize from the best ones.
    best_path = max(current_paths, key=lambda x: x[2], default=(None, "No solution found", 0.0))
    print(f"Final best reasoning path: {best_path[1]} (Score: {best_path[2]:.2f})")
    return best_path[1]

# Example conceptual usage:
# class MockLLM:
#     def generate_thoughts(self, prompt):
#         if "optimize algorithm" in prompt.lower():
#             return ["analyze time complexity", "identify bottlenecks", "apply caching"]
#         return ["approach A", "approach B"]
#     def evaluate_thought(self, prompt):
#         if "logical" in prompt.lower():
#             if "analyze time complexity" in prompt or "identify bottlenecks" in prompt:
#                 return 0.92
#             return 0.35
#         return 0.70
#
# mock_llm_api = MockLLM()
# problem = "optimize algorithm for large dataset processing"
# result = tree_of_thought_reasoning(problem, mock_llm_api, depth_limit=2)
# print(f"\nToT Result: {result}")

Chain-of-Thought (CoT)

Chain-of-Thought, or CoT, prompting is a simpler but highly effective technique where the LLM is explicitly instructed to think step by step or show its reasoning. By forcing the model to articulate its intermediate steps, CoT makes the reasoning process transparent, allowing for easier identification of errors or logical leaps that might lead to hallucinations. It often improves the accuracy of complex reasoning tasks.

A Chain-of-Thought prompt example:

# This prompt encourages the LLM to explain its reasoning step-by-step,
# which helps in identifying and reducing hallucinations.
def create_chain_of_thought_prompt(question):
    prompt = f"""
    Please answer the following question. First, break down the problem into
    logical steps and explain your reasoning for each step. Then, provide
    your final answer.
    
    Question: {question}
    """
    return prompt

# Example conceptual usage:
# class MockLLM:
#     def generate_text(self, prompt):
#         if "logical steps" in prompt.lower():
#             return """Step 1: Identify the key components of photosynthesis (light, water, CO2).
# Step 2: Understand the light-dependent reactions in chloroplasts.
# Step 3: Examine the Calvin cycle for carbon fixation.
# Final Answer: Photosynthesis converts light energy into chemical energy stored in glucose."""
#         return "Photosynthesis is how plants make food."
#
# mock_llm_api = MockLLM()
# question_cot = create_chain_of_thought_prompt("How does photosynthesis work?")
# print(f"Prompt with CoT:\n{question_cot}")
# print(f"\nLLM Response:\n{mock_llm_api.generate_text(question_cot)}")

Self-Reflection

Self-reflection involves prompting the LLM to critically evaluate its own generated output. After generating an initial response, the LLM is given its own output and asked to critique it, identify potential errors or inconsistencies, and then revise its answer. This iterative self-correction mechanism can significantly improve the factual accuracy and coherence of the generated text.

A conceptual function demonstrating self-reflection:

# This function demonstrates an LLM performing self-reflection to review
# and correct its own output, reducing errors and hallucinations.
def self_reflect_and_correct(initial_query, llm_api):
    print(f"Initial generation for: {initial_query}")
    # Stage 1: Generate an initial response.
    initial_response = llm_api.generate_text(prompt=initial_query)
    print(f"  Initial Response: {initial_response}")
    
    # Stage 2: Prompt the LLM to reflect on its own response.
    reflection_prompt = f"""
    You previously generated the following response to the query "{initial_query}":
    ---
    {initial_response}
    ---
    
    Please critically review this response.
    1. Identify any potential inaccuracies, inconsistencies, or areas that could be improved.
    2. Based on your critique, provide a revised and improved response.
    
    Critique and Revised Response:
    """
    
    print("  LLM performing self-reflection...")
    revised_output = llm_api.generate_text(prompt=reflection_prompt)
    
    print(f"  Revised Output: {revised_output}")
    return revised_output

# Example conceptual usage:
# class MockLLM:
#     def generate_text(self, prompt):
#         if "What is the capital" in prompt:
#             return "The capital of Australia is Sydney." # Intentional hallucination
#         elif "critically review" in prompt:
#             return "Critique: The previous response incorrectly stated the capital of Australia. Sydney is the largest city, but Canberra is the capital. Revised Response: The capital of Australia is Canberra."
#         return "Mock response."
#
# mock_llm_api = MockLLM()
# corrected_response = self_reflect_and_correct("What is the capital of Australia?", mock_llm_api)
# print(f"\nFinal Corrected Response: {corrected_response}")

Fact-Checking and Verification Tools

Integrating LLMs with external fact-checking and verification tools provides an automated layer of defense against hallucinations. After an LLM generates a factual statement, a specialized tool can cross-reference that statement against trusted databases, scientific literature, or reputable news sources. If a discrepancy is found, the LLM's output can be flagged, revised, or sent for human review.

A conceptual function for fact-checking:

# This function simulates using an external fact-checking tool to verify
# statements generated by an LLM, thereby combating hallucinations.
def fact_check_llm_statement(llm_statement, fact_checker_api):
    print(f"Fact-checking statement: '{llm_statement}'")
    
    # Call the external fact-checking API.
    verification_result = fact_checker_api.verify(llm_statement)
    
    if verification_result.is_accurate:
        print("  Statement verified as accurate.")
        return f"VERIFIED: {llm_statement}"
    else:
        print(f"  Statement flagged as potentially inaccurate: {verification_result.reason}")
        return f"FLAGGED (Reason: {verification_result.reason}): {llm_statement}"

# Example conceptual usage:
# class MockFactCheckerAPI:
#     class VerificationResult:
#         def __init__(self, is_accurate, reason=""):
#             self.is_accurate = is_accurate
#             self.reason = reason
#
#     def verify(self, statement):
#         if "Mount Everest" in statement and "tallest mountain" in statement:
#             return self.VerificationResult(True)
#         elif "Mount Kilimanjaro" in statement and "tallest mountain" in statement:
#             return self.VerificationResult(False, "Mount Kilimanjaro is not the tallest mountain; Mount Everest is.")
#         return self.VerificationResult(True) # Default to true for other statements
#
# mock_fact_checker = MockFactCheckerAPI()
#
# statement1 = "Mount Everest is the tallest mountain on Earth."
# statement2 = "Mount Kilimanjaro is the tallest mountain on Earth."
#
# print(fact_check_llm_statement(statement1, mock_fact_checker))
# print(fact_check_llm_statement(statement2, mock_fact_checker))

CONCLUSION

The challenges of constrained context windows, inherent biases, and factual hallucinations represent significant hurdles in the widespread and reliable deployment of Large Language Models. However, as this article has demonstrated, a rich and evolving toolkit of strategies and techniques is available to address these issues effectively. From architectural patterns like Retrieval Augmented Generation and GraphRAG, which ground models in external knowledge, to sophisticated prompting methods like Chain-of-Thought and Tree-of-Thought that enhance reasoning capabilities, and post-processing steps that filter and verify outputs, a multi-faceted approach is essential. Continuous research and development in these areas, combined with diligent application of best practices, will pave the way for more robust, trustworthy, and ethically sound LLM applications. The journey towards truly intelligent and reliable artificial intelligence systems is ongoing, and proactively tackling these challenges represents a critical step forward in realizing the full potential of this transformative technology.

BUILDING A SELF-LEARNING LLM CHATBOT WITH HYBRID RAG ARCHITECTURE

 

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.