Interrogazioni basate sull’intelligenza artificiale per la scoperta della conoscenza basata sul linguaggio naturale

Esplorando la conoscenza attraverso l'intelligenza artificiale interrogazioni basate sul linguaggio naturale

In questo articolo, volevo condividere un progetto di prova a cui ho lavorato chiamato UE5_documentalist. Si tratta di un progetto entusiasmante che utilizza l’Elaborazione del Linguaggio Naturale (NLP) per potenziare potenzialmente la tua esperienza con la documentazione massiccia.

Mentre lavoravo alla documentazione di Unreal Engine 5 per questo progetto, può essere applicato a qualsiasi altro caso d’uso, come la documentazione interna della tua azienda.

Cos’è UE5_documentalist?

UE5_documentalist è un assistente intelligente progettato per semplificare la tua navigazione in Unreal Engine 5.1 o qualsiasi altra documentazione. Sfruttando le tecniche di NLP, questo progetto ti consente di fare query in linguaggio naturale e trovare facilmente le sezioni più rilevanti in oltre 1700 pagine web.

Ad esempio, puoi fare una query del tipo “quale sistema posso usare per evitare collisioni tra agenti?” e sarai indirizzato alla documentazione più adatta alle tue esigenze.

Puoi controllare il mio codice su questo repository.

Dimostrazione

Per darti un’idea migliore di cosa può fare UE5_documentalist, ecco una breve dimostrazione delle sue capacità:

Demo (Immagine dell’autore)

Come funziona?

Passaggio 1 – Scraping

Innanzitutto ho effettuato uno scraping della documentazione web di Unreal Engine 5.1, convertito l’output HTML in markdown e salvato il testo in un dizionario.

La funzione di scraping si presenta così:

def main(limit, urls_registry, subsections_path):    urls_registry = "./src/utils/urls.txt"    with open(urls_registry, 'r') as f:        urls = f.read()    urls = urls.split('\n')    # inizializza il dizionario per memorizzare le sezioni    subsections = {}    for idx, url in enumerate(urls):        # ferma se è stato raggiunto il limite        if limit is not None:            if idx > limit:                break        if idx % 100 == 0:            print(f"Elaborazione dell'url {idx}")        # effettua la richiesta        try:            with urllib.request.urlopen(url) as f:                content = f.read()        except HTTPError as e:            print(f"Errore con l'url {url}")            print('Il server non ha potuto soddisfare la richiesta.')            print('Codice di errore: ', e.code)            continue        except URLError as e:            print(f"Errore con l'url {url}")            print('Impossibile raggiungere il server.')            print('Motivo: ', e.reason)            continue        # elabora il contenuto        md_content = md(content.decode('utf-8'))        preproc_content = split_text_into_components(md_content)        # estrai informazioni dal nome dell'url        subsection_title = extract_info_from_url(url)        # aggiungi al dizionario        subsections[url] = {            "title": subsection_title,            "content": preproc_content        }    # salva il dizionario    with open(subsections_path, 'w') as f:        json.dump(subsections, f)

La funzione chiama un’altra funzione chiamata split_text_into_components, che è una sequenza di passaggi di preprocessing che rimuovono caratteri speciali come link, immagini, testo in grassetto e intestazioni.

Alla fine, questa funzione restituisce un dizionario che si presenta così:

{url :   {'titolo' : 'un_titolo_contextuale',   'contenuto' : 'contenuto della pagina'  }...}

Passaggio 2 – Embeddings

Da lì, ho generato gli embeddings utilizzando il modello Instructor-XL. All’inizio, ho utilizzato il modello text-embedding-ada-002 di OpenAI. Era estremamente economico (meno di $2 per oltre 170 milioni di caratteri), veloce e globalmente adatto al mio caso d’uso. Tuttavia, sono passato al modello Instructor-XL, che funziona leggermente più lentamente ma può essere utilizzato localmente senza comunicare con un’API online. Questa soluzione è migliore se hai preoccupazioni sulla privacy o lavori con dati sensibili.

La funzione di embedding si presenta così:

