79697373

Date: 2025-07-10 16:53:58
Score: 4.5
Natty:
Report link
# self_learning_ea_system.py

import spacy
import numpy as np
import networkx as nx
import pickle
import os
import json
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
from tensorflow import keras
from tensorflow.keras import layers
from sentence_transformers import SentenceTransformer

# --- 1. Knowledge Graph using triples + networkx

class KnowledgeGraph:
    def __init__(self):
        self.triples = []  # (s, r, o)
        self.graph = nx.DiGraph()

    def update(self, s, r, o):
        if (s, r, o) not in self.triples:
            self.triples.append((s, r, o))
            self.graph.add_edge(s, o, label=r)

    def query(self, s=None, r=None, o=None):
        return [
            (s0, r0, o0) for (s0, r0, o0) in self.triples
            if (s is None or s0 == s) and (r is None or r0 == r) and (o is None or o0 == o)
        ]

    def __str__(self):
        return "\n".join([f"{s} -[{r}]-> {o}" for s, r, o in self.triples])

# --- 2. NLP extractor with spaCy

nlp = spacy.load("en_core_web_sm")
embedder = SentenceTransformer('all-MiniLM-L6-v2')

def extract_triples(text):
    doc = nlp(text)
    triples = []
    for token in doc:
        if token.dep_ == "ROOT":
            subjects = [w for w in token.lefts if w.dep_ in ("nsubj", "nsubjpass")]
            objects = [w for w in token.rights if w.dep_ in ("dobj", "attr", "prep", "pobj")]
            for s in subjects:
                for o in objects:
                    triples.append((s.text, token.lemma_, o.text))
    if not triples:
        parts = text.split()
        for rel in ("is", "has", "part"):
            if rel in parts:
                i = parts.index(rel)
                if i >= 1 and i < len(parts) - 1:
                    triples.append((parts[i - 1], rel, parts[i + 1]))
    return triples

def triple_to_vec(s, r, o):
    return embedder.encode(f"{s} {r} {o}")

# --- 3. Relation prediction model

def build_model(input_dim):
    model = keras.Sequential([
        layers.Dense(64, activation="relu", input_shape=(input_dim,)),
        layers.Dense(32, activation="relu"),
        layers.Dense(1, activation="sigmoid"),
    ])
    model.compile(optimizer="adam", loss="binary_crossentropy")
    return model

# --- 4. Evolutionary algorithm

class EvolutionaryAlgorithm:
    def __init__(self, system, base_rate=0.02):
        self.system = system
        self.base_rate = base_rate
        self.mutation_rate = base_rate

    def update_mutation_rate(self, accuracy):
        self.mutation_rate = max(0.005, self.base_rate * (1 - accuracy))

    def evolve(self):
        model = self.system["model"]
        weights = model.get_weights()
        mutated = [w + self.mutation_rate * np.random.randn(*w.shape) for w in weights]
        model.set_weights(mutated)
        print(f"๐Ÿ” Mutated model weights with rate {self.mutation_rate:.4f}.")

# --- 5. Learning Module

class LearningModule:
    def __init__(self, kg, system):
        self.kg = kg
        self.system = system
        self.training_data = []

    def add_training_example(self, s, r, o, label):
        self.training_data.append((s, r, o, label))
        acc = self.train()
        self.system["ea"].update_mutation_rate(acc)

    def train(self, epochs=10, batch_size=16):
        if not self.training_data:
            print("No training data available.")
            return 0.0

        X, y = [], []
        for s, r, o, label in self.training_data:
            vec = triple_to_vec(s, r, o)
            X.append(vec)
            y.append(label)

        X = np.vstack(X)
        y = np.array(y)
        X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)

        model = self.system["model"]
        model.fit(X_train, y_train, epochs=epochs, batch_size=batch_size, verbose=0)
        preds = (model.predict(X_val) > 0.5).astype(int).flatten()
        acc = accuracy_score(y_val, preds)
        print(f"๐Ÿงช Trained model โ€” validation accuracy: {acc:.2f}")
        return acc

