Hier erst einmal "Mother.py":
Code: Alles auswählen
import re
import inspect
import pandas as pd
import os
import spacy
import Levenshtein
from flask import Flask, request, jsonify
from mesa import Model
from mesa.time import BaseScheduler
import json
from datetime import datetime
# Importieren Sie hier alle Ihre Agentenklassen
from AAgent import AAgent
from AberAgent import AberAgent
from AbstraktionNomenAgent import AbstraktionNomenAgent
from AbstraktionVerbenAgent import AbstraktionVerbenAgent
from AccAgent import AccAgent
from AlkAgent import AlkAgent
from PseudoSatisfierAgent import PseudoSatisfierAgent
from AngstAgent import AngstAgent
from BerufeAgent import BerufeAgent
from BesitzerUndBesessenesAgent import BesitzerUndBesessenesAgent
from BirthDefectAgent import BirthDefectAgent
from BisAgent import BisAgent
from CreateAgent import CreateAgent
from DatAgent import DatAgent
from DenkenRedenTunAgent import DenkenRedenTunAgent
from DestroyerAgent import DestroyerAgent
from DiebstahlAgent import DiebstahlAgent
from DingAgent import DingAgent
from DuDisambigAgent import DuDisambigAgent
from EkelAgent import EkelAgent
from ErwachsenenAgent import ErwachsenenAgent
from FalseAgent import FalseAgent
from FertilitaetsAgent import FertilitaetsAgent
from FoodHealthyAgent import FoodHealthyAgent
from FoodNOTHealthyAgent import FoodNOTHealthyAgent
from FrauAgent import FrauAgent
from FreiwilligkeitsAgent import FreiwilligkeitsAgent
from FreudeAgent import FreudeAgent
from FreundschaftsAgent import FreundschaftsAgent
from GebenAgent import GebenAgent
from GesternAgent import GesternAgent
from GottesAgent import GottesAgent
from GruppenAgent import GruppenAgent
from HierarchieAgent import HierarchieAgent
from IchDisambigAgent import IchDisambigAgent
from IhrDisambigAgent import IhrDisambigAgent
from JungeErwachseneAgent import JungeErwachseneAgent
from KaufAgent import KaufAgent
from KindAgent import KindAgent
from LiebesAgent import LiebesAgent
from MannAgent import MannAgent
from MeinAgent import MeinAgent
from MorgenDisambigAgent import MorgenDisambigAgent
from MuedigkeitsAgent import MuedigkeitsAgent
from NachnamenAgent import NachnamenAgent
from NehmenAgent import NehmenAgent
from AltAgent import AltAgent
from OntologieAgent import OntologieAgent
from OrtsAgent import OrtsAgent
from PflanzenAgent import PflanzenAgent
from PlanungAgent import PlanungsAgent
from SchenkenAgent import SchenkenAgent
from SeinerSeineSeinesAgent import SeinerSeineSeinesAgent
from StartAgent import StartAgent
from StaunenAgent import StaunenAgent
from SterbenAgent import SterbenAgent
from StopAgent import StopAgent
from SuchtAgent import SuchtAgent
from HalbwuechsigenAgent import HalbwuechsigenAgent
from TierAgent import TierAgent
from TraurigkeitsAgent import TraurigkeitsAgent
from TrueAgent import TrueAgent
from UebermorgenAgent import UebermorgenAgent
from UhrzeitAgent import UhrzeitAgent
from UndAgent import UndAgent
from VerachtenAgent import VerachtenAgent
from VerkaufenAgent import VerkaufenAgent
from VerneinungsAgent import VerneinungsAgent
from VersprechensAgent import VersprechensAgent
from VersuchenAgent import VersuchenAgent
from VorgesternAgent import VorgesternAgent
from WahrnehmungsartAgent import WahrnehmungsartAgent
from WennAgent import WennAgent
from WertungNegativgent import WertungNegativAgent
from WertungPositivAgent import WertungPositivAgent
from ZeitWochenAgent import ZeitWochenAgent
from JungeErwachseneAgent import JungeErwachseneAgent
from ZeitJahreAgent import ZeitJahreAgent
from ZeitMonateAgent import ZeitMonateAgent
from ZeitStundenAgent import ZeitStundenAgent
from ZeitTageAgent import ZeitTageAgent
from ZornAgent import ZornAgent
# SpaCy-Modell laden
nlp = spacy.load("de_core_news_sm")
# Funktion zur Überprüfung der grammatikalischen Form
def get_grammatical_case(word, sentence):
doc = nlp(sentence)
for token in doc:
if Levenshtein.distance(token.text, word) <= 2:
return token.morph.get('Case')[0] # Nimmt den ersten Wert, falls mehrere vorhanden sind
return None
# Klasse zum Verwalten von DataFrames
class DataFrameManager:
def __init__(self, directory):
self.directory = directory
if not os.path.exists(directory):
os.makedirs(directory)
def save_dataframe(self, dataframe, filename):
path = os.path.join(self.directory, f"{filename}.csv")
dataframe.to_csv(path, index=False)
def load_dataframe(self, filename):
path = os.path.join(self.directory, f"{filename}.csv")
if os.path.exists(path):
return pd.read_csv(path)
return pd.DataFrame() # Leeres DataFrame, falls Datei nicht existiert
# Klasse für das Kurzzeitgedächtnis
class ShortTermMemory:
def __init__(self):
self.dataframe = pd.DataFrame(columns=[
"ICoderMX", # Spalte für Slot 1
"sagt", # Spalte für Slot 2
"EmoHandelnde", # Spalte für Slot 3
"FreiwilligHandelnde", # Spalte für Slot 4
"WertungHandelnde", # Spalte für Slot 5
"Handelnde", # Spalte für Slot 6
"WennOderAber", # Spalte für Slot 7
"Verneinung", # Spalte für Slot 8
"AbstraktesVerb", # Spalte für Slot 9
"DenkenRedenTun", # Spalte für Slot 10
"EmoBehandelteAcc", # Spalte für Slot 11
"FreiwilligBehandelteAcc", # Spalte für Slot 12
"WertungBehandelteAcc", # Spalte für Slot 13
"BehandelteAcc", # Spalte für Slot 14
"EmoBehandelteDat", # Spalte für Slot 15
"FreiwilligBehandelteDat", # Spalte für Slot 16
"WertungBehandelteDat", # Spalte für Slot 17
"BehandelteDat", # Spalte für Slot 18
"Hierarchie", # Spalte für Slot 19
"Ort", # Spalte für Slot 20
"Zeit", # Spalte für Slot 21
"Ethik", # Spalte für Slot 22
"Ontologie", # Spalte für Slot 23
"Beduerfnis", # Spalte für Slot 24
"Beduerfniserfueller", # Spalte für Slot 25
"Beduerfniszerstoerer", # Spalte für Slot 26
"Pseudoerfueller", # Spalte für Slot 27
"Wahrnehmungsart", # Spalte für Slot 28
"Systemzeit", # Spalte für Slot 29
"ZweiKlammern" # Spalte für Slot 30
])
self.slots = {i: [] for i in range(1, 31)}
self.slots[2] = [":"] # Slot 2 hat immer einen Doppelpunkt
self.slots[30] = ["))"]
def update_slot(self, slot_number, data):
if slot_number in self.slots:
self.slots[slot_number] = data
else:
raise ValueError(f"Slot {slot_number} does not exist.")
def save_to_dataframe(self):
row = pd.Series(self.slots, index=self.dataframe.columns)
self.dataframe = pd.concat([self.dataframe, pd.DataFrame([row])], ignore_index=True)
def reset_slots(self):
self.slots = {i: [] for i in range(1, 31)}
def get_dataframe(self):
return self.dataframe
def map_agent_to_slot(self, agent_name, word, sentence):
case = get_grammatical_case(word, sentence)
if agent_name in ["TierAgent", "DingAgent", "FrauAgent", "MannAgent", "PflanzenAgent",
"HalbwuechsigenAgent", "KindAgent", "JungeErwachseneAgent", "AltAgent",
"NachnamenAgent", "ErwachsenenAgent", "AbstraktionNomenAgent",
"GruppenAgent", "VornamenAgent", "MeinAgent"]:
if case == "Nom":
return 6
elif case == "Acc":
return 14
elif case == "Dat":
return 18
elif agent_name == "FreiwilligkeitsAgent":
if case == "Nom":
return 4
elif case == "Acc":
return 12
elif case == "Dat":
return 16
elif agent_name in ["EkelAgent", "AngstAgent", "FreudeAgent", "StaunenAgent",
"TraurigkeitsAgent", "ZornAgent", "FreundschaftsAgent",
"LiebesAgent", "VerachtenAgent"]:
if case == "Nom":
return 3
elif case == "Acc":
return 11
elif case == "Dat":
return 15
elif agent_name in ["WertungPositivAgent", "WertungNegativAgent"]:
if case == "Nom":
return 5
elif case == "Acc":
return 13
elif case == "Dat":
return 17
return None
def update_short_term_memory(self, agent_results, sentence):
# Setze feste Werte für bestimmte Slots
self.slots[2] = [":"] # Slot 2 hat immer einen Doppelpunkt
self.slots[30] = ["))"] # Slot 30 hat immer zwei geschlossene Klammern
# Gehe durch die Ergebnisse der Agenten und aktualisiere die Slots
for agent_name, result in agent_results.items():
slot_number = self.map_agent_to_slot(agent_name, result, sentence)
if slot_number:
self.update_slot(slot_number, result)
# Die aktuelle Zeit in Slot 29 speichern
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.slots[29] = [current_time]
# Wenn WahrnehmungsartAgent in den Ergebnissen ist, aktualisiere Slot 28
if "WahrnehmungsartAgent" in agent_results:
self.slots[28] = agent_results["WahrnehmungsartAgent"]
# Eventuell weitere Logik hier einfügen für weitere feste Slotwerte oder falls
# manche Ergebnisse immer an den gleichen Platz sollen
# Speichern Sie den aktuellen Stand der Slots im DataFrame
self.save_to_dataframe()
# Setzen Sie die Slots zurück, um bereit für die nächste Runde von Anfragen zu sein
self.reset_slots()
def save_to_csv(self, filename):
self.dataframe.to_csv(filename, index=False)
def load_from_csv(self, filename):
if os.path.exists(filename):
self.dataframe = pd.read_csv(filename)
# Definition der Flask-App
app = Flask(__name__)
short_term_memory = ShortTermMemory()
# Definition des Modells
class MyModel(Model):
def __init__(self, input_sentence):
super().__init__() # Aufruf von Model-Konstruktor
self.schedule = BaseScheduler(self)
self.input_sentence = input_sentence
agent_types = [
AAgent,
AberAgent,
AbstraktionNomenAgent,
AbstraktionVerbenAgent,
AccAgent,
AlkAgent,
AltAgent,
PseudoSatisfierAgent,
AngstAgent,
BerufeAgent,
BesitzerUndBesessenesAgent,
BirthDefectAgent,
BisAgent,
CreateAgent,
DatAgent,
DenkenRedenTunAgent,
DestroyerAgent,
DiebstahlAgent,
DingAgent,
DuDisambigAgent,
EkelAgent,
ErwachsenenAgent,
FalseAgent,
FertilitaetsAgent,
FoodHealthyAgent,
FoodNOTHealthyAgent,
FrauAgent,
FreiwilligkeitsAgent,
FreudeAgent,
FreundschaftsAgent,
GebenAgent,
GesternAgent,
GottesAgent,
ErwachsenenAgent,
GruppenAgent,
HalbwuechsigenAgent,
HierarchieAgent,
IchDisambigAgent,
IhrDisambigAgent,
JungeErwachseneAgent,
KaufAgent,
KindAgent,
LiebesAgent,
MannAgent,
MeinAgent,
MorgenDisambigAgent,
MuedigkeitsAgent,
NachnamenAgent,
NehmenAgent,
OntologieAgent,
OrtsAgent,
PflanzenAgent,
PlanungsAgent,
SchenkenAgent,
SeinerSeineSeinesAgent,
StartAgent,
StaunenAgent,
SterbenAgent,
StopAgent,
SuchtAgent,
TierAgent,
TraurigkeitsAgent,
TrueAgent,
UebermorgenAgent,
UhrzeitAgent,
UndAgent,
VerachtenAgent,
VerkaufenAgent,
VerneinungsAgent,
VersprechensAgent,
VersuchenAgent,
VorgesternAgent,
WahrnehmungsartAgent,
WennAgent,
WertungNegativAgent,
WertungPositivAgent,
ZeitWochenAgent,
ZeitJahreAgent,
ZeitMonateAgent,
ZeitStundenAgent,
ZeitTageAgent,
ZornAgent
]
for i, agent_type in enumerate(agent_types):
# Parameter der init-Methode der Agent-Klasse abrufen
params = inspect.signature(agent_type.__init__).parameters
# Wörterbuch zur Übergabe an den Agent-Konstruktor erstellen
init_args = {}
for key in params.keys():
if key == 'unique_id':
init_args[key] = i
elif key == 'model':
init_args[key] = self
elif key == 'input_sentence':
init_args[key] = input_sentence
# Hier könnten andere Parameter hinzugefügt werden, die Ihre Agenten benötigen
# Agent instanziieren und zum Scheduler hinzufügen
agent = agent_type(**init_args)
self.schedule.add(agent)
def step(self):
# Ausführung jeder Agent `step`-Methode
self.schedule.step()
# Erfassen der Daten der Agenten
agent_results = {}
for agent in self.schedule.agents:
result = agent.step() # Ruft die step()-Methode auf und erwartet ein Ergebnis, das sie zurückgibt
agent_results[type(agent).__name__] = result
# Hier fügen Sie den Print-Befehl ein
print("Agent-Ergebnisse:", agent_results)
return agent_results
# Globale Variable, die das Model speichert
my_model = None
@app.before_first_request
def initialize_model():
global my_model
# Erstelle eine Instanz von MyModel
# Hier können Sie entscheiden, wie Sie die `input_sentence` initialisieren möchten
my_model = MyModel(input_sentence="Start")
@app.route('/process_input', methods=['POST'])
def process_input():
global my_model
data = request.json
speaker = data.get('speaker', 'Unknown')
message = data.get('message', '')
# Aktualisiere Slot 1 mit dem 'speaker'
short_term_memory.update_slot(1, [speaker])
# Füge die aktuelle Zeit in Slot 29 ein
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
short_term_memory.update_slot(29, [current_time])
# Daten an die Agenten senden
for agent in my_model.schedule.agents:
if hasattr(agent, 'receive_data_from_mother'):
agent.receive_data_from_mother({'speaker': speaker, 'message': message})
# Verarbeiten Sie die eingehenden Daten von allen Agenten
if 'processed_data' in data:
for agent_name, processed_data in data['processed_data'].items():
matches = find_parenthetical(processed_data)
for match in matches:
# Entferne die Klammern und extrahiere den Inhalt
content = match[match.find("(") + 1:match.rfind(")")]
# Füge den Inhalt nach den in Mother beschriebenen Regeln in die Slots ein
slot_number = short_term_memory.map_agent_to_slot(agent_name, content, message)
if slot_number is not None:
short_term_memory.update_slot(slot_number, [content])
# Speichern und Zurücksetzen des Zustands, wie zuvor beschrieben
short_term_memory.save_to_dataframe()
short_term_memory.reset_slots()
# Sende eine Bestätigung, dass die Daten erfolgreich verarbeitet wurden
return jsonify({"message": "Data processed and slots updated"}), 200
def find_parenthetical(processed_data):
# Suche nach Mustern mit beliebigen Zeichen vor der öffnenden Klammer, einschließlich dieser Zeichen
pattern = r'[^\s]+?\([^\)]+\)'
return re.findall(pattern, processed_data)
@app.route('/get_dataframe', methods=['GET'])
def get_dataframe():
df = short_term_memory.get_dataframe()
response = df.replace({pd.NA: None}).to_json(orient='records')
return response, 200
if __name__ == '__main__':
app.run(port=5001, debug=False)
Code: Alles auswählen
import re
import json
import requests
from mesa import Agent
class KaufAgent(Agent):
def __init__(self, unique_id, model):
super().__init__(unique_id, model)
self.config = self.load_config()
self.input_data = None # Hinzugefügt, um Daten zu speichern
def step(self):
# Überprüfen, ob Daten empfangen wurden und entsprechend handeln
if self.input_data is not None:
processed_data = self.detect_purchase_intent(self.input_data)
self.send_to_mother(processed_data)
self.input_data = None # Daten zurücksetzen
def detect_purchase_intent(self, text):
kauf_synonyme = {
"kaufen": r"\bkaufen\b|\bkaufe\b|\bkaufst\b|\bkaufte\b|\bgekauft\b",
"erwerben": r"\berwerben\b|\berwirb\b|\berwarb\b|\berworben\b",
"anschaffen": r"\banschaffen\b|\banschafft\b|\banschaffte\b|\bangeschafft\b",
"besorgen": r"\bbesorgen\b|\bbesorgt\b|\bbesorgte\b|\bbesorgt\b",
"erhalten": r"\berhalten\b|\berhältst\b|\berhielt\b|\berhalten\b",
"holen": r"\bholen\b|\bhole\b|\bholst\b|\bholte\b|\bgeholt\b"
}
pattern = re.compile("|".join(kauf_synonyme.values()), re.IGNORECASE)
matches = pattern.findall(text)
matched_phrases = [f"(<$(+x):)({match})" for match in matches] if matches else ["KEIN_KAUF_ERKANNT"]
return ' '.join(matched_phrases)
def send_to_mother(self, data):
url = self.config.get("mother_url")
headers = {'Content-Type': 'application/json'}
payload = {"processed_sentence": data}
try:
response = requests.post(url, json=payload, headers=headers)
if response.status_code != 200:
print(f"Fehler beim Senden: {response.status_code}, Antwort: {response.text}")
except Exception as e:
print(f"Fehler im KaufAgent {self.unique_id}: {e}")
def receive_data_from_mother(self, data):
# Empfangene Daten speichern
self.input_data = data.get('message', '')Vielleicht sieht ja jemand von euch, woran es liegen könnte.