Pular para conteúdo

Pipeline

Motor lógico de processamento para extrair entidades e normalizar frases em PT-BR.

Orquestrador de pré-processamento.

Encadeia componentes (EntityNormalizer -> TextCleaner -> Tokenizer -> Stopwords -> Lemmatizer) passando um PipelineContext entre cada estágio.

Source code in src/text_similarity/pipeline/pipeline.py 
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
class PreprocessingPipeline:
    """Orquestrador de pré-processamento.

    Encadeia componentes (EntityNormalizer -> TextCleaner ->
    Tokenizer -> Stopwords -> Lemmatizer) passando um `PipelineContext`
    entre cada estágio.
    """

    def __init__(self, stages: List[PipelineStage] | None = None) -> None:
        """Inicializa a pipeline opcionalmente com estágios pré-carregados."""
        self.stages = stages or []

    def add_stage(self, stage: PipelineStage) -> None:
        """Anexa um novo estágio ao fim da fila de execução da pipeline."""
        self.stages.append(stage)

    def process(self, text: str) -> tuple[str, PipelineContext]:
        """Executa os estágios em sequência sobre o texto fornecido.

        Cria um `PipelineContext` inicial e o passa por cada estágio,
        acumulando transformações e metadados ao longo do caminho.

        Args:
            text: Texto bruto de entrada.

        Returns:
            Tuple: (texto_processado, contexto_completo_da_execução)
        """
        ctx = PipelineContext(text=text)

        for stage in self.stages:
            try:
                ctx = stage.process(ctx)
            except StageProcessingError:
                # Já é exceção de pipeline, propague
                raise
            except (TypeError, ValueError) as e:
                # Erro de dados/input inválido
                raise StageProcessingError(stage.__class__.__name__, e) from e
            except (UnicodeDecodeError, UnicodeEncodeError) as e:
                # Erro de encoding
                raise StageProcessingError(stage.__class__.__name__, e) from e
            except OSError as e:
                # Erro de I/O ou modelo não encontrado
                raise StageProcessingError(stage.__class__.__name__, e) from e
            except RuntimeError as e:
                # Erro de runtime de backend NLP (ex: Regex, SpaCy)
                raise StageProcessingError(stage.__class__.__name__, e) from e

        return ctx.text, ctx

__init__(stages=None)

Inicializa a pipeline opcionalmente com estágios pré-carregados.

Source code in src/text_similarity/pipeline/pipeline.py 
23
24
25
def __init__(self, stages: List[PipelineStage] | None = None) -> None:
    """Inicializa a pipeline opcionalmente com estágios pré-carregados."""
    self.stages = stages or []

add_stage(stage)

Anexa um novo estágio ao fim da fila de execução da pipeline.

Source code in src/text_similarity/pipeline/pipeline.py 
27
28
29
def add_stage(self, stage: PipelineStage) -> None:
    """Anexa um novo estágio ao fim da fila de execução da pipeline."""
    self.stages.append(stage)

process(text)

Executa os estágios em sequência sobre o texto fornecido.

Cria um PipelineContext inicial e o passa por cada estágio, acumulando transformações e metadados ao longo do caminho.

Parameters:

Name Type Description Default
text str

Texto bruto de entrada.

 required

Returns:

Name Type Description
Tuple tuple[str, PipelineContext]

(texto_processado, contexto_completo_da_execução)
Source code in src/text_similarity/pipeline/pipeline.py 
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
def process(self, text: str) -> tuple[str, PipelineContext]:
    """Executa os estágios em sequência sobre o texto fornecido.

    Cria um `PipelineContext` inicial e o passa por cada estágio,
    acumulando transformações e metadados ao longo do caminho.

    Args:
        text: Texto bruto de entrada.

    Returns:
        Tuple: (texto_processado, contexto_completo_da_execução)
    """
    ctx = PipelineContext(text=text)

    for stage in self.stages:
        try:
            ctx = stage.process(ctx)
        except StageProcessingError:
            # Já é exceção de pipeline, propague
            raise
        except (TypeError, ValueError) as e:
            # Erro de dados/input inválido
            raise StageProcessingError(stage.__class__.__name__, e) from e
        except (UnicodeDecodeError, UnicodeEncodeError) as e:
            # Erro de encoding
            raise StageProcessingError(stage.__class__.__name__, e) from e
        except OSError as e:
            # Erro de I/O ou modelo não encontrado
            raise StageProcessingError(stage.__class__.__name__, e) from e
        except RuntimeError as e:
            # Erro de runtime de backend NLP (ex: Regex, SpaCy)
            raise StageProcessingError(stage.__class__.__name__, e) from e

    return ctx.text, ctx