Separate key terms from other training data on S3 (#846)

pmachapman · web-flow · commit f5919f2004e8 · 2025-12-16T13:51:40.000+13:00
diff --git a/src/Echo/src/EchoEngine/TranslationEngineServiceV1.cs b/src/Echo/src/EchoEngine/TranslationEngineServiceV1.cs
@@ -124,7 +124,7 @@ await client.BuildStartedAsync(
                     List<InsertPretranslationsRequest> pretranslationsRequests = [];
                     await _parallelCorpusPreprocessingService.PreprocessAsync(
                         request.Corpora.Select(Map).ToList(),
-                        row => Task.CompletedTask,
+                        (_, _) => Task.CompletedTask,
                         (row, _, corpus) =>
                         {
                             string[] tokens = row.SourceSegment.Split();
diff --git a/src/Echo/src/EchoEngine/WordAlignmentEngineServiceV1.cs b/src/Echo/src/EchoEngine/WordAlignmentEngineServiceV1.cs
@@ -79,7 +79,7 @@ await client.BuildStartedAsync(
                     List<InsertWordAlignmentsRequest> wordAlignmentsRequests = [];
                     await _parallelCorpusPreprocessingService.PreprocessAsync(
                         request.Corpora.Select(Map).ToList(),
-                        row => Task.CompletedTask,
+                        (_, _) => Task.CompletedTask,
                         (row, _, corpus) =>
                         {
                             wordAlignmentsRequests.Add(
diff --git a/src/Machine/src/Serval.Machine.Shared/Services/SmtTransferTrainBuildJob.cs b/src/Machine/src/Serval.Machine.Shared/Services/SmtTransferTrainBuildJob.cs
@@ -38,8 +38,14 @@ CancellationToken cancellationToken
         await DownloadDataAsync(buildId, corpusDir, cancellationToken);
 
         // assemble corpus
-        ITextCorpus sourceCorpus = new TextFileTextCorpus(Path.Combine(corpusDir, "train.src.txt"));
-        ITextCorpus targetCorpus = new TextFileTextCorpus(Path.Combine(corpusDir, "train.trg.txt"));
+        ITextCorpus sourceCorpus = new TextFileTextCorpus(
+            Path.Combine(corpusDir, "train.src.txt"),
+            Path.Combine(corpusDir, "train.key-terms.src.txt")
+        );
+        ITextCorpus targetCorpus = new TextFileTextCorpus(
+            Path.Combine(corpusDir, "train.trg.txt"),
+            Path.Combine(corpusDir, "train.key-terms.trg.txt")
+        );
         IParallelTextCorpus parallelCorpus = sourceCorpus.AlignRows(targetCorpus);
 
         // train SMT model
@@ -106,6 +112,20 @@ private async Task DownloadDataAsync(string buildId, string corpusDir, Cancellat
         );
         await using FileStream tgtFileStream = File.Create(Path.Combine(corpusDir, "train.trg.txt"));
         await tgtText.CopyToAsync(tgtFileStream, cancellationToken);
+
+        await using Stream srcKeyTermsText = await _sharedFileService.OpenReadAsync(
+            $"builds/{buildId}/train.key-terms.src.txt",
+            cancellationToken
+        );
+        await using FileStream srcKeyTermsFileStream = File.Create(Path.Combine(corpusDir, "train.key-terms.src.txt"));
+        await srcKeyTermsText.CopyToAsync(srcKeyTermsFileStream, cancellationToken);
+
+        await using Stream tgtKeyTermsText = await _sharedFileService.OpenReadAsync(
+            $"builds/{buildId}/train.key-terms.trg.txt",
+            cancellationToken
+        );
+        await using FileStream tgtKeyTermsFileStream = File.Create(Path.Combine(corpusDir, "train.key-terms.trg.txt"));
+        await tgtKeyTermsFileStream.CopyToAsync(tgtKeyTermsText, cancellationToken);
     }
 
     private async Task<(int TrainCorpusSize, double Confidence)> TrainAsync(
diff --git a/src/Machine/src/Serval.Machine.Shared/Services/StatisticalTrainBuildJob.cs b/src/Machine/src/Serval.Machine.Shared/Services/StatisticalTrainBuildJob.cs
@@ -38,8 +38,14 @@ CancellationToken cancellationToken
         await DownloadDataAsync(buildId, corpusDir, cancellationToken);
 
         // assemble corpus
-        ITextCorpus sourceCorpus = new TextFileTextCorpus(Path.Combine(corpusDir, "train.src.txt"));
-        ITextCorpus targetCorpus = new TextFileTextCorpus(Path.Combine(corpusDir, "train.trg.txt"));
+        ITextCorpus sourceCorpus = new TextFileTextCorpus(
+            Path.Combine(corpusDir, "train.src.txt"),
+            Path.Combine(corpusDir, "train.key-terms.src.txt")
+        );
+        ITextCorpus targetCorpus = new TextFileTextCorpus(
+            Path.Combine(corpusDir, "train.trg.txt"),
+            Path.Combine(corpusDir, "train.key-terms.trg.txt")
+        );
         IParallelTextCorpus parallelCorpus = sourceCorpus.AlignRows(targetCorpus);
 
         // train word alignment model
@@ -100,6 +106,20 @@ private async Task DownloadDataAsync(string buildId, string corpusDir, Cancellat
         );
         await using FileStream tgtFileStream = File.Create(Path.Combine(corpusDir, "train.trg.txt"));
         await tgtText.CopyToAsync(tgtFileStream, cancellationToken);
+
+        await using Stream srcKeyTermsText = await _sharedFileService.OpenReadAsync(
+            $"builds/{buildId}/train.key-terms.src.txt",
+            cancellationToken
+        );
+        await using FileStream srcKeyTermsFileStream = File.Create(Path.Combine(corpusDir, "train.key-terms.src.txt"));
+        await srcKeyTermsText.CopyToAsync(srcKeyTermsFileStream, cancellationToken);
+
+        await using Stream tgtKeyTermsText = await _sharedFileService.OpenReadAsync(
+            $"builds/{buildId}/train.key-terms.trg.txt",
+            cancellationToken
+        );
+        await using FileStream tgtKeyTermsFileStream = File.Create(Path.Combine(corpusDir, "train.key-terms.trg.txt"));
+        await tgtKeyTermsFileStream.CopyToAsync(tgtKeyTermsText, cancellationToken);
     }
 
     private async Task<int> TrainAsync(
diff --git a/src/Machine/src/Serval.Machine.Shared/Services/TranslationPreprocessBuildJob.cs b/src/Machine/src/Serval.Machine.Shared/Services/TranslationPreprocessBuildJob.cs
@@ -35,6 +35,11 @@ CancellationToken cancellationToken
         await using StreamWriter targetTrainWriter =
             new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.trg.txt", cancellationToken));
 
+        await using StreamWriter sourceKeyTermsTrainWriter =
+            new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.key-terms.src.txt", cancellationToken));
+        await using StreamWriter targetKeyTermsTrainWriter =
+            new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.key-terms.trg.txt", cancellationToken));
+
         await using Stream pretranslateStream = await SharedFileService.OpenWriteAsync(
             $"builds/{buildId}/pretranslate.src.json",
             cancellationToken
@@ -46,12 +51,20 @@ CancellationToken cancellationToken
         pretranslateWriter.WriteStartArray();
         await ParallelCorpusPreprocessingService.PreprocessAsync(
             corpora,
-            async row =>
+            async (row, trainingDataType) =>
             {
                 if (row.SourceSegment.Length > 0 || row.TargetSegment.Length > 0)
                 {
-                    await sourceTrainWriter.WriteAsync($"{row.SourceSegment}\n");
-                    await targetTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    if (trainingDataType == TrainingDataType.KeyTerms)
+                    {
+                        await sourceKeyTermsTrainWriter.WriteAsync($"{row.SourceSegment}\n");
+                        await targetKeyTermsTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    }
+                    else
+                    {
+                        await sourceTrainWriter.WriteAsync($"{row.SourceSegment}\n");
+                        await targetTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    }
                 }
                 if (row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)
                     trainCount++;
diff --git a/src/Machine/src/Serval.Machine.Shared/Services/WordAlignmentPreprocessBuildJob.cs b/src/Machine/src/Serval.Machine.Shared/Services/WordAlignmentPreprocessBuildJob.cs
@@ -35,6 +35,11 @@ CancellationToken cancellationToken
         await using StreamWriter targetTrainWriter =
             new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.trg.txt", cancellationToken));
 
+        await using StreamWriter sourceKeyTermsTrainWriter =
+            new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.key-terms.src.txt", cancellationToken));
+        await using StreamWriter targetKeyTermsTrainWriter =
+            new(await SharedFileService.OpenWriteAsync($"builds/{buildId}/train.key-terms.trg.txt", cancellationToken));
+
         await using Stream wordAlignmentStream = await SharedFileService.OpenWriteAsync(
             $"builds/{buildId}/word_alignments.inputs.json",
             cancellationToken
@@ -46,12 +51,21 @@ CancellationToken cancellationToken
         wordAlignmentWriter.WriteStartArray();
         await ParallelCorpusPreprocessingService.PreprocessAsync(
             corpora,
-            async row =>
+            async (row, trainingDataType) =>
             {
                 if (row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)
                 {
-                    await sourceTrainWriter.WriteAsync($"{row.SourceSegment}\n");
-                    await targetTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    if (trainingDataType == TrainingDataType.KeyTerms)
+                    {
+                        await sourceKeyTermsTrainWriter.WriteAsync($"{row.SourceSegment}\n");
+                        await targetKeyTermsTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    }
+                    else
+                    {
+                        await sourceTrainWriter.WriteAsync($"{row.SourceSegment}\n");
+                        await targetTrainWriter.WriteAsync($"{row.TargetSegment}\n");
+                    }
+
                     trainCount++;
                 }
             },
diff --git a/src/Machine/test/Serval.Machine.Shared.Tests/Services/PreprocessBuildJobTests.cs b/src/Machine/test/Serval.Machine.Shared.Tests/Services/PreprocessBuildJobTests.cs
@@ -978,6 +978,10 @@ public async Task<string> GetTargetExtractAsync()
         {
             using StreamReader srcReader = new(await SharedFileService.OpenReadAsync("builds/build1/train.src.txt"));
             using StreamReader trgReader = new(await SharedFileService.OpenReadAsync("builds/build1/train.trg.txt"));
+            using StreamReader srcTermReader =
+                new(await SharedFileService.OpenReadAsync("builds/build1/train.key-terms.src.txt"));
+            using StreamReader trgTermReader =
+                new(await SharedFileService.OpenReadAsync("builds/build1/train.key-terms.trg.txt"));
             int src1Count = 0;
             int src2Count = 0;
             int trgCount = 0;
@@ -998,8 +1002,17 @@ public async Task<string> GetTargetExtractAsync()
                 else if (srcLine.Length == 0)
                     trgCount++;
                 else
-                    termCount++;
+                    throw new ArgumentException("Unexpected line in test output");
             }
+
+            while (
+                (srcLine = await srcTermReader.ReadLineAsync()) is not null
+                && (trgLine = await trgTermReader.ReadLineAsync()) is not null
+            )
+            {
+                termCount++;
+            }
+
             return (src1Count, src2Count, trgCount, termCount);
         }
 
diff --git a/src/ServiceToolkit/src/SIL.ServiceToolkit/Models/TrainingDataType.cs b/src/ServiceToolkit/src/SIL.ServiceToolkit/Models/TrainingDataType.cs
@@ -0,0 +1,7 @@
+namespace SIL.ServiceToolkit.Models;
+
+public enum TrainingDataType
+{
+    Text = 0,
+    KeyTerms = 1,
+}
diff --git a/src/ServiceToolkit/src/SIL.ServiceToolkit/Services/IParallelCorpusPreprocessingService.cs b/src/ServiceToolkit/src/SIL.ServiceToolkit/Services/IParallelCorpusPreprocessingService.cs
@@ -9,7 +9,7 @@ ParallelCorpus corpus
 
     Task PreprocessAsync(
         IReadOnlyList<ParallelCorpus> corpora,
-        Func<Row, Task> train,
+        Func<Row, TrainingDataType, Task> train,
         Func<Row, bool, ParallelCorpus, Task> inference,
         bool useKeyTerms = false,
         HashSet<string>? ignoreUsfmMarkers = null
diff --git a/src/ServiceToolkit/src/SIL.ServiceToolkit/Services/ParallelCorpusPreprocessingService.cs b/src/ServiceToolkit/src/SIL.ServiceToolkit/Services/ParallelCorpusPreprocessingService.cs
@@ -68,7 +68,7 @@ MonolingualCorpus monolingualCorpus in parallelCorpus.SourceCorpora.Concat(paral
 
     public async Task PreprocessAsync(
         IReadOnlyList<ParallelCorpus> corpora,
-        Func<Row, Task> train,
+        Func<Row, TrainingDataType, Task> train,
         Func<Row, bool, ParallelCorpus, Task> inference,
         bool useKeyTerms = false,
         HashSet<string>? ignoreUsfmMarkers = null
@@ -128,7 +128,7 @@ public async Task PreprocessAsync(
 
             foreach (Row row in CollapseRanges(trainingRows))
             {
-                await train(row);
+                await train(row, TrainingDataType.Text);
                 if (!parallelTrainingDataPresent && row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)
                 {
                     parallelTrainingDataPresent = true;
@@ -177,7 +177,7 @@ ParallelTextRow row in parallelKeyTermsCorpus.DistinctBy(row =>
         {
             foreach (Row row in keyTermTrainingData)
             {
-                await train(row);
+                await train(row, TrainingDataType.KeyTerms);
             }
         }
     }
diff --git a/src/ServiceToolkit/test/SIL.ServiceToolkit.Tests/Services/ParallelCorpusProcessingServiceTests.cs b/src/ServiceToolkit/test/SIL.ServiceToolkit.Tests/Services/ParallelCorpusProcessingServiceTests.cs
@@ -47,7 +47,7 @@ public async Task TestParallelCorpusPreprocessor_FileFormatText()
         int inferenceCount = 0;
         await env.Processor.PreprocessAsync(
             corpora,
-            row =>
+            (row, _) =>
             {
                 if (row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)
                     trainCount++;
@@ -83,7 +83,7 @@ public async Task TestParallelCorpusPreprocessor_FileFormatParatext()
         var inferenceRefs = new List<string>();
         await env.Processor.PreprocessAsync(
             corpora,
-            row =>
+            (row, _) =>
             {
                 if (row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)
                 {

Original file line number	Diff line number	Diff line change
`@@ -124,7 +124,7 @@ await client.BuildStartedAsync(`
`124`	`124`	`List<InsertPretranslationsRequest> pretranslationsRequests = [];`
`125`	`125`	`await _parallelCorpusPreprocessingService.PreprocessAsync(`
`126`	`126`	`request.Corpora.Select(Map).ToList(),`
`127`		`- row => Task.CompletedTask,`
	`127`	`+ (_, _) => Task.CompletedTask,`
`128`	`128`	`(row, _, corpus) =>`
`129`	`129`	`{`
`130`	`130`	`string[] tokens = row.SourceSegment.Split();`
Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ await client.BuildStartedAsync(`
`79`	`79`	`List<InsertWordAlignmentsRequest> wordAlignmentsRequests = [];`
`80`	`80`	`await _parallelCorpusPreprocessingService.PreprocessAsync(`
`81`	`81`	`request.Corpora.Select(Map).ToList(),`
`82`		`- row => Task.CompletedTask,`
	`82`	`+ (_, _) => Task.CompletedTask,`
`83`	`83`	`(row, _, corpus) =>`
`84`	`84`	`{`
`85`	`85`	`wordAlignmentsRequests.Add(`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,7 @@ MonolingualCorpus monolingualCorpus in parallelCorpus.SourceCorpora.Concat(paral`
`68`	`68`
`69`	`69`	`public async Task PreprocessAsync(`
`70`	`70`	`IReadOnlyList<ParallelCorpus> corpora,`
`71`		`- Func<Row, Task> train,`
	`71`	`+ Func<Row, TrainingDataType, Task> train,`
`72`	`72`	`Func<Row, bool, ParallelCorpus, Task> inference,`
`73`	`73`	`bool useKeyTerms = false,`
`74`	`74`	`HashSet<string>? ignoreUsfmMarkers = null`
`@@ -128,7 +128,7 @@ public async Task PreprocessAsync(`
`128`	`128`
`129`	`129`	`foreach (Row row in CollapseRanges(trainingRows))`
`130`	`130`	`{`
`131`		`- await train(row);`
	`131`	`+ await train(row, TrainingDataType.Text);`
`132`	`132`	`if (!parallelTrainingDataPresent && row.SourceSegment.Length > 0 && row.TargetSegment.Length > 0)`
`133`	`133`	`{`
`134`	`134`	`parallelTrainingDataPresent = true;`
`@@ -177,7 +177,7 @@ ParallelTextRow row in parallelKeyTermsCorpus.DistinctBy(row =>`
`177`	`177`	`{`
`178`	`178`	`foreach (Row row in keyTermTrainingData)`
`179`	`179`	`{`
`180`		`- await train(row);`
	`180`	`+ await train(row, TrainingDataType.KeyTerms);`
`181`	`181`	`}`
`182`	`182`	`}`
`183`	`183`	`}`