How to Build a Text Correction and Grammar Checking Pipeline

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import language_tool_python

# Initialize the tool (downloads ~200MB on first run)
tool = language_tool_python.LanguageTool('en-US')

# Check text
text = "She dont like pizza. He go to school everyday."
matches = tool.check(text)

# Apply corrections
corrected = language_tool_python.utils.correct(text, matches)
print(corrected)
# Output: "She doesn't like pizza. He goes to school every day."

# Get detailed error info
for match in matches:
    print(f"Error: {match.ruleId}")
    print(f"Message: {match.message}")
    print(f"Suggestion: {match.replacements}")
    print(f"Position: {match.offset}-{match.errorLength}")
    print("---")

tool.close()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

import language_tool_python
from typing import Dict, List, Tuple
import re

class TextCorrectionPipeline:
    def __init__(self, language='en-US'):
        self.tool = language_tool_python.LanguageTool(language)
        self.stats = {
            'spelling': 0,
            'grammar': 0,
            'style': 0,
            'total_corrections': 0
        }

    def correct(self, text: str, explain: bool = False) -> Dict:
        """
        Multi-stage correction with detailed tracking.

        Args:
            text: Input text to correct
            explain: Return correction explanations

        Returns:
            Dict with corrected text, stats, and optional explanations
        """
        matches = self.tool.check(text)
        corrected = language_tool_python.utils.correct(text, matches)

        # Categorize errors
        error_types = self._categorize_errors(matches)

        # Update stats
        for category in error_types:
            self.stats[category] += len(error_types[category])
        self.stats['total_corrections'] += len(matches)

        result = {
            'original': text,
            'corrected': corrected,
            'num_corrections': len(matches),
            'stats': self.stats.copy()
        }

        if explain:
            result['explanations'] = [
                {
                    'rule': m.ruleId,
                    'message': m.message,
                    'original': text[m.offset:m.offset+m.errorLength],
                    'suggestion': m.replacements[0] if m.replacements else None
                }
                for m in matches
            ]

        return result

    def _categorize_errors(self, matches) -> Dict[str, List]:
        """Group errors by type based on rule ID."""
        categories = {
            'spelling': [],
            'grammar': [],
            'style': []
        }

        for match in matches:
            rule_id = match.ruleId.upper()
            if 'SPELL' in rule_id or 'MORFOLOGIK' in rule_id:
                categories['spelling'].append(match)
            elif 'STYLE' in rule_id or 'REDUNDANCY' in rule_id:
                categories['style'].append(match)
            else:
                categories['grammar'].append(match)

        return categories

    def batch_correct(self, texts: List[str]) -> List[Dict]:
        """Process multiple documents efficiently."""
        return [self.correct(text) for text in texts]

    def __del__(self):
        """Clean up on object destruction."""
        if hasattr(self, 'tool'):
            self.tool.close()

# Usage
pipeline = TextCorrectionPipeline()

test_text = """
Theyre going to they're friends house tomorrow.
Its important to remember it's proper usage.
Me and him went to the store yesterday.
"""

result = pipeline.correct(test_text, explain=True)
print(result['corrected'])
print(f"\nStats: {result['stats']}")