def embed(subsection_dict_path, embedder, security):    """Incorpora i file nella directory.    Args:        subsection_dict_path (dict): Percorso del dizionario contenente le sottosezioni.        security (str): Impostazione di sicurezza. Sia "attivato" o "disattivato".                        impedisce l'esecuzione della funzione se non "disattivato"                          e evita costi imprevisti.    Returns:        embeddings (dict): Dizionario contenente gli embeddings.    """    # Se gli embeddings esistono già, caricali    if os.path.exists(os.path.join("./embeddings", f'{embedder}_embeddings.json')):        print("Gli embeddings esistono già. Li sto caricando.")        with open(os.path.join("./embeddings", f'{embedder}_embeddings.json'), 'r') as f:            embeddings = json.load(f)    else:        # Inizializza il dizionario per memorizzare gli embeddings        embeddings = {}    # controlla la sicurezza se l'embedder è openai (evita di spendere $$$ per errore)    if security != "disattivato":        if embedder == 'openai':            raise Exception("La sicurezza non è disattivata.")        # carica le sottosezioni    with open(subsection_dict_path, 'r') as f:        subsection_dict = json.load(f)    # Solo per scopi di debug    # Calcola la lunghezza media del testo da incorporare    dict_len = len(subsection_dict)    total_text_len = 0    for url, subsection in subsection_dict.items():        total_text_len += len(subsection['content'])    avg_text_len = total_text_len / dict_len    # Inizializza l'API di OpenAI se l'embedder è 'openai'    if embedder == "openai":        openai_model = "text-embedding-ada-002"        # Ottieni l'API key dalla variabile d'ambiente o chiedi all'utente di inserirla        api_key = os.getenv('API_KEY')        if api_key is None:            api_key = input("Per favore, inserisci la tua chiave API di OpenAI: ")        openai.api_key = api_key    # Inizializza il modello di istruttore se l'embedder è 'instructor'    elif embedder == "instructor":        instructor_model = INSTRUCTOR('hkunlp/instructor-xl')        # Imposta il dispositivo su gpu se disponibile        if (torch.backends.mps.is_available()) and (torch.backends.mps.is_built()):            device = torch.device("mps")        elif torch.cuda.is_available():            device = torch.device("cuda")        else:            device = torch.device("cpu")    else:        raise ValueError(f"L'embedder deve essere 'openai' o 'instructor'. Non {embedder}")        # scorri le sottosezioni    for url, subsection in tqdm(subsection_dict.items()):        subsection_name = subsection['title']        text_to_embed = subsection['content']        # salta se già incorporato        if url in embeddings.keys():            continue        # effettua la richiesta di embedding        # caso 1: openai        if embedder == 'openai':            try:                response = openai.Embedding.create(                    input=text_to_embed,                    model=openai_model                )                embedding = response['data'][0]['embedding']            except InvalidRequestError as e:                print(f"Errore con URL {url}")                print("Il server non ha potuto soddisfare la richiesta.")                print('Codice errore: ', e.code)                print(f"Sono stati incorporati {len(text_to_embed)} caratteri mentre in media ce ne sono {avg_text_len}")                continue        # caso 2: instructor        elif embedder == 'instructor':            instruction = "Rappresenta la documentazione UnrealEngine per il recupero:"            embedding = instructor_model.encode([[instruction, text_to_embed]], device=device)            embedding = [float(x) for x in embedding.squeeze().tolist()]        else:            raise ValueError(f"L'embedder deve essere 'openai' o 'instructor'. Non {embedder}")                # aggiungi l'embedding al dizionario        embeddings[url] = {            "title": subsection_name,            "embedding": embedding        }        # salva il dizionario ogni 100 iterazioni        if len(embeddings) % 100 == 0:            print(f"Salvataggio degli embeddings dopo {len(embeddings)} iterazioni.")            # salva gli embeddings in un file pickle            with open(os.path.join("./embeddings", f'{embedder}_embeddings.pkl'), 'wb') as f:                pickle.dump(embeddings, f)            # salva gli embeddings in un file json            with open(os.path.join("./embeddings", f'{embedder}_embeddings.json'), 'w') as f:                json.dump(embeddings, f)    return embeddings

Questa funzione restituisce un dizionario che si presenta così:

{url :   {'title' : 'some_contextual_title',   'embedding' : vector representation of the content  }...}

Passo 3 — Database di indice vettoriale

Questi embedding sono stati quindi caricati in un database di indice vettoriale Qdrant in esecuzione su un contenitore Docker.

Il database viene avviato con i seguenti comandi Docker:

docker pull qdrant/qdrantdocker run -d -p 6333:6333 qdrant/qdrant

E viene popolato con la seguente funzione (è necessario il pacchetto qdrant_client: pip install qdrant-client)