# --- 6. Reasoning Engine

class ReasoningEngine:
    def __init__(self, kg, system):
        self.kg = kg
        self.system = system

    def reason(self, query):
        doc = nlp(query)
        for ent in doc.ents:
            facts = self.kg.query(s=ent.text)
            if facts:
                return "Known: " + "; ".join(f"{s} {r} {o}" for s, r, o in facts)
        s, r, o = self.extract_subject_relation_object(query)
        if s and r and o:
            prob = self.predict_relation(s, r, o)
            if prob > 0.7:
                return f"Predicted with confidence {prob:.2f}: {s} {r} {o}"
        return "Unknown โ€” please provide feedback to improve me!"

    def extract_subject_relation_object(self, text):
        parts = text.split()
        if len(parts) >= 3:
            return parts[0], parts[1], " ".join(parts[2:])
        return None, None, None

    def predict_relation(self, s, r, o):
        vec = triple_to_vec(s, r, o)
        prob = self.system["model"].predict(vec.reshape(1, -1))[0, 0]
        return prob

# --- 7. Save/Load System State

def save_system(path="system_state.pkl"):
    with open(path, "wb") as f:
        pickle.dump({
            "triples": SYSTEM["kg"].triples,
            "training_data": SYSTEM["learner"].training_data,
            "model_weights": SYSTEM["model"].get_weights(),
        }, f)

def load_system(path="system_state.pkl"):
    if os.path.exists(path):
        with open(path, "rb") as f:
            data = pickle.load(f)
            for s, r, o in data["triples"]:
                SYSTEM["kg"].update(s, r, o)
            SYSTEM["learner"].training_data = data["training_data"]
            SYSTEM["model"].set_weights(data["model_weights"])
            print("โœ… System state loaded.")
    else:
        print("โš ๏ธ No saved system state found.")

# --- 8. Main EA system assembly

input_dim = 384
SYSTEM = {
    "kg": KnowledgeGraph(),
    "input_dim": input_dim,
    "model": build_model(input_dim),
}
SYSTEM["ea"] = EvolutionaryAlgorithm(SYSTEM)
SYSTEM["learner"] = LearningModule(SYSTEM["kg"], SYSTEM)
SYSTEM["reasoner"] = ReasoningEngine(SYSTEM["kg"], SYSTEM)

# --- 9. User interaction

def interact(query):
    resp = SYSTEM["reasoner"].reason(query)
    print("๐Ÿค–:", resp)
    if resp.startswith("Unknown"):
        feedback = input("โœ… Please provide correct answer (S R O, pipe-separated): ")
        try:
            s, r, o = feedback.split("|")
            SYSTEM["kg"].update(s.strip(), r.strip(), o.strip())
            SYSTEM["learner"].add_training_example(s, r, o, label=1)
            SYSTEM["ea"].evolve()
        except ValueError:
            print("โš ๏ธ Invalid format. Skipping update.")
    return resp

# --- 10. Command-line interface

def cli():
    print("๐Ÿค– Welcome to the Evolving AI System. Type 'quit' to exit.")
    while True:
        q = input("\nAsk a question or type a command ('save', 'load'): ")
        if q.lower() == "quit":
            save_system()
            print("๐Ÿ›‘ Goodbye!")
            break
        elif q.lower() == "save":
            save_system()
            print("๐Ÿ’พ System saved.")
        elif q.lower() == "load":
            load_system()
        else:
            interact(q)

# --- 11. Main

if __name__ == "__main__":
    load_system()
    cli()
Reasons:
  • RegEx Blacklisted phrase (2.5): please provide
  • RegEx Blacklisted phrase (2.5): Please provide
  • Long answer (-1):
  • Has code block (-0.5):
  • Low reputation (1):
Posted by: user31005825