# Print explanations
for exp in result['explanations']:
    print(f"\n'{exp['original']}' -> '{exp['suggestion']}'")
    print(f"  Reason: {exp['message']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

# Install: pip install gector

from gector.gec_model import GecBERTModel
import torch

# Download pretrained model (first run)
# Model files: https://github.com/grammarly/gector

class NeuralCorrector:
    def __init__(self, model_path='gector-roberta', device='cpu'):
        """
        Initialize GECToR model.

        Note: Download models from:
        https://github.com/grammarly/gector/tree/master/pretrained_models
        """
        self.model = GecBERTModel(
            vocab_path=f'{model_path}/vocabulary',
            model_paths=[f'{model_path}/model.th'],
            max_len=50,
            min_len=3,
            iterations=5,
            min_error_probability=0.0,
            lowercase_tokens=0,
            model_name='roberta',
            special_tokens_fix=1,
            log=False,
            confidence=0,
            is_ensemble=0,
            weigths=None
        )
        self.device = device

    def correct(self, text: str) -> Tuple[str, int]:
        """
        Correct text using neural model.

        Returns:
            Tuple of (corrected_text, num_edits)
        """
        # Split into sentences for better performance
        sentences = text.split('. ')
        corrected_sentences = []
        total_edits = 0

        for sent in sentences:
            if not sent.strip():
                continue

            # Model expects list of tokens
            tokens = sent.split()
            corrected_batch, edits = self.model.handle_batch([tokens])
            corrected_sentences.append(' '.join(corrected_batch[0]))
            total_edits += len(edits[0])

        return '. '.join(corrected_sentences), total_edits

# Usage example (requires model download)
# corrector = NeuralCorrector()
# text = "He are going to store for buy some milk."
# corrected, edits = corrector.correct(text)
# print(f"Corrected: {corrected}")
# print(f"Edits made: {edits}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

from openai import OpenAI
import os
from typing import Optional

class LLMCorrector:
    def __init__(self, api_key: Optional[str] = None):
        self.client = OpenAI(api_key=api_key or os.getenv('OPENAI_API_KEY'))

    def correct(self, text: str, style: str = 'professional') -> Dict:
        """
        Correct and improve text using GPT-4.

        Args:
            text: Input text
            style: Target style - 'professional', 'casual', 'academic', 'concise'

        Returns:
            Dict with corrected text and explanation
        """
        style_prompts = {
            'professional': 'Fix grammar and spelling. Keep a professional business tone.',
            'casual': 'Fix errors but keep the casual, conversational tone.',
            'academic': 'Correct errors and use formal academic language.',
            'concise': 'Fix errors and make the text more concise without losing meaning.'
        }

        prompt = f"""
{style_prompts.get(style, style_prompts['professional'])}

Original text:
{text}

Provide:
1. Corrected text
2. Brief explanation of key changes made

Format your response as:
CORRECTED: <corrected text>
CHANGES: <explanation>
"""

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert editor who improves text quality."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        result_text = response.choices[0].message.content

        # Parse response
        lines = result_text.split('\n')
        corrected = ""
        changes = ""

        for i, line in enumerate(lines):
            if line.startswith('CORRECTED:'):
                corrected = line.replace('CORRECTED:', '').strip()
                # Get remaining lines until CHANGES
                for j in range(i+1, len(lines)):
                    if lines[j].startswith('CHANGES:'):
                        break
                    corrected += ' ' + lines[j].strip()
            elif line.startswith('CHANGES:'):
                changes = line.replace('CHANGES:', '').strip()
                # Get remaining lines
                for j in range(i+1, len(lines)):
                    changes += ' ' + lines[j].strip()

        return {
            'original': text,
            'corrected': corrected.strip(),
            'changes': changes.strip(),
            'model': 'gpt-4'
        }

# Usage
corrector = LLMCorrector()

text = """
The data shows that customers is more satisfied when they receives
personalized recommendations. However, implementation of this features
has been challenging due to lack of resources.
"""

result = corrector.correct(text, style='professional')
print(result['corrected'])
print(f"\nChanges made:\n{result['changes']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

import hashlib
import json
from pathlib import Path
from typing import List, Dict
import concurrent.futures

class BatchTextProcessor:
    def __init__(self, cache_dir: str = '.correction_cache'):
        self.pipeline = TextCorrectionPipeline()
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def _get_cache_key(self, text: str) -> str:
        """Generate cache key from text hash."""
        return hashlib.md5(text.encode()).hexdigest()

    def _get_cached(self, text: str) -> Optional[Dict]:
        """Retrieve cached result if available."""
        cache_key = self._get_cache_key(text)
        cache_file = self.cache_dir / f"{cache_key}.json"

        if cache_file.exists():
            with open(cache_file) as f:
                return json.load(f)
        return None

    def _save_cache(self, text: str, result: Dict):
        """Save result to cache."""
        cache_key = self._get_cache_key(text)
        cache_file = self.cache_dir / f"{cache_key}.json"

        with open(cache_file, 'w') as f:
            json.dump(result, f)

    def process_file(self, file_path: str) -> Dict:
        """Process a single text file."""
        with open(file_path) as f:
            text = f.read()

        # Check cache
        cached = self._get_cached(text)
        if cached:
            return {**cached, 'from_cache': True}

        # Process
        result = self.pipeline.correct(text)

        # Cache result
        self._save_cache(text, result)

        return {**result, 'from_cache': False}

    def process_directory(self, dir_path: str, pattern: str = '*.txt',
                         max_workers: int = 4) -> List[Dict]:
        """
        Process all matching files in directory using parallel workers.

        Args:
            dir_path: Directory to process
            pattern: File pattern to match
            max_workers: Number of parallel workers

        Returns:
            List of results for each file
        """
        files = list(Path(dir_path).glob(pattern))
        results = []

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_file = {
                executor.submit(self.process_file, str(f)): f
                for f in files
            }

            for future in concurrent.futures.as_completed(future_to_file):
                file_path = future_to_file[future]
                try:
                    result = future.get()
                    result['file'] = str(file_path)
                    results.append(result)

                    cache_status = "cached" if result['from_cache'] else "processed"
                    print(f"✓ {file_path.name} ({cache_status})")
                except Exception as e:
                    print(f"✗ {file_path.name}: {e}")

        return results

# Usage
processor = BatchTextProcessor()

# Process single file
result = processor.process_file('document.txt')
print(f"Corrections: {result['num_corrections']}")

# Process entire directory
results = processor.process_directory('documents/', pattern='*.txt', max_workers=8)
print(f"\nProcessed {len(results)} files")
print(f"Total corrections: {sum(r['num_corrections'] for r in results)}")

1
2
3
4
5
6
7

# Set proxy before importing
import os
os.environ['HTTP_PROXY'] = 'http://proxy.company.com:8080'
os.environ['HTTPS_PROXY'] = 'http://proxy.company.com:8080'

import language_tool_python
tool = language_tool_python.LanguageTool('en-US')

1
2
3
4
5
6
7
8
9

import gc

def process_large_batch(texts: List[str], chunk_size: int = 100):
    results = []
    for i in range(0, len(texts), chunk_size):
        chunk = texts[i:i+chunk_size]
        results.extend(pipeline.batch_correct(chunk))
        gc.collect()  # Force cleanup between chunks
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def correct_long_text(text: str, max_chunk_length: int = 5000) -> str:
    # Split on paragraph breaks
    paragraphs = text.split('\n\n')
    corrected_paragraphs = []

    for para in paragraphs:
        if len(para) > max_chunk_length:
            # Split long paragraphs on sentence boundaries
            sentences = para.split('. ')
            corrected = '. '.join([
                tool.correct(sent) for sent in sentences
            ])
        else:
            corrected = tool.correct(para)
        corrected_paragraphs.append(corrected)

    return '\n\n'.join(corrected_paragraphs)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# Alternative: use transformers library with GECToR-style models
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM

tokenizer = AutoTokenizer.from_pretrained("grammarly/coedit-large")
model = AutoModelForSeq2SeqLM.from_pretrained("grammarly/coedit-large")

def correct_with_coedit(text: str) -> str:
    input_text = f"Fix grammatical errors in this sentence: {text}"
    inputs = tokenizer(input_text, return_tensors="pt")
    outputs = model.generate(**inputs, max_length=256)
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import time
from openai import RateLimitError

def correct_with_retry(text: str, max_retries: int = 3) -> Dict:
    for attempt in range(max_retries):
        try:
            return corrector.correct(text)
        except RateLimitError:
            wait_time = (2 ** attempt) + random.random()
            print(f"Rate limited. Waiting {wait_time:.2f}s...")
            time.sleep(wait_time)
    raise Exception("Max retries exceeded")