import qdrant_client as qcimport qdrant_client.http.models as qmodelsimport uuidimport jsonimport argparsefrom tqdm import tqdmclient = qc.QdrantClient(url="localhost")METRIC = qmodels.Distance.DOTCOLLECTION_NAME = "ue5_docs"def create_index():    client.recreate_collection(    collection_name=COLLECTION_NAME,    vectors_config = qmodels.VectorParams(            size=DIMENSION,            distance=METRIC,        )    )def create_subsection_vector(    subsection_content,    section_anchor,    page_url    ):    id = str(uuid.uuid1().int)[:32]    payload = {        "text": subsection_content,        "url": page_url,        "section_anchor": section_anchor,        "block_type": 'text'    }    return id, payloaddef add_doc_to_index(embeddings, content_dict):    ids = []    vectors = []    payloads = []        for url, content in tqdm(embeddings.items()):        section_anchor = content['title']        section_vector = content['embedding']        section_content = content_dict[url]['content']        id, payload = create_subsection_vector(            section_content,            section_anchor,            url    )        ids.append(id)        vectors.append(section_vector)        payloads.append(payload)        # Aggiungi vettori alla collezione        client.upsert(            collection_name=COLLECTION_NAME,            points=qmodels.Batch(                ids = [id],                vectors=[section_vector],                payloads=[payload]            ),        )

Questa funzione prende l’URL della pagina web e il contenuto associato, lo incorpora e lo carica nel database Qdrant.

Devi caricare un ID, un vettore e un payload come database di vettori. Qui, i nostri ID vengono generati proceduralmente, i nostri vettori sono gli embedding (che verranno abbinati alle interrogazioni in seguito) e il payload è l’informazione aggiuntiva.

Nel nostro caso, le informazioni principali da includere nel payload sono l’URL web e il contenuto della pagina web. In questo modo, quando abbiniamo una query all’ingresso più rilevante nel nostro database Qdrant, possiamo stampare la documentazione e aprire la pagina web.

Passo finale – Interrogazione

Possiamo ora fare una query al database.

La query deve prima essere incorporata con lo stesso embedder utilizzato nella documentazione. Facciamo ciò con la seguente funzione:

def embed_query(query, embedder):    if embedder == "openai":        # Recupera la chiave API dalla variabile di ambiente o richiedi all'utente di inserirla        api_key = os.getenv('API_KEY')        if api_key is None:            api_key = input("Inserisci la tua chiave API OpenAI: ")                openai_model = "text-embedding-ada-002"        openai.api_key = api_key        response = openai.Embedding.create(                    input=query,                    model=openai_model        )        embedding = response['data'][0]['embedding']    elif embedder == "instructor":        instructor_model = INSTRUCTOR('hkunlp/instructor-xl')         # Imposta il dispositivo su GPU se disponibile        if (torch.backends.mps.is_available()) and (torch.backends.mps.is_built()):            device = torch.device("mps")        elif torch.cuda.is_available():            device = torch.device("cuda")        else:            device = torch.device("cpu")        instruction = "Rappresenta la query UnrealEngine per recuperare i documenti di supporto:"        embedding = instructor_model.encode([[instruction, query]], device=device)        embedding = [float(x) for x in embedding.squeeze().tolist()]    else:        raise ValueError("L'embedder deve essere 'openai' o 'instructor'")    return embedding

La query incorporata viene quindi utilizzata per interrogare il database Qdrant:

