Self-QA Technology Overview

Motivation for Self-QA Implementation

The need for generating question-answer pairs stems from training large language models using business-related problems, materials, and documents. Previously, we extracted QA pairs by segmenting and tokenizing structured texts like encyclopedias, where questions and answers followed predictable patterns.

How ever, many valuable resources remain unstructured and未经人工预处理, making traditional extraction methods ineffective. For instance, academic journals containing decades of authoritative research lack clear question-answer formatting. These documents require careful reading and analysis to identify potential questions and their corresponding answers.

Manual processing of such extensive archives would demand excessive human resources and time. Large language models offer an automated solution for analyzing these materials.

Understanding Self-QA Technology

Definition

Self Question-Answering (Self-QA) represents a natural language processing approach that enhances machine comprehension by generating questions and providing answers autonomously. This technique improves machine learning model performance, particularly in Q&A systems, text understanding, and generation tasks.

The core principle involves models analyzing given content, creating relevant questions, and using the same content to generate answers. This process enables self-training without requiring additional annotated data.

Key Characteristics

1. Self-supervised Learning: Models don't require extensive labeled datasets. They self-train through question generation and answering processes.
2. Enhanced Comprehension: Forced deep text analysis improves semantic understanding, crucial for Q&A systems and text summarization.
3. Dataset Creation: Generates large-scale QA datasets automatically, providing valuable training material for NLP models.
4. Improved Interpretability: The question-answer generation process helps researchers understand how models interpret text and make decisions.

Building a Self-QA Application

We'll use the Zhipu AI development platform API, similar to previous implementations described in related documentation.

Fault-Tolerant Architecture for Power Outages

Original source materials follow a hierarchical structure: root directories contain year-based folders, which contain issue-based subfolders, ultimately containing individual articles.

Processing requires traversing this tree structure to handle each document. However, power outages at campus locations create challenges - applications lose progress tracking when systems shut down.

To address this, we implement a fault-tolerant logging system:

1. Complete a folder when all subfolders and files are processed
2. Create "Workstation.txt" in each folder listing subfolder and file names
3. Applications check Workstation.txt to determine remaining tasks
4. Process subfiles via API calls
5. Recursively process subfolders
6. Clear Workstation.txt only after complete folder processing

During power failures, restarting may redundantly process some articles, but ensures no tasks are missed.

For Workstation.txt generation, Java provides efficient DFS implementation:

package Scripts;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;

class DocumentFilter {
    public String fileExtension;

    public DocumentFilter(String extension) {
        this.fileExtension = extension;
    }

    public boolean isValidFile(String filename) {
        return filename.endsWith(fileExtension);
    }
}

public class WorkstationGenerator {
    static DocumentFilter filter;
    
    public static void main(String[] args) throws IOException {
        String baseDir = args[0];

        File directory = new File(baseDir);
        assert directory.isDirectory();

        filter = new DocumentFilter(args[1]);

        traverseDirectory(directory);
    }

    private static void traverseDirectory(File dir) throws IOException {
        StringBuilder contentBuilder = new StringBuilder();

        File[] items = dir.listFiles();
        assert items != null;
        for (File item : items) {
            String name = item.getName();
            if (item.isDirectory()) {
                traverseDirectory(item);
                contentBuilder.append(name).append("\n");
            }
            if (filter.isValidFile(name)) {
                contentBuilder.append(name).append("\n");
            }
        }

        File logFile = new File(dir.getAbsolutePath() + "\\Workstation.txt");
        if (!logFile.exists()) {
            boolean created = logFile.createNewFile();
            assert created;
        }
        try (FileWriter writer = new FileWriter(logFile)) {
            writer.append(contentBuilder.toString());
        }
        System.out.println("Created workspace for " + dir.getName());
    }
}

The strategy pattern handles file selection, focusing on DOCX format due to Python's PDF limitations:

class DocumentFilter {
    public String fileExtension;

    public DocumentFilter(String extension) {
        this.fileExtension = extension;
    }

    public boolean isValidFile(String filename) {
        return filename.endsWith(fileExtension);
    }
}

Recursive Task Processing

Depth-First Search Implementation

