Implementation:Neuml Txtai Application Init

Overview

The Application class is the central orchestrator for txtai's YAML-configured applications. It reads a declarative configuration and constructs all components -- embeddings indexes, pipelines, workflows, and agents -- wiring them together with proper dependency resolution. This page documents the Application.__init__() constructor and the Application.read() static method.

API Signature

Application.__init__

from txtai import Application

app = Application(config, loaddata=True)

Parameter	Type	Default	Description
config	str, dict	required	YAML file path, YAML string, or configuration dictionary
loaddata	bool	True	If True, load existing index data when available. If False, only load models.

Returns: An Application instance with all configured components initialized.

Application.read

from txtai import Application

config = Application.read(data)

Parameter	Type	Default	Description
data	str, dict	required	YAML file path, YAML string, or dictionary

Returns: A Python dict containing the parsed configuration.

Source Reference

File: src/python/txtai/app/base.py (Lines 19-82)

Application.read Implementation

@staticmethod
def read(data):
    if isinstance(data, str):
        if os.path.exists(data):
            # Read yaml from file
            with open(data, "r", encoding="utf-8") as f:
                return yaml.safe_load(f)

        # Attempt to read yaml from input
        data = yaml.safe_load(data)
        if not isinstance(data, str):
            return data

        # File not found and input is not yaml, raise error
        raise FileNotFoundError(f"Unable to load file '{data}'")

    # Return unmodified
    return data

Key behavior:

• If data is a string and the file exists on disk, it reads and parses the YAML file
• If the file does not exist, it attempts to parse the string as inline YAML
• If the string is not valid YAML (parses back to a plain string), a FileNotFoundError is raised
• If data is already a dictionary, it is returned unmodified

Application.__init__ Implementation

def __init__(self, config, loaddata=True):
    # Initialize member variables
    self.config, self.documents, self.embeddings = Application.read(config), None, None

    # Write lock - allows only a single thread to update embeddings
    self.lock = RLock()

    # ThreadPool - runs scheduled workflows
    self.pool = None

    # Create pipelines
    self.createpipelines()

    # Create workflows
    self.createworkflows()

    # Create agents
    self.createagents()

    # Create embeddings index
    self.indexes(loaddata)

Initialization sequence:

1. Application.read(config) parses the YAML configuration
2. A reentrant lock (RLock) is created for thread-safe embeddings updates
3. createpipelines() instantiates all configured NLP pipelines
4. createworkflows() builds workflow chains referencing the created pipelines
5. createagents() configures LLM-driven agents with tool access
6. indexes(loaddata) initializes the embeddings index, loading existing data if available

Instance Attributes

After construction, the Application instance exposes:

Attribute	Type	Description
self.config	dict	Parsed YAML configuration
self.embeddings	Embeddings or None	Embeddings index instance
self.documents	Documents or None	Document batch buffer (set during add operations)
self.pipelines	dict	Map of pipeline name to pipeline instance
self.workflows	dict	Map of workflow name to workflow instance
self.agents	dict	Map of agent name to agent instance
self.lock	RLock	Reentrant lock for thread-safe index updates
self.pool	ThreadPool or None	Thread pool for scheduled workflow execution

Usage Examples

Loading from a YAML File

from txtai import Application

# Load application from YAML file
app = Application("config.yml")

# Search the configured embeddings index
results = app.search("machine learning", limit=5)

Loading from a Dictionary

from txtai import Application

config = {
    "path": "/tmp/index",
    "writable": True,
    "embeddings": {
        "path": "sentence-transformers/all-MiniLM-L6-v2",
        "content": True
    }
}

app = Application(config)
app.add([{"id": "0", "text": "txtai is an embeddings database"}])
app.index()

Loading Models Only (No Data)

from txtai import Application

# Load models but skip loading existing index data
app = Application("config.yml", loaddata=False)

This is used in Docker builds to pre-cache models in the container image without requiring existing index data.

Error Handling

Error	Condition
FileNotFoundError	String input is neither a valid file path nor valid YAML
yaml.YAMLError	Malformed YAML content
ReadOnlyError	Attempting write operations (add, index, delete) when writable is not True

Pipeline Dependency Resolution

The createpipelines() method sorts pipelines to ensure dependent ones are created after their dependencies:

dependent = ["similarity", "extractor", "rag", "reranker"]
pipelines = sorted(pipelines, key=lambda x: dependent.index(x) + 1 if x in dependent else 0)

This ensures that when an extractor pipeline references a similarity pipeline, the similarity pipeline already exists.

See Also

• Neuml_Txtai_YAML_Application_Configuration - Principle behind declarative YAML configuration
• Neuml_Txtai_API_Create - How the API server uses Application during startup
• Neuml_Txtai_API_Server_Bootstrap - FastAPI lifecycle management and route registration

Requires Environment

• Environment:Neuml_Txtai_API_Server_Configuration
• Environment:Neuml_Txtai_Optional_Extras_Dependencies