Data Models

This section documents the data models and structures used throughout the EVE Pipeline.

Document Model

The primary data structure representing documents in the pipeline.

Unified document object that encapsulates content and metadata throughout the pipeline.

This replaces the need to pass (Path, str) tuples and provides a consistent interface for document handling across all pipeline stages.

Attributes:

Name	Type	Description
content	str	The actual document text content
file_path	Path	Path to the source file
file_format	str	Format of the source file (pdf, md, html, etc.)
metadata	Dict[str, Any]	Original metadata from the document (preserved from source)
embedding	Optional[List[float]]	Optional embedding vector for the document
pipeline_metadata	Dict[str, Any]	Metadata added by pipeline steps (filters, processing, etc.)

Source code in eve/model/document.py

@dataclass
class Document:
    """
    Unified document object that encapsulates content and metadata throughout the pipeline.

    This replaces the need to pass (Path, str) tuples and provides a consistent
    interface for document handling across all pipeline stages.

    Attributes:
        content: The actual document text content
        file_path: Path to the source file
        file_format: Format of the source file (pdf, md, html, etc.)
        metadata: Original metadata from the document (preserved from source)
        embedding: Optional embedding vector for the document
        pipeline_metadata: Metadata added by pipeline steps (filters, processing, etc.)
    """

    content: str
    file_path: Path
    file_format: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None
    pipeline_metadata: Dict[str, Any] = field(default_factory=dict)

    def __dict__(self) -> Dict[str, Any]:
        result = {
            "content": self.content,
            "file_path": str(self.file_path),
            "file_format": self.file_format,
            "metadata": self.metadata.copy(),
            "pipeline_metadata": self.pipeline_metadata.copy(),
        }
        if self.embedding is not None:
            result["embedding"] = self.embedding
        return result

    def __hash__(self):
        return hash(self.file_path)

    def __eq__(self, other):
        return isinstance(other, Document) and self.file_path == other.file_path

    @property
    def filename(self) -> str:
        """Get the filename without path."""
        return self.file_path.name

    @property
    def extension(self) -> str:
        """Get the file extension."""
        return self.file_path.suffix.lstrip(".")

    @property
    def content_length(self) -> int:
        """Get the length of the content."""
        return len(self.content)

    def is_empty(self) -> bool:
        """Check if the document content is empty."""
        return not self.content.strip()

    def add_metadata(self, key: str, value: Any) -> None:
        """Add an entry to the original metadata."""
        self.metadata[key] = value

    def get_metadata(self, key: str, default: Any = None) -> Any:
        """Get a value from the original metadata with optional default."""
        return self.metadata.get(key, default)

    def add_pipeline_metadata(self, key: str, value: Any) -> None:
        """Add an entry to pipeline metadata (for tracking pipeline processing)."""
        self.pipeline_metadata[key] = value

    def get_pipeline_metadata(self, key: str, default: Any = None) -> Any:
        """Get a value from pipeline metadata with optional default."""
        return self.pipeline_metadata.get(key, default)

    def update_content(self, new_content: str) -> None:
        """Update the document content and track the change in metadata."""
        old_length = self.content_length
        self.content = new_content
        new_length = self.content_length

        # Track content changes in metadata
        changes = self.metadata.get("content_changes", [])
        changes.append(
            {
                "old_length": old_length,
                "new_length": new_length,
                "size_change": new_length - old_length,
            }
        )
        self.metadata["content_changes"] = changes

    @classmethod
    def from_path_and_content(
        cls, file_path: Path, content: str, **metadata
    ) -> "Document":
        """Create a Document from a file path and content string."""
        return cls(content=content, file_path=file_path, metadata=metadata)

    @classmethod
    def from_tuple(cls, path_content_tuple: tuple[Path, str], **metadata) -> "Document":
        """Create a Document from a (Path, str) tuple for backwards compatibility."""
        file_path, content = path_content_tuple
        return cls.from_path_and_content(file_path, content, **metadata)

    def to_tuple(self) -> tuple[Path, str]:
        """Convert to (Path, str) tuple for backwards compatibility."""
        return (self.file_path, self.content)

    def __str__(self) -> str:
        """String representation showing filename and content length."""
        return f"Document({self.filename}, {self.file_format} format)"

    def __repr__(self) -> str:
        """Detailed representation."""
        return f"Document(file_path={self.file_path}, format={self.file_format}, metadata_keys={list(self.metadata.keys())})"

content_length property

Get the length of the content.

extension property

Get the file extension.

filename property

Get the filename without path.

__repr__()

Detailed representation.

