# 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()