Following Workstation.txt guidance:

1. Open current directory's Workstation.txt
2. Read pending tasks
3. Process each task - files call processing functon, folders recurse
4. Clear Workstation.txt upon completion

def process_directory(path):
    log_file = path + '\\' + 'Workstation.txt'
    with open(log_file, 'r+', encoding='UTF-8') as f:
        pending_tasks = f.readlines()
        while len(pending_tasks) > 0:
            current_task = path + '\\' + pending_tasks[0].rstrip()
            if os.path.isdir(current_task):
                process_directory(current_task)
            else:
                execute_processing(current_task)
            pending_tasks = pending_tasks[1:]
        f.truncate(0)

Processing Source Materials

The execute_processing function handles document conversion:

def create_instruction(text: str) -> str:
    instruction = "You are an expert in I Ching studies. Here is an article about I Ching: \n[" + text + "]\n"
    instruction += ("Generate 5 professional and diverse Q&A pairs about I Ching based on this article. "
                   "Questions can be factual or involve understanding and evaluation. Avoid demonstrative pronouns like "
                   "'this' or 'that' in questions.\n")
    instruction += "Format responses as:\n"
    instruction += "Question1: ......\nAnswer1: ......\n\nQuestion2: ......\nAnswer2: ......"
    return instruction

Extract text from DOCX using python-docx:

def extract_text_from_docx(docx_path):
    doc = Document(docx_path)
    text_collection = []

    for paragraph in doc.paragraphs:
        text_collection.append(paragraph.text)
    
    return '\n'.join(text_collection)

API request handling:

    global api_response
    try:
        api_response = client.chat.completions.create(
            model="glm-4",
            messages=[
                {"role": "user",
                 "content": content}
            ],
        )
    except Exception as e:
        print(f"Error occurred: {e}")

Parse generated Q&A pairs:

def parse_qa_pairs(input_str):
    qa_list = []
    pairs = input_str.split("\n\n")

    for i in range(len(pairs)):
        pair = pairs[i]
        lines = pair.strip().split("\n")
        if len(lines) != 2:
            continue
        quest = lines[0].strip("Question" + str(i + 1) + ": ")
        ans = lines[1].strip("Answer" + str(i + 1) + ": ")

        qa_list.append({
            "question": quest,
            "answer": ans
        })

    return qa_list

Deduplicate similar content using ROUGE-L scoring:

def calculate_rouge_l_score(reference, candidate):
    rouge = Rouge()
    scores = rouge.get_scores(candidate, reference, avg=True)
    return scores['rouge-l']


def merge_unique_content(base_list, new_content, threshold=0.7):
    for new_item in new_content:
        new_text = json_to_string(new_item)
        similar_found = False

        for existing_item in base_list:
            existing_text = json_to_string(existing_item)
            rouge_score = calculate_rouge_l_score(existing_text, new_text)
            print(rouge_score)

            if rouge_score['f'] > threshold:
                similar_found = True
                break

        if not similar_found:
            base_list.append(new_item)

The ROUGE-L metric uses Longest Common Subsequence for similarity assessment between longer texts. F1 scores above 0.7 indicate high similarity requiring exclusion.

Store results in JSON format:

    qa_results = parse_qa_pairs(response_text)
    with open('./self_QA/output/self_qa.json', 'r+', encoding='utf-8') as output_file:
        existing_content = output_file.read()
        processed_data = []
        if existing_content:
            output_file.seek(0)
            processed_data = json.load(output_file)
        final_json = convert_qa_to_json(qa_results, processed_data)
        output_file.truncate(0)
        output_file.seek(0)
        output_file.write(final_json)

Potential Issues

Language models occasionally deviate from expected formats. Instead of single blank lines between Q&A pairs, they might insert spaces between questions and answers within pairs. Such formatting inconsistencies become less frequent with increased processing volume.

Optimization opportunities exist in improving string pattern matching for more accurate Q&A extraction.

Repository and Reproduction

Source code available at: https://github.com/Liyanhao1209/ZhouYiLLM.git

Branches include java_scripts and python_scripts. When running self_qa scripts, first generate Workstation.txt using the Java script from the java_scripts branch.