Source code in eve/model/document.py

def __repr__(self) -> str:
    """Detailed representation."""
    return f"Document(file_path={self.file_path}, format={self.file_format}, metadata_keys={list(self.metadata.keys())})"

__str__()

String representation showing filename and content length.

Source code in eve/model/document.py

def __str__(self) -> str:
    """String representation showing filename and content length."""
    return f"Document({self.filename}, {self.file_format} format)"

add_metadata(key, value)

Add an entry to the original metadata.

Source code in eve/model/document.py

def add_metadata(self, key: str, value: Any) -> None:
    """Add an entry to the original metadata."""
    self.metadata[key] = value

add_pipeline_metadata(key, value)

Add an entry to pipeline metadata (for tracking pipeline processing).

Source code in eve/model/document.py

def add_pipeline_metadata(self, key: str, value: Any) -> None:
    """Add an entry to pipeline metadata (for tracking pipeline processing)."""
    self.pipeline_metadata[key] = value

from_path_and_content(file_path, content, **metadata) classmethod

Create a Document from a file path and content string.

Source code in eve/model/document.py

@classmethod
def from_path_and_content(
    cls, file_path: Path, content: str, **metadata
) -> "Document":
    """Create a Document from a file path and content string."""
    return cls(content=content, file_path=file_path, metadata=metadata)

from_tuple(path_content_tuple, **metadata) classmethod

Create a Document from a (Path, str) tuple for backwards compatibility.

Source code in eve/model/document.py

@classmethod
def from_tuple(cls, path_content_tuple: tuple[Path, str], **metadata) -> "Document":
    """Create a Document from a (Path, str) tuple for backwards compatibility."""
    file_path, content = path_content_tuple
    return cls.from_path_and_content(file_path, content, **metadata)

get_metadata(key, default=None)

Get a value from the original metadata with optional default.

Source code in eve/model/document.py

def get_metadata(self, key: str, default: Any = None) -> Any:
    """Get a value from the original metadata with optional default."""
    return self.metadata.get(key, default)

get_pipeline_metadata(key, default=None)

Get a value from pipeline metadata with optional default.

Source code in eve/model/document.py

def get_pipeline_metadata(self, key: str, default: Any = None) -> Any:
    """Get a value from pipeline metadata with optional default."""
    return self.pipeline_metadata.get(key, default)

is_empty()

Check if the document content is empty.

Source code in eve/model/document.py

def is_empty(self) -> bool:
    """Check if the document content is empty."""
    return not self.content.strip()

to_tuple()

Convert to (Path, str) tuple for backwards compatibility.

Source code in eve/model/document.py

def to_tuple(self) -> tuple[Path, str]:
    """Convert to (Path, str) tuple for backwards compatibility."""
    return (self.file_path, self.content)

update_content(new_content)

Update the document content and track the change in metadata.

Source code in eve/model/document.py

def update_content(self, new_content: str) -> None:
    """Update the document content and track the change in metadata."""
    old_length = self.content_length
    self.content = new_content
    new_length = self.content_length

    # Track content changes in metadata
    changes = self.metadata.get("content_changes", [])
    changes.append(
        {
            "old_length": old_length,
            "new_length": new_length,
            "size_change": new_length - old_length,
        }
    )
    self.metadata["content_changes"] = changes

Configuration Models

Data models for pipeline configuration.

Inputs Configuration

Bases: BaseModel

Source code in eve/config.py

class Inputs(BaseModel):
    mode: str = "file"  # file | directory
    path: Union[str, list[str]]

    def get_files(self) -> list[Path]:
        paths = [self.path] if isinstance(self.path, str) else self.path
        files = []

        for p in paths:
            p = Path(p)

            if p.is_file():
                files.append(p)
            elif p.is_dir():
                files.extend([f for f in p.rglob("*") if f.is_file()]) # recursive search across multiple levels
        return files

Pipeline Configuration

Bases: BaseModel

Source code in eve/config.py

class PipelineConfig(BaseModel):
    batch_size: int = 20
    inputs: Inputs
    stages: list[dict[str, Any]]  # list of dict since we have stage name + stage configs

    @validator("stages")
    def check_stages(cls, v):
        allowed = {"ingestion", "cleaning", "export", "duplication", "extraction", "pii", "metadata", "chunker", "export_jsonl", "perplexity", "pii_filter", "length_filter", "newline_filter", "reference_filter", "qdrant_upload"}
        for stage in v:
            if stage["name"] not in allowed:
                raise ValueError(f"Unsupported stage: {stage['name']}. Allowed: {allowed}")
        return v