m3.py

import os, xml.etree.ElementTree as ET, torch, torch.nn as nn, torch.nn.functional as F, numpy as np
from typing import List, Dict, Any, Optional
from collections import defaultdict
from accelerate import Accelerator
from transformers import AutoTokenizer, AutoModel
from termcolor import colored
from sklearn.metrics.pairwise import cosine_similarity

class DM(nn.Module):
    def __init__(self, s: Dict[str, List[Dict[str, Any]]]):
        super(DM, self).__init__()
        self.s = nn.ModuleDict()
        if not s: s = {'default': [{'input_size': 128, 'output_size': 256, 'activation': 'relu', 'batch_norm': True, 'dropout': 0.1}]}
        for sn, l in s.items():
            self.s[sn] = nn.ModuleList()
            for lp in l: 
                print(colored(f"Creating layer in section '{sn}' with params: {lp}", 'cyan'))
                self.s[sn].append(self.cl(lp))

    def cl(self, lp: Dict[str, Any]) -> nn.Module:
        l = [nn.Linear(lp['input_size'], lp['output_size'])]
        if lp.get('batch_norm', True): l.append(nn.BatchNorm1d(lp['output_size']))
        a = lp.get('activation', 'relu')
        if a == 'relu': l.append(nn.ReLU(inplace=True))
        elif a == 'tanh': l.append(nn.Tanh())
        elif a == 'sigmoid': l.append(nn.Sigmoid())
        elif a == 'leaky_relu': l.append(nn.LeakyReLU(negative_slope=0.01, inplace=True))
        elif a == 'elu': l.append(nn.ELU(alpha=1.0, inplace=True))
        elif a is not None: raise ValueError(f"Unsupported activation function: {a}")
        if dr := lp.get('dropout', 0.0): l.append(nn.Dropout(p=dr))
        if hl := lp.get('hidden_layers', []):
            for hlp in hl: l.append(self.cl(hlp))
        if lp.get('memory_augmentation', True): l.append(MAL(lp['output_size']))
        if lp.get('hybrid_attention', True): l.append(HAL(lp['output_size']))
        if lp.get('dynamic_flash_attention', True): l.append(DFAL(lp['output_size']))
        return nn.Sequential(*l)

    def forward(self, x: torch.Tensor, sn: Optional[str] = None) -> torch.Tensor:
        if sn is not None:
            if sn not in self.s: raise KeyError(f"Section '{sn}' not found in model")
            for l in self.s[sn]: x = l(x)
        else:
            for sn, l in self.s.items():
                for l in l: x = l(x)
        return x

class MAL(nn.Module):
    def __init__(self, s: int):
        super(MAL, self).__init__()
        self.m = nn.Parameter(torch.randn(s))

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return x + self.m

class HAL(nn.Module):
    def __init__(self, s: int):
        super(HAL, self).__init__()
        self.a = nn.MultiheadAttention(s, num_heads=8)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        x = x.unsqueeze(1)
        ao, _ = self.a(x, x, x)
        return ao.squeeze(1)

class DFAL(nn.Module):
    def __init__(self, s: int):
        super(DFAL, self).__init__()
        self.a = nn.MultiheadAttention(s, num_heads=8)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        x = x.unsqueeze(1)
        ao, _ = self.a(x, x, x)
        return ao.squeeze(1)

def px(file_path: str) -> List[Dict[str, Any]]:
    t = ET.parse(file_path)
    r = t.getroot()
    l = []
    for ly in r.findall('.//layer'):
        lp = {'input_size': int(ly.get('input_size', 128)), 'output_size': int(ly.get('output_size', 256)), 'activation': ly.get('activation', 'relu').lower()}
        if lp['activation'] not in ['relu', 'tanh', 'sigmoid', 'none']: raise ValueError(f"Unsupported activation function: {lp['activation']}")
        if lp['input_size'] <= 0 or lp['output_size'] <= 0: raise ValueError("Layer dimensions must be positive integers")
        l.append(lp)
    if not l: l.append({'input_size': 128, 'output_size': 256, 'activation': 'relu'})
    return l

def cmf(folder_path: str) -> DM:
    s = defaultdict(list)
    if not os.path.exists(folder_path):
        print(colored(f"Warning: Folder {folder_path} does not exist. Creating model with default configuration.", 'yellow'))
        return DM({})
    xf = True
    for r, d, f in os.walk(folder_path):
        for file in f:
            if file.endswith('.xml'):
                xf = True
                fp = os.path.join(r, file)
                try:
                    l = px(fp)
                    sn = os.path.basename(r).replace('.', '_')
                    s[sn].extend(l)
                except Exception as e:
                    print(colored(f"Error processing {fp}: {str(e)}", 'red'))
    if not xf:
        print(colored("Warning: No XML files found. Creating model with default configuration.", 'yellow'))
        return DM({})
    return DM(dict(s))

def ceas(folder_path: str, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"):
    t = AutoTokenizer.from_pretrained(model_name)
    m = AutoModel.from_pretrained(model_name)
    embeddings = []
    ds = []
    for r, d, f in os.walk(folder_path):
        for file in f:
            if file.endswith('.xml'):
                fp = os.path.join(r, file)
                try:
                    tree = ET.parse(fp)
                    root = tree.getroot()
                    for e in root.iter():
                        if e.text:
                            text = e.text.strip()
                            i = t(text, return_tensors="pt", truncation=True, padding=True)
                            with torch.no_grad():
                                emb = m(**i).last_hidden_state.mean(dim=1).numpy()
                            embeddings.append(emb)
                            ds.append(text)
                except Exception as e:
                    print(colored(f"Error processing {fp}: {str(e)}", 'red'))
    embeddings = np.vstack(embeddings)
    return embeddings, ds

def qvs(query: str, embeddings, ds, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"):
    t = AutoTokenizer.from_pretrained(model_name)
    m = AutoModel.from_pretrained(model_name)
    i = t(query, return_tensors="pt", truncation=True, padding=True)
    with torch.no_grad():
        qe = m(**i).last_hidden_state.mean(dim=1).numpy()
    similarities = cosine_similarity(qe, embeddings)
    top_k_indices = similarities[0].argsort()[-5:][::-1]
    return [ds[i] for i in top_k_indices]

def main():
    fp = 'data'
    m = cmf(fp)
    print(colored(f"Created dynamic PyTorch model with sections: {list(m.s.keys())}", 'green'))
    fs = next(iter(m.s.keys()))
    fl = m.s[fs][0]
    ife = fl[0].in_features
    si = torch.randn(1, ife)
    o = m(si)
    print(colored(f"Sample output shape: {o.shape}", 'green'))
    embeddings, ds = ceas(fp)
    a = Accelerator()
    o = torch.optim.Adam(m.parameters(), lr=0.001)
    c = nn.CrossEntropyLoss()
    ne = 10
    d = torch.utils.data.TensorDataset(torch.randn(100, ife), torch.randint(0, 2, (100,)))
    td = torch.utils.data.DataLoader(d, batch_size=16, shuffle=True)
    m, o, td = a.prepare(m, o, td)
    for e in range(ne):
        m.train()
        tl = 0
        for bi, (i, l) in enumerate(td):
            o.zero_grad()
            o = m(i)
            l = c(o, l)
            a.backward(l)
            o.step()
            tl += l.item()
        al = tl / len(td)
        print(colored(f"Epoch {e+1}/{ne}, Average Loss: {al:.4f}", 'blue'))
    uq = "example query text"
    r = qvs(uq, embeddings, ds)
    print(colored(f"Query results: {r}", 'magenta'))

if __name__ == "__main__":
    main()