def query_index(query, embedder, top_k=10, block_types=None):    """    Interroga il database di indice vettoriale Qdrant per i documenti che corrispondono alla query fornita.    Args:        query (str): La query da cercare.        embedder (str): L'embedder da utilizzare. Deve essere "openai" o "instructor".        top_k (int, opzionale): Il numero massimo di documenti da restituire. Predefinito a 10.        block_types (str o lista di str, opzionale): I tipi di blocchi di documento da cercare. Predefinito a "text".    Returns:        Una lista di dizionari che rappresentano i documenti corrispondenti, ordinati per rilevanza. Ogni dizionario contiene le seguenti chiavi:        - "id": L'ID del documento.        - "score": Il punteggio di rilevanza del documento.        - "text": Il contenuto testuale del documento.        - "block_type": Il tipo di blocco di documento che corrisponde alla query.    """    collection_name = get_collection_name()    if not collection_exists(collection_name):        raise Exception(f"La collezione {collection_name} non esiste. Le collezioni esistenti sono: {list_collections()}")    vector = embed_query(query, embedder)    _search_params = models.SearchParams(        hnsw_ef=128,        exact=False    )    block_types = parse_block_types(block_types)    _filter = models.Filter(        must=[            models.Filter(                should= [                    models.FieldCondition(                        key="block_type",                        match=models.MatchValue(value=bt),                    )                for bt in block_types                ]              )        ]    )    results = CLIENT.search(        collection_name=collection_name,        query_vector=vector,        query_filter=_filter,        limit=top_k,        with_payload=True,        search_params=_search_params,    )    results = [        (            f"{res.payload['url']}#{res.payload['section_anchor']}",             res.payload["text"],            res.score        )        for res in results    ]    return resultsdef ue5_docs_search(    query,     embedder=None,    top_k=10,     block_types=None,    score=False,    open_url=True):    """    Cerca nel database di indice vettoriale Qdrant i documenti correlati alla query fornita e stampa i risultati principali.    Args:        query (str): La query da cercare.        embedder (str): L'embedder da utilizzare. Deve essere "openai" o "instructor".        top_k (int, opzionale): Il numero massimo di documenti da restituire. Predefinito a 10.        block_types (str o lista di str, opzionale): I tipi di blocchi di documento da cercare. Predefinito a "text".        score (bool, opzionale): Se includere o meno il punteggio di rilevanza nell'output. Predefinito a False.        open_url (bool, opzionale): Se aprire o meno l'URL principale in un browser web. Predefinito a True.    Returns:        None    """    # Verifica se l'embedder è 'openai' o 'instructor'. Genera un errore se non lo è    assert embedder in ['openai', 'instructor'], f"L'embedder deve essere 'openai' o 'instructor'. Non {embedder}"    results = query_index(        query,        embedder=embedder,        top_k=top_k,        block_types=block_types    )    print_results(query, results, score=score)    if open_url:        top_url = results[0][0]        webbrowser.open(top_url)

Queste funzioni richiamano altre funzioni di elaborazione interne trovate nel mio repository.

Ecco fatto! La query è incorporata, abbinata all’incorporamento della documentazione più vicina (puoi scegliere la metrica di valutazione per questo, ma per questo caso d’uso ho scelto il prodotto scalare). Il risultato viene stampato nel terminale e la pagina web associata si apre automaticamente!

Vuoi provarlo tu stesso?

Se sei interessato a esplorare UE5_documentalist, ho preparato una guida passo-passo completa su come configurarlo sulla tua macchina locale. Gli incorporamenti sono già stati effettuati, quindi puoi iniziare direttamente a popolare il database di Qdrant e troverai tutte le risorse necessarie e istruzioni dettagliate nel mio repository GitHub.

Per ora, funziona con comandi Python.

Perché provarlo?

UE5_documentalist mira a semplificare la ricerca in documentazioni lunghe, risparmiandoti potenzialmente tempo prezioso nello sviluppo. Ponendo domande in inglese semplice, sarai indirizzato alle sezioni specifiche che riguardano le tue richieste. Questo strumento potrebbe migliorare la tua esperienza e consentirti di concentrarti di più nella realizzazione di progetti straordinari con Unreal Engine 5.1 o sulla tua documentazione.

Sono orgoglioso dei miei progressi con UE5_documentalist finora e sono ansioso di sentire i tuoi feedback e suggerimenti per migliorarlo.

TL;DR:

Presentando UE5_documentalist – un assistente intelligente per la documentazione alimentato da NLP. Consente agli sviluppatori di navigare senza sforzo nella documentazione di Unreal Engine 5.1 utilizzando interrogazioni in linguaggio naturale. Dai un’occhiata alla demo e trova le istruzioni per configurarlo sulla tua macchina nel repository GitHub. Chissà, se funziona bene, potresti dire addio alle fastidiose ricerche di documentazione e migliorare potenzialmente la tua esperienza di scoperta della conoscenza!

Repository GitHub

Esempio di utilizzo (Gif)

Credits

Ho avuto l’idea di semplificare la documentazione di UE5 da un utente di Reddit che ha avviato un progetto utilizzando LLM (ho dimenticato il suo nome, ma ecco un link al suo repository).

Ho basato la mia implementazione su questo articolo TDS di Jacob Marks. Il suo codice è stato estremamente utile per la maggior parte delle fasi di analisi e indicizzazione e non avrei avuto successo senza il suo articolo dettagliato. Non esitare a dare un’occhiata al suo lavoro!

Articolo originariamente pubblicato qui. Ripubblicato con il permesso dell’autore.