Forward backend options during streaming speech generation (#1389) (#1390)

Switch SQLite ANN connections when load() is given a new path

docker/base/Dockerfile

@@ -27,7 +27,7 @@ RUN \
     # Install txtai project and dependencies
     ln -s /usr/bin/python${PYTHON_VERSION} /usr/bin/python && \
     python -m pip install --no-cache-dir -U pip wheel setuptools && \
-    if [ -z ${GPU} ]; then pip install --no-cache-dir torch==2.12.1+cpu torchvision==0.27.1+cpu -f https://download.pytorch.org/whl/torch -f https://download.pytorch.org/whl/torchvision; fi && \
+    if [ -z ${GPU} ]; then pip install --no-cache-dir torch==2.14.1+cpu torchvision==0.29.1+cpu -f https://download.pytorch.org/whl/torch -f https://download.pytorch.org/whl/torchvision; fi && \
     python -m pip install --no-cache-dir txtai${COMPONENTS} && \
     python -c "import sys, importlib.util as util; 1 if util.find_spec('nltk') else sys.exit(); import nltk; nltk.download(['punkt', 'punkt_tab', 'averaged_perceptron_tagger_eng'])" && \
     \

setup.py

@@ -181,7 +181,7 @@ extras["all"] = (
 
 setup(
     name=package,
-    version="9.14.0",
+    version="9.15.0",
     author="NeuML",
     description="All-in-one open-source AI framework for semantic search, LLM orchestration and language model workflows",
     long_description=DESCRIPTION,

src/python/txtai/agent/tool/bash.py

@@ -32,7 +32,7 @@ class BashTool(Tool):
         self.output_type = "any"
 
         # Default list of allowed commands
-        self.allowed = allowed if allowed else ["cat", "cut", "diff", "grep", "head", "ls", "tail"]
+        self.allowed = allowed if allowed is not None else ["cat", "cut", "diff", "grep", "head", "ls", "tail"]
 
         # Maximum time to wait for a command, None waits indefinitely
         self.timeout = timeout

src/python/txtai/ann/dense/sqlite.py

@@ -35,6 +35,8 @@ class SQLite(ANN):
         self.quantize = 8 if isinstance(self.quantize, bool) and self.quantize else int(self.quantize) if self.quantize else None
 
     def load(self, path):
+        if self.path != path:
+            self.close()
         self.path = path
 
     def index(self, embeddings):

src/python/txtai/app/base.py

@@ -258,7 +258,7 @@ class Application:
         """
 
         # Check for task shorthand syntax
-        task = {"action": task} if isinstance(task, (str, list)) else task
+        task = {"action": task} if isinstance(task, (str, list)) else task.copy()
 
         if "action" in task:
             action = task["action"]

src/python/txtai/database/duckdb.py

@@ -18,6 +18,7 @@ except ImportError:
 
 from .embedded import Embedded
 from .schema import Statement
+from .sql import SQLError
 
 
 class DuckDB(Embedded):
@@ -29,6 +30,11 @@ class DuckDB(Embedded):
     DELETE_DOCUMENT = "DELETE FROM documents WHERE id = ?"
     DELETE_OBJECT = "DELETE FROM objects WHERE id = ?"
 
+    # Added to errors that abort the transaction
+    ABORTED = (
+        "This error aborted the DuckDB transaction. Unsaved changes are lost and later operations fail until the index is loaded or built again."
+    )
+
     def __init__(self, config):
         super().__init__(config)
 
@@ -36,8 +42,15 @@ class DuckDB(Embedded):
             raise ImportError('DuckDB is not available - install "database" extra to enable')
 
     def execute(self, function, *args):
-        # Call parent method with DuckDB compatible arguments
-        return super().execute(function, *self.formatargs(args))
+        try:
+            # Call parent method with DuckDB compatible arguments
+            return super().execute(function, *self.formatargs(args))
+        except SQLError as e:
+            # Runtime errors abort the transaction. DuckDB has no savepoints, so it can't be recovered without a rollback.
+            if self.aborted():
+                raise SQLError(f"{e}\n\n{DuckDB.ABORTED}") from None
+
+            raise
 
     def insertdocument(self, uid, data, tags, entry):
         # Delete existing document
@@ -123,6 +136,20 @@ class DuckDB(Embedded):
 
         return connection
 
+    def aborted(self):
+        """
+        Checks if the current transaction is aborted.
+
+        Returns:
+            True if the current transaction is aborted
+        """
+
+        try:
+            self.connection.execute("SELECT 1")
+            return False
+        except duckdb.TransactionException:
+            return True
+
     def formatargs(self, args):
         """
         DuckDB doesn't support named parameters. This method replaces named parameters with question marks

src/python/txtai/database/sql/aggregate.py

@@ -63,14 +63,15 @@ class Aggregate(SQL):
 
         aggregates = {}
         for column in columns:
-            column = column.lower()
-            if column.startswith(("count(", "sum(", "total(")):
+            # Match function names case-insensitively but keep the column name as returned
+            name = column.lower()
+            if name.startswith(("count(", "sum(", "total(")):
                 aggregates[column] = sum
-            elif column.startswith("max("):
+            elif name.startswith("max("):
                 aggregates[column] = max
-            elif column.startswith("min("):
+            elif name.startswith("min("):
                 aggregates[column] = min
-            elif column.startswith("avg("):
+            elif name.startswith("avg("):
                 aggregates[column] = self.avg
 
         return aggregates
@@ -182,9 +183,9 @@ class Aggregate(SQL):
                 clause = clause.rsplit(" ")[0]
                 reverse = True
 
-            # Order by columns must be in select clause
+            # Order by columns must be in select clause. NULLs sort first like in SQLite (last when descending).
             if clause in query["select"]:
-                results = sorted(results, key=operator.itemgetter(clause), reverse=reverse)
+                results = sorted(results, key=lambda row, column=clause: (row[column] is not None, row[column]), reverse=reverse)
 
         return results
 

src/python/txtai/database/sql/base.py

@@ -5,6 +5,7 @@ SQL module
 from io import StringIO
 from shlex import shlex
 
+from .error import SQLError
 from .expression import Expression
 
 
@@ -42,8 +43,11 @@ class SQL:
 
         clauses = None
         if self.issql(query):
-            # Ignore multiple statements
-            query = query.split(";")[0]
+            # Multiple statements are not supported
+            components = [x for x in query.split(";") if x.strip()]
+            if len(components) > 1:
+                raise SQLError("Multiple SQL statements are not supported. Pass values that contain a semicolon as a bind parameter.")
+            query = components[0]
 
             # Tokenize query
             tokens, positions = self.tokenize(query)

src/python/txtai/embeddings/search/base.py

@@ -92,7 +92,7 @@ class Search:
         Args:
             queries: list of queries
             limit: maximum results
-            weights: hybrid score weights
+            weights: hybrid score weights, either one set of weights for all queries or a list of [dense, sparse] weights per query
             index: index name
 
         Returns:
@@ -114,9 +114,13 @@ class Search:
             if isinstance(weights, (int, float)):
                 weights = [weights, 1 - weights]
 
+            # Use the same weights for each query, unless a list of weights per query is passed
+            if not isinstance(weights[0], (list, tuple)):
+                weights = [weights] * len(queries)
+
             # Create weighted scores via hybrid fusion strategy
             fusion = Hybrid(self.scoring)
-            return [fusion(vectors, weights, limit) for vectors in zip(dense, sparse)]
+            return [fusion(vectors, weights[x], limit) for x, vectors in enumerate(zip(dense, sparse))]
 
         # Raise an error if when no indexes are available
         if not sparse and not dense:

src/python/txtai/embeddings/search/explain.py

@@ -86,8 +86,9 @@ class Explain:
             # Parse query
             query = self.database.parse(query)
 
-            # Extract query from similar clause
-            query = " ".join([" ".join(clause) for clause in query["similar"]]) if "similar" in query else None
+            # Extract query text from similar clauses. Only the first argument is query text, the rest are
+            # candidates, index and weights.
+            query = " ".join([clause[0] for clause in query["similar"]]) if "similar" in query else None
 
         # Return original texts if query, text or score not present
         if not query or not texts or "score" not in texts[0] or "text" not in texts[0]:

src/python/txtai/embeddings/search/hybrid.py

@@ -194,7 +194,8 @@ class LogOdds:
         array = [s for s in raw if s > 0]
         if array:
             median = sorted(array)[len(array) // 2]
-            std = (sum((x - sum(array) / len(array)) ** 2 for x in array) / len(array)) ** 0.5
+            mean = sum(array) / len(array)
+            std = (sum((x - mean) ** 2 for x in array) / len(array)) ** 0.5
             alpha = 1.0 / std if std > 0 else 1.0
 
         return median, alpha

src/python/txtai/embeddings/search/scan.py

@@ -61,9 +61,9 @@ class Scan:
 
             candidates = max(candidates) if candidates else default
 
-            # Query weights to pass to batch search
-            weights = [query.weights for query in iqueries if query.weights is not None]
-            weights = max(weights) if weights else self.weights
+            # Query weights to pass to batch search. Each query keeps its own weights, queries without weights use the default.
+            weights = [query.weights if query.weights is not None else self.weights for query in iqueries]
+            weights = [[weight, 1 - weight] if isinstance(weight, (int, float)) else weight for weight in weights]
 
             # Index to run query against
             index = index if index else self.index

src/python/txtai/embeddings/search/terms.py

@@ -37,9 +37,10 @@ class Terms:
                 # Parse query
                 parse = self.database.parse(query)
 
-                # Join terms from similar clauses (absent when the query has no similar() clause)
+                # Join query text from similar clauses (absent when the query has no similar() clause).
+                # Only the first argument is query text, the rest are candidates, index and weights.
                 if "similar" in parse:
-                    terms.append(" ".join(" ".join(s) for s in parse["similar"]))
+                    terms.append(" ".join(s[0] for s in parse["similar"]))
 
             return terms
 

src/python/txtai/graph/base.py

@@ -760,7 +760,7 @@ class Graph:
             ids = ids.keys() if ids else None
 
             # Infer topic
-            topic = Counter(self.attribute(x, "topic") for x in ids).most_common(1)[0][0] if ids else None
+            topic = self.commonattribute(ids, "topic")
             if topic:
                 # Add id to topic list and set topic attribute
                 self.topics[topic].append(node)
@@ -770,5 +770,21 @@ class Graph:
                 self.addattribute(node, "topicrank", len(self.topics[topic]) - 1)
 
                 # Infer category
-                category = Counter(self.attribute(x, "category") for x in ids).most_common(1)[0][0]
-                self.addattribute(node, "category", category)
+                category = self.commonattribute([x for x in ids if self.attribute(x, "topic") == topic], "category")
+                if category:
+                    self.addattribute(node, "category", category)
+
+    def commonattribute(self, ids, attribute):
+        """
+        Gets the most common non-empty attribute value across a list of nodes.
+
+        Args:
+            ids: list of node ids
+            attribute: attribute name
+
+        Returns:
+            most common value or None if no node has the attribute
+        """
+
+        values = Counter(value for value in (self.attribute(x, attribute) for x in ids or []) if value is not None)
+        return values.most_common(1)[0][0] if values else None

src/python/txtai/graph/rdbms.py

@@ -59,6 +59,10 @@ class RDBMS(NetworkX):
         else:
             yield from super().scan(attribute, data)
 
+    def addattribute(self, node, field, value):
+        if self.hasnode(node):
+            self.graph.backend.add_node(node, {field: value})
+
     def load(self, path):
         # Create graph instance
         self.graph, self.database = self.connect()

src/python/txtai/pipeline/audio/signal.py

@@ -169,11 +169,15 @@ class Signal:
             audio with silence removed
         """
 
-        # Process in 20ms chunks
+        # Empty audio has no chunks to concatenate.
+        if len(audio) == 0:
+            return audio
+
+        # Process in chunks, including the final partial chunk
         n, offset = int(rate * (20 / 1000.0) * 2), 0
 
         chunks = []
-        while offset + n <= len(audio):
+        while offset < len(audio):
             # Calculate energy for chunk and detection result
             chunk = audio[offset : offset + n]
             energyfreq = Signal.energy(chunk, rate)

src/python/txtai/pipeline/audio/texttospeech.py

@@ -92,7 +92,7 @@ class TextToSpeech(Pipeline):
 
         # Streaming response
         if stream:
-            return self.stream(texts, speaker, encoding)
+            return self.stream(texts, speaker, encoding, **kwargs)
 
         # Transform text to speech
         results = [self.execute(x, speaker, encoding, **kwargs) for x in texts]
@@ -137,7 +137,7 @@ class TextToSpeech(Pipeline):
 
         return exists
 
-    def stream(self, texts, speaker, encoding):
+    def stream(self, texts, speaker, encoding, **kwargs):
         """
         Iterates over texts, splits into segments and yields snippets of audio.
         This method is designed to integrate with streaming LLM generation.
@@ -146,6 +146,7 @@ class TextToSpeech(Pipeline):
             texts: list of input texts
             speaker: speaker id
             encoding: audio encoding format
+            kwargs: additional keyword args
 
         Returns:
             snippets of audio as NumPy arrays or audio bytes depending on encoding parameter
@@ -157,11 +158,11 @@ class TextToSpeech(Pipeline):
 
             if x == "\n" or (x.strip().endswith(".") and len([y for y in buffer if y]) > 2):
                 data, buffer = "".join(buffer), []
-                yield self.execute(data, speaker, encoding)
+                yield self.execute(data, speaker, encoding, **kwargs)
 
         if buffer:
             data = "".join(buffer)
-            yield self.execute(data, speaker, encoding)
+            yield self.execute(data, speaker, encoding, **kwargs)
 
     def execute(self, text, speaker, encoding, **kwargs):
         """

src/python/txtai/pipeline/data/htmltomd.py

@@ -60,6 +60,14 @@ class HTMLToMarkdown(Pipeline):
         for script in soup.find_all(["script", "style"]):
             script.decompose()
 
+        # Replace line breaks with newlines. Table cells and headings must stay on a single line, use a space there.
+        for br in soup.find_all("br"):
+            parent = br.parent
+            br.replace_with(" " if br.find_parent(["th", "td", "h1", "h2", "h3", "h4", "h5", "h6"]) else "\n")
+
+            # Merge the line break with the text around it
+            parent.smooth()
+
         # Check for article sections
         article = next((x for x in ["article", "main"] if soup.find(x)), None)
 
@@ -186,12 +194,17 @@ class HTMLToMarkdown(Pipeline):
             target, text = x if x.name else node, x.text
 
             if text.strip():
+                # Leading whitespace separates this text from the previous word, keep it outside of the formatting
+                space = " " if text[0].isspace() and texts and not texts[-1][-1:].isspace() else ""
+
                 if target.name in ("b", "strong"):
-                    text = f"**{text.strip()}** "
+                    text = f"{space}**{text.strip()}** "
                 elif target.name in ("i", "em"):
-                    text = f"*{text.strip()}* "
+                    text = f"{space}*{text.strip()}* "
                 elif target.name == "a":
-                    text = f"[{text.strip()}]({target.get('href')}) "
+                    # Links without a target are added as plain text
+                    href = target.get("href")
+                    text = f"{space}[{text.strip()}]({href}) " if href else f"{space}{text.strip()} "
 
             texts.append(text)
 
@@ -256,29 +269,34 @@ class HTMLToMarkdown(Pipeline):
         """
 
         elements, count = [], 0
+        if node.name == "ol":
+            try:
+                count = int(node.get("start", 1)) - 1
+            except ValueError:
+                pass
+
         for element in node.find_all(("li", "ul", "ol"), recursive=False):
             # Nested list that is a direct child of this list
             if element.name != "li":
                 self.nested(element, article, elements)
-                continue
+            else:
+                # Unordered lists use dashes. Ordered lists use numbers.
+                count += 1
+                prefix = "-" if node.name == "ul" else f"{count}."
 
-            # Unordered lists use dashes. Ordered lists use numbers.
-            count += 1
-            prefix = "-" if node.name == "ul" else f"{count}."
+                # Detach nested lists, these are formatted separately
+                nested = [child.extract() for child in element.find_all(("ul", "ol"), recursive=False)]
 
-            # Detach nested lists, these are formatted separately
-            nested = [child.extract() for child in element.find_all(("ul", "ol"), recursive=False)]
+                # List item text
+                text = self.process(element, article)
 
-            # List item text
-            text = self.process(element, article)
+                # Add list element
+                if text:
+                    elements.append(f"{prefix} {text}")
 
-            # Add list element
-            if text:
-                elements.append(f"{prefix} {text}")
-
-            # Add nested lists
-            for child in nested:
-                self.nested(child, article, elements)
+                # Add nested lists
+                for child in nested:
+                    self.nested(child, article, elements)
 
         # Join elements together as string
         return "\n".join(elements)

src/python/txtai/pipeline/data/tabular.py

@@ -106,7 +106,8 @@ class Tabular(Pipeline):
         # Tuple iteration preserves column types instead of coercing each row to one dtype.
         for index, values in zip(df.index, df.itertuples(index=False, name=None)):
             row = dict(zip(df.columns, values))
-            uid = row[self.idcolumn] if self.idcolumn is not None else index
+            # Coerce NaN ids to None so the row-index fallback applies.
+            uid = self.column(row[self.idcolumn]) if self.idcolumn is not None else index
             uid = uid if uid is not None else index
             text = self.concat(row, columns)
 

src/python/txtai/pipeline/text/reranker.py

@@ -63,20 +63,19 @@ class Reranker(Pipeline):
         # Re-rank using similarity pipeline
         ranked = []
         for x, result in enumerate(results):
-            texts = [row["text"] for row in result]
+            # Only score when a search has results
+            if result:
+                texts = [row["text"] for row in result]
 
-            if self.cache is not None and (namespace := self.modelkey()) and all("id" in row for row in result):
-                with self.lock:
-                    texts = self.vectors(result, namespace)
-                    if not result:
-                        ranked.append([])
-                        continue
+                if self.cache is not None and (namespace := self.modelkey()) and all("id" in row for row in result):
+                    with self.lock:
+                        texts = self.vectors(result, namespace)
+                        for uid, score in self.similarity(queries[x], texts):
+                            result[uid]["score"] = score
+                else:
+                    # Score results and merge
                     for uid, score in self.similarity(queries[x], texts):
                         result[uid]["score"] = score
-            else:
-                # Score results and merge
-                for uid, score in self.similarity(queries[x], texts):
-                    result[uid]["score"] = score
 
             # Sort and take top n sorted results
             ranked.append(sorted(result, key=lambda row: row["score"], reverse=True)[:limit])
@@ -118,8 +117,6 @@ class Reranker(Pipeline):
 
         with self.lock:
             self.reset(namespace)
-            if not rows:
-                return []
             keys, vectors, misses = [], [], []
             for row in rows:
                 key = (namespace, row["id"], hashlib.sha256(row["text"].encode("utf-8")).hexdigest())

src/python/txtai/version.py

@@ -3,4 +3,4 @@ Version strings
 """
 
 # Current version tag
-__version__ = "9.14.0"
+__version__ = "9.15.0"

src/python/txtai/workflow/factory.py

@@ -24,16 +24,20 @@ class WorkflowFactory:
             Workflow
         """
 
+        # Copy configuration before resolving tasks
+        config = config.copy()
+
         # Resolve workflow tasks
         tasks = []
         for tconfig in config["tasks"]:
+            tconfig = tconfig.copy()
             task = tconfig.pop("task") if "task" in tconfig else ""
             tasks.append(TaskFactory.create(tconfig, task))
 
         config["tasks"] = tasks
 
         if "stream" in config:
-            sconfig = config["stream"]
+            sconfig = config["stream"].copy()
             task = sconfig.pop("task") if "task" in sconfig else "stream"
 
             config["stream"] = TaskFactory.create(sconfig, task)

src/python/txtai/workflow/task/factory.py

@@ -47,6 +47,9 @@ class TaskFactory:
             Task
         """
 
+        # Copy configuration before binding arguments
+        config = config.copy()
+
         # Create lambda function if additional arguments present
         if "args" in config:
             args = config.pop("args")

src/python/txtai/workflow/task/service.py

@@ -54,7 +54,10 @@ class ServiceTask(Task):
 
     def execute(self, elements, executor=None):
         if self.batch:
-            elements = self.request(elements)
+            response = self.request(elements)
+
+            # Empty response body returns None for each element, same as batch=False
+            elements = [None] * len(elements) if response is None else response
         else:
             elements = [self.request(element) for element in elements]
 
@@ -91,8 +94,12 @@ class ServiceTask(Task):
         else:
             response = requests.post(self.url, json=params, timeout=self.timeout)
 
+        # Empty response body (e.g. 204 No Content), nothing to parse
+        if not response.content:
+            return None
+
         # Parse data based on content-type
-        mimetype = response.headers["Content-Type"].split(";")[0]
+        mimetype = response.headers.get("Content-Type", "").split(";")[0]
         if mimetype.lower().endswith("xml"):
             data = xmltodict.parse(response.text)
         else:

src/python/txtai/workflow/task/storage.py

@@ -54,8 +54,8 @@ class StorageTask(Task):
         outputs = []
         for element in elements:
             if self.matches(element):
-                # Get directory listing and run actions
-                outputs.extend(super().__call__(self.list(element), executor))
+                # Unpack tagged input, list container and re-tag each listing to match input format
+                outputs.extend(self.pack(element, result) for result in super().__call__(self.list(self.upack(element, True)), executor))
             else:
                 outputs.append(element)
 

test/python/testagent.py

@@ -232,6 +232,21 @@ class TestAgent(unittest.TestCase):
         # Timeout is disabled when set to None
         self.assertIsNone(BashTool(timeout=None).timeout)
 
+    def testToolsBashAllowedEmpty(self):
+        """
+        Test an empty allowed commands list permits no commands
+        """
+
+        # An empty allow list is an explicit choice and must be honored, not replaced by the defaults
+        self.assertEqual(BashTool(allowed=[]).allowed, [])
+
+        # No command runs when nothing is allowed
+        self.assertIsNone(BashTool(allowed=[])(["cat", "file.txt"]))
+
+        # The defaults still apply when allowed isn't set, including an explicit None
+        self.assertEqual(BashTool().allowed, ["cat", "cut", "diff", "grep", "head", "ls", "tail"])
+        self.assertEqual(BashTool(allowed=None).allowed, ["cat", "cut", "diff", "grep", "head", "ls", "tail"])
+
     @patch("subprocess.run")
     def testToolsBashEncoding(self, run):
         """

test/python/testann/testdense/testsqlite.py

@@ -140,6 +140,8 @@ class TestSQLite(DenseTest):
 
             # Load new copy and check that it has all the changes
             model = ANNFactory.create(model.config)
+            model.load(index)
+            self.assertEqual(model.count(), 500)
             model.load(new)
             self.assertEqual(model.count(), 509)
             self.assertEqual(len(model.search(np.random.rand(1, 240).astype(np.float32), 10)[0]), 10)

test/python/testapi/testapiworkflow.py

@@ -85,6 +85,12 @@ workflow:
               url: http://127.0.0.1:8001/testpost
               params:
 
+    empty:
+        tasks:
+            - task: service
+              url: http://127.0.0.1:8001/empty
+              params:
+
     xml:
         tasks:
             - task: service
@@ -144,6 +150,12 @@ class RequestHandler(BaseHTTPRequestHandler):
         length = int(self.headers["content-length"])
         data = json.loads(self.rfile.read(length))
 
+        if self.path.startswith("/empty"):
+            # Respond without a body or content-type
+            self.send_response(204)
+            self.end_headers()
+            return
+
         response = json.dumps([[y for y in x.split(".") if y] for x in data]).encode("utf-8")
 
         self.send_response(200)
@@ -235,6 +247,16 @@ class TestWorkflow(unittest.TestCase):
         self.assertEqual(len(results), 1)
         self.assertEqual(len(results[0]), 2)
 
+    def testServiceEmpty(self):
+        """
+        Test workflow with ServiceTask POST via API and an empty response
+        """
+
+        text = "This is a test sentence. And another sentence to split."
+        results = self.client.post("workflow", json={"name": "empty", "elements": [text]}).json()
+
+        self.assertEqual(results, [None])
+
     def testServiceXml(self):
         """
         Test workflow with ServiceTask GET via API and XML response

test/python/testdatabase/testduckdb.py

@@ -5,6 +5,7 @@ DuckDB module tests
 import os
 import unittest
 
+from txtai.database import DuckDB, SQLError
 from txtai.embeddings import Embeddings
 
 from .testrdbms import Common
@@ -45,6 +46,28 @@ class TestDuckDB(Common.TestRDBMS):
         if cls.embeddings:
             cls.embeddings.close()
 
+    def testAborted(self):
+        """
+        Test SQL errors that abort the transaction
+        """
+
+        embeddings = Embeddings(keyword=True, content=self.backend)
+        embeddings.index([(0, "apple pie", None), (1, "banana bread", None)])
+
+        # Catalog errors don't abort the transaction
+        with self.assertRaises(SQLError) as context:
+            embeddings.search("select id from txtai where unknownfn(text)")
+
+        self.assertNotIn(DuckDB.ABORTED, str(context.exception))
+
+        # Runtime errors abort the transaction
+        with self.assertRaises(SQLError) as context:
+            embeddings.search("select id from txtai where text = :x", parameters={"x": 1})
+
+        self.assertIn("This error aborted the DuckDB transaction.", str(context.exception))
+
+        embeddings.close()
+
     @unittest.skipIf(os.name == "nt", "testArchive skipped on Windows")
     def testArchive(self):
         """

test/python/testdatabase/testrdbms.py

@@ -239,6 +239,21 @@ class Common:
             self.assertEqual(result["text"], self.data[4])
             self.assertEqual(len(result.get("tokens")), 8)
 
+        def testExplainSimilarArgs(self):
+            """
+            Test query explain ignores extra similar clause arguments
+            """
+
+            self.embeddings.index([(uid, text, None) for uid, text in enumerate(self.data)])
+
+            expected = self.embeddings.explain("select id, text, score from txtai where similar('feel good story')", limit=1)[0]
+
+            # Candidates and weights are not part of the query text
+            for query in ["similar('feel good story', 10)", "similar('feel good story', 0.5)"]:
+                result = self.embeddings.explain(f"select id, text, score from txtai where {query}", limit=1)[0]
+                self.assertEqual(result["text"], self.data[4])
+                self.assertEqual(result["tokens"], expected["tokens"])
+
         def testExplainEmpty(self):
             """
             Test query explain with no filtering criteria
@@ -329,6 +344,33 @@ class Common:
             result = embeddings.search("feel good story", 1)[0]
             self.assertEqual(result["text"], data[0][1])
 
+        def testHybridWeights(self):
+            """
+            Test hybrid search with different similar clause weights in the same batch
+            """
+
+            queries = [
+                "select id, score from txtai where similar('a day at work', 0.1)",
+                "select id, score from txtai where similar('feel good story', 0.9)",
+            ]
+
+            # Test with sparse + dense vectors in the top-level index and in a subindex
+            hybrid = {"path": "sentence-transformers/nli-mpnet-base-v2", "hybrid": True}
+            for config in [hybrid, {"defaults": False, "indexes": {"index1": hybrid}}]:
+                embeddings = Embeddings({**config, "content": self.backend})
+                embeddings.index(self.data)
+
+                # Each query in a batch must use its own weights
+                batch = embeddings.batchsearch(queries, 6)
+                for query, results in zip(queries, batch):
+                    expected = embeddings.search(query, 6)
+                    self.assertEqual([r["id"] for r in results], [r["id"] for r in expected])
+                    for result, value in zip(results, expected):
+                        self.assertAlmostEqual(result["score"], value["score"], places=5)
+
+                # Cleanup connection
+                embeddings.close()
+
         def testIndex(self):
             """
             Test index

test/python/testdatabase/testsql.py

@@ -62,6 +62,17 @@ class TestSQL(unittest.TestCase):
         with self.assertRaises(SQLError):
             aggregate("select avg(price) from txtai", [{"avg(price)": 100.0}, {"avg(price)": 10.0}])
 
+    def testAggregateCase(self):
+        """
+        Test Aggregate combines aggregate columns written in any case
+        """
+
+        aggregate = Aggregate()
+
+        query = "select COUNT(*), Sum(price), MAX(price) from txtai"
+        results = [{"COUNT(*)": 1, "Sum(price)": 1.0, "MAX(price)": 1.0}, {"COUNT(*)": 2, "Sum(price)": 5.0, "MAX(price)": 4.0}]
+        self.assertEqual(aggregate(query, results), [{"COUNT(*)": 3, "Sum(price)": 6.0, "MAX(price)": 4.0}])
+
     def testAggregateNull(self):
         """
         Test Aggregate skips NULL values returned by shards with no matching rows
@@ -75,6 +86,17 @@ class TestSQL(unittest.TestCase):
         self.assertEqual(aggregate(query, [empty, result]), [result])
         self.assertEqual(aggregate(query, [empty, empty]), [empty])
 
+    def testAggregateOrderByNull(self):
+        """
+        Test Aggregate orders NULL values the same way as SQLite
+        """
+
+        aggregate = Aggregate()
+
+        results = [{"id": "a", "price": 3.0}, {"id": "b", "price": None}, {"id": "c", "price": 1.0}]
+        self.assertEqual([x["id"] for x in aggregate("select id, price from txtai order by price", results)], ["b", "c", "a"])
+        self.assertEqual([x["id"] for x in aggregate("select id, price from txtai order by price desc", results)], ["a", "c", "b"])
+
     def testAlias(self):
         """
         Test alias clauses
@@ -121,6 +143,17 @@ class TestSQL(unittest.TestCase):
         with self.assertRaises(SQLError):
             self.db.search("select a b c from txtai where id match id")
 
+    def testMultipleStatements(self):
+        """
+        Test only the first statement is run
+        """
+
+        with self.assertRaises(SQLError):
+            self.db.search("select * from txtai ; delete from sections")
+
+        self.assertSql("select", ";select id from txtai", "s.id")
+        self.assertSql("select", "select id from txtai;", "s.id")
+
     def testBracket(self):
         """
         Test bracket expressions

test/python/testembeddings.py

@@ -785,6 +785,10 @@ class TestEmbeddings(unittest.TestCase):
         # SQL query with a similar() clause extracts its terms
         self.assertEqual(embeddings.terms("select id, text from txtai where similar('feel good story')"), "feel good story")
 
+        # Candidates, index and weights arguments are not keyword terms
+        self.assertEqual(embeddings.terms("select id, text from txtai where similar('feel good story', 10)"), "feel good story")
+        self.assertEqual(embeddings.terms("select id, text from txtai where similar('feel good story', 'index1', 0.5)"), "feel good story")
+
         # SQL query without a similar() clause has no keyword terms (previously raised KeyError)
         self.assertEqual(embeddings.terms("select id, text from txtai where id = 1"), None)
 

test/python/testgraph.py

@@ -173,10 +173,16 @@ class TestGraph(unittest.TestCase):
         self.assertEqual(list(graph.scan(attribute="field")), [str(x) for x in range(5)])
         self.assertEqual(list(graph.filter([0]).scan()), [0])
 
+        # Attribute changes must be written through to the database
+        graph.addattribute(0, "field", "updated")
+        graph.addattribute(0, "label", "new")
+
         # Test save/load
         graph.save(None)
         graph.load(None)
         self.assertEqual(list(graph.scan()), [str(x) for x in range(5)])
+        self.assertEqual(graph.node(0), {"field": "updated", "label": "new"})
+        self.assertEqual(graph.edgecount(), 10)
 
         # Test remove node
         graph.delete([0])
@@ -314,6 +320,30 @@ class TestGraph(unittest.TestCase):
         self.assertEqual(len(graph.topics), 6)
         self.assertEqual(len(graph.categories), 6)
 
+    def testInferTopics(self):
+        """
+        Test inferring topics when most neighbors have no topic
+        """
+
+        graph = GraphFactory.create({"backend": "networkx"})
+        graph.initialize()
+        graph.topics = {"t": [0]}
+
+        graph.addnode(0, topic="t", topicrank=0, category="c")
+        for x in range(1, 4):
+            graph.addnode(x, updated=True)
+        graph.addnode(4, updated=True)
+
+        for x, y in [(1, 2), (2, 3), (1, 3), (1, 0)]:
+            graph.addedge(x, y)
+
+        graph.infertopics()
+
+        self.assertEqual([graph.attribute(x, "topic") for x in range(4)], ["t"] * 4)
+        self.assertEqual(graph.topics["t"], [0, 1, 2, 3])
+        self.assertEqual(graph.attribute(1, "category"), "c")
+        self.assertIsNone(graph.attribute(4, "topic"))
+
     @patch.dict(os.environ, {"ALLOW_PICKLE": "True"})
     def testLegacy(self):
         """

test/python/testpipeline/testaudio/testsignal.py

@@ -0,0 +1,33 @@
+"""
+Signal module tests
+"""
+
+import unittest
+
+import numpy as np
+
+from txtai.pipeline.audio.signal import Signal
+
+
+class TestSignal(unittest.TestCase):
+    """
+    Signal tests.
+    """
+
+    def testTrimPartialWindow(self):
+        """
+        Test that short inputs and partial windows are not lost when trimming
+        """
+
+        # At 1000 Hz the detection window has 40 samples.
+        pattern = np.array([1.0, -1.0], dtype=np.float32)
+        for size in (0, 20, 43, 83):
+            audio = np.resize(pattern, size)
+            for trailing in (False, True):
+                with self.subTest(size=size, trailing=trailing):
+                    np.testing.assert_array_equal(Signal.trim(audio, 1000, trailing=trailing), audio)
+
+        # Leading silence is still removed while the partial tail is retained.
+        audio = np.resize(pattern, 83)
+        padded = np.concatenate((np.zeros(40, dtype=np.float32), audio))
+        np.testing.assert_array_equal(Signal.trim(padded, 1000, trailing=False), audio)

test/python/testpipeline/testaudio/testtexttospeech.py

@@ -4,7 +4,9 @@ TextToSpeech module tests
 
 import unittest
 
-from unittest.mock import patch
+from unittest.mock import call, patch
+
+import numpy as np
 
 from txtai.pipeline import TextToSpeech
 
@@ -80,3 +82,26 @@ class TestTextToSpeech(unittest.TestCase):
         # Check that data is generated
         self.assertGreater(len(speech), 0)
         self.assertEqual(rate, 22050)
+
+    @patch("txtai.pipeline.audio.texttospeech.Kokoro")
+    @patch.object(TextToSpeech, "hasfile")
+    def testStreamingOptions(self, hasfile, kokoro):
+        """
+        Test backend options reach both flushed and final streaming segments
+        """
+
+        hasfile.side_effect = lambda _, name: name in ("model.onnx", "voices.json")
+        audio = np.array([0.1, -0.1], dtype=np.float32)
+        kokoro.return_value.return_value = (audio, 24000)
+        tts = TextToSpeech("neuml/kokoro-int8-onnx", rate=None)
+
+        chunks = list(tts(iter(["One ", "two ", "three.", "Tail"]), stream=True, speaker="af", speed=1.5, transcribe=False))
+
+        self.assertEqual(
+            kokoro.return_value.call_args_list,
+            [call("One two three.", "af", speed=1.5, transcribe=False), call("Tail", "af", speed=1.5, transcribe=False)],
+        )
+        self.assertEqual(len(chunks), 2)
+        for data, rate in chunks:
+            np.testing.assert_array_equal(data, audio)
+            self.assertEqual(rate, 24000)

test/python/testpipeline/testdata/testtabular.py

@@ -23,53 +23,6 @@ class TestTabular(unittest.TestCase):
 
         cls.tabular = Tabular("id", ["text"])
 
-    def testFalsyValues(self):
-        """
-        Test that zero and False are indexed as content
-        """
-
-        tabular = Tabular("id", ["quantity", "text"])
-
-        # A quantity of 0 is real data. Before this was fixed, the truthy check in concat
-        # dropped it and the row indexed as "Widget", losing the quantity.
-        rows = tabular([{"id": 1, "quantity": 0, "text": "Widget"}])
-        self.assertEqual(rows[0][1], "0. Widget")
-
-        rows = tabular([{"id": 2, "quantity": False, "text": "Widget"}])
-        self.assertEqual(rows[0][1], "False. Widget")
-
-    def testNullValues(self):
-        """
-        Test that null and empty values are still skipped
-        """
-
-        tabular = Tabular("id", ["quantity", "text"])
-
-        # NaN is the null marker column() normalizes to None, and an empty string adds
-        # nothing but a separator. Both stay excluded.
-        rows = tabular([{"id": 1, "quantity": float("nan"), "text": "Widget"}])
-        self.assertEqual(rows[0][1], "Widget")
-
-        rows = tabular([{"id": 2, "quantity": "", "text": "Widget"}])
-        self.assertEqual(rows[0][1], "Widget")
-
-    def testNumericTypes(self):
-        """
-        Mixed numeric rows preserve integer IDs and content without float rounding.
-        """
-
-        data = [{"id": 2**53 + 1, "count": 2, "value": 1.5}, {"id": 2**53, "count": 3, "value": 2.5}]
-        for content in (False, True, ["id"]):
-            with self.subTest(content=content):
-                rows = Tabular("id", ["count", "value"], content)(data)
-                step = 2 if content else 1
-                self.assertTrue(all(isinstance(row[0], int) for row in rows))
-                self.assertEqual([row[0] for row in rows[::step]], [row["id"] for row in data])
-                self.assertEqual([row[1] for row in rows[::step]], ["2. 1.5", "3. 2.5"])
-                if content:
-                    expected = data if content is True else [{"id": row["id"]} for row in data]
-                    self.assertEqual([row[1] for row in rows[1::2]], expected)
-
     def testContent(self):
         """
         Test parsing additional content
@@ -120,6 +73,31 @@ class TestTabular(unittest.TestCase):
         self.assertEqual(uid, 0)
         self.assertEqual(text, "This is a test")
 
+    def testFalsyIDColumn(self):
+        """
+        Explicit column labels are used for IDs and excluded from default text.
+        """
+
+        for column in ("", 0, "id"):
+            with self.subTest(column=column):
+                rows = Tabular(idcolumn=column)([{column: "doc-7", "text": "hello"}])
+                self.assertEqual(rows, [("doc-7", "hello", None)])
+
+    def testFalsyValues(self):
+        """
+        Test that zero and False are indexed as content
+        """
+
+        tabular = Tabular("id", ["quantity", "text"])
+
+        # A quantity of 0 is real data. Before this was fixed, the truthy check in concat
+        # dropped it and the row indexed as "Widget", losing the quantity.
+        rows = tabular([{"id": 1, "quantity": 0, "text": "Widget"}])
+        self.assertEqual(rows[0][1], "0. Widget")
+
+        rows = tabular([{"id": 2, "quantity": False, "text": "Widget"}])
+        self.assertEqual(rows[0][1], "False. Widget")
+
     def testInvalid(self):
         """
         Test invalid file paths
@@ -157,15 +135,25 @@ class TestTabular(unittest.TestCase):
 
         self.assertIsNone(data["metadata"])
 
-    def testFalsyIDColumn(self):
+    def testMissingIDFallsBack(self):
         """
-        Explicit column labels are used for IDs and excluded from default text.
+        Rows without an ID (missing key, None or NaN) fall back to the row index.
+        pandas coerces the missing value to NaN, which previously bypassed the
+        is-not-None fallback and indexed rows with a NaN uid.
         """
 
-        for column in ("", 0, "id"):
-            with self.subTest(column=column):
-                rows = Tabular(idcolumn=column)([{column: "doc-7", "text": "hello"}])
-                self.assertEqual(rows, [("doc-7", "hello", None)])
+        tabular = Tabular("id", ["text"])
+
+        rows = tabular(
+            [
+                {"id": "a", "text": "first"},
+                {"text": "second"},
+                {"id": None, "text": "third"},
+                {"id": float("nan"), "text": "fourth"},
+                {"id": "c", "text": "fifth"},
+            ]
+        )
+        self.assertEqual([row[0] for row in rows], ["a", 1, 2, 3, "c"])
 
     def testNoColumns(self):
         """
@@ -178,3 +166,35 @@ class TestTabular(unittest.TestCase):
 
         self.assertEqual(uid, 0)
         self.assertEqual(text, "This is a test. Describes text in more detail")
+
+    def testNullValues(self):
+        """
+        Test that null and empty values are still skipped
+        """
+
+        tabular = Tabular("id", ["quantity", "text"])
+
+        # NaN is the null marker column() normalizes to None, and an empty string adds
+        # nothing but a separator. Both stay excluded.
+        rows = tabular([{"id": 1, "quantity": float("nan"), "text": "Widget"}])
+        self.assertEqual(rows[0][1], "Widget")
+
+        rows = tabular([{"id": 2, "quantity": "", "text": "Widget"}])
+        self.assertEqual(rows[0][1], "Widget")
+
+    def testNumericTypes(self):
+        """
+        Mixed numeric rows preserve integer IDs and content without float rounding.
+        """
+
+        data = [{"id": 2**53 + 1, "count": 2, "value": 1.5}, {"id": 2**53, "count": 3, "value": 2.5}]
+        for content in (False, True, ["id"]):
+            with self.subTest(content=content):
+                rows = Tabular("id", ["count", "value"], content)(data)
+                step = 2 if content else 1
+                self.assertTrue(all(isinstance(row[0], int) for row in rows))
+                self.assertEqual([row[0] for row in rows[::step]], [row["id"] for row in data])
+                self.assertEqual([row[1] for row in rows[::step]], ["2. 1.5", "3. 2.5"])
+                if content:
+                    expected = data if content is True else [{"id": row["id"]} for row in data]
+                    self.assertEqual([row[1] for row in rows[1::2]], expected)

test/python/testpipeline/testdata/testtextractor.py

@@ -147,10 +147,54 @@ class TestTextractor(unittest.TestCase):
         self.assertMarkdown("<p>This is a <em>test</em></p>", "This is a *test*")
         self.assertMarkdown("<p>This is a <a href='link'>test</a>", "This is a [test](link)")
 
+        # Leading whitespace inside a formatting tag
+        self.assertMarkdown("<p>This is<b> a test</b></p>", "This is **a test**")
+        self.assertMarkdown("<p>This is<em> a test</em></p>", "This is *a test*")
+        self.assertMarkdown("<p>This is<a href='link'> a test</a></p>", "This is [a test](link)")
+        self.assertMarkdown("<p>This is a te<b>st</b></p>", "This is a te**st**")
+        self.assertMarkdown("<p>This is\n<b> a test</b></p>", "This is\n**a test**")
+
+        # Links without a target are plain text
+        self.assertMarkdown("<p>This is a <a name='anchor'>test</a></p>", "This is a test")
+        self.assertMarkdown("<p>This is<a name='anchor'> a test</a></p>", "This is a test")
+        self.assertMarkdown("<h2><a id='anchor'>This is a test</a></h2>", "## This is a test")
+
         # Collapse to outer tag
         self.assertMarkdown("<p>This is a <strong><em>test</em></strong></p>", "This is a **test**")
         self.assertMarkdown("<p>This is a <em><strong>test</strong></em></p>", "This is a *test*")
 
+        # Line breaks
+        self.assertMarkdown("<p>Test1<br>Test2</p>", "Test1\nTest2")
+        self.assertMarkdown("<p>This is a <b>test<br/>case</b></p>", "This is a **test\ncase**")
+        self.assertMarkdown("<pre>Test1<br>Test2</pre>", "```\nTest1\nTest2\n```")
+
+        # Line breaks in headings and table cells are replaced with a space
+        self.assertMarkdown("<h1>This is<br>a test</h1>", "# This is a test")
+        self.assertMarkdown(
+            "<table><tr><th>Header1</th><th>Header2</th></tr><tr><td>Test1<br>Test2</td><td>Test3</td></tr></table>",
+            "|Header1|Header2|\n|---|---|\n|Test1 Test2|Test3|",
+        )
+
+    def testHTMLListStart(self):
+        """
+        Test ordered lists retain their starting number
+        """
+
+        for start, expected in [("5", "5. Test1\n6. Test2"), ("0", "0. Test1\n1. Test2"), ("1", "1. Test1\n2. Test2")]:
+            with self.subTest(start=start):
+                self.assertMarkdown(f"<ol start='{start}'><li>Test1</li><li>Test2</li></ol>", expected)
+
+        self.assertMarkdown(
+            "<ol start='5'><li>Test1<ol start='3'><li>Test2</li></ol></li><li>Test3</li></ol>",
+            "5. Test1\n\t3. Test2\n6. Test3",
+        )
+
+        for start in ["", "invalid", "1.5"]:
+            with self.subTest(start=start):
+                self.assertMarkdown(f"<ol start='{start}'><li>Test1</li><li>Test2</li></ol>", "1. Test1\n2. Test2")
+
+        self.assertMarkdown("<ul start='invalid'><li>Test1</li><li>Test2</li></ul>", "- Test1\n- Test2")
+
     def testHTMLMetadataNoContent(self):
         """
         Test HTML with a meta description tag that has no content attribute

test/python/testpipeline/testtext/testreranker.py

@@ -238,6 +238,26 @@ class TestReranker(unittest.TestCase):
             self.assertEqual(ranker("select id, text, score from txtai where id = 'missing'"), [])
             self.assertEqual(encode.call_count, 0)
 
+    def testEmpty(self):
+        """Return no results when a search has no matches, with or without a cache."""
+
+        empty = "select id, text, score from txtai where id = 'missing'"
+        models = [
+            self.similarity,
+            Similarity("cross-encoder/ms-marco-MiniLM-L-2-v2", crossencode=True),
+            Similarity("prajjwal1/bert-medium-mnli"),
+        ]
+
+        for similarity in models:
+            for cache in (None, True):
+                ranker = Reranker(self.embeddings, similarity, cache=cache)
+                self.assertEqual(ranker(empty), [])
+
+                # Empty and non-empty queries in the same batch
+                results = ranker([empty, "lottery winner"], 1)
+                self.assertEqual(results[0], [])
+                self.assertEqual(len(results[1]), 1)
+
     def testCacheBypass(self):
         """Keep text scoring for missing ids, non-late encoders and batch centering."""
 

test/python/testworkflow.py

@@ -3,6 +3,7 @@ Workflow module tests
 """
 
 import contextlib
+import copy
 import glob
 import io
 import os
@@ -16,10 +17,13 @@ import numpy as np
 import torch
 
 from txtai.api import API
+from txtai.app import Application
 from txtai.embeddings import Documents, Embeddings
 from txtai.pipeline import Nop, Segmentation, Summary, Translation, Textractor
 from txtai.workflow import (
     Workflow,
+    WorkflowFactory,
+    TaskFactory,
     Task,
     ConsoleTask,
     ExportTask,
@@ -76,6 +80,41 @@ class TestWorkflow(unittest.TestCase):
                     - transform
         """
 
+    def testApplicationConfig(self):
+        """
+        Test that an application can reuse its workflow configuration
+        """
+
+        workflow = {
+            "tasks": [{"task": "template", "template": "Hi {text}"}, {"action": "nop", "args": []}],
+            "stream": {"task": "stream", "action": "nop", "batch": True},
+        }
+        config = {"workflow": {"test": workflow}}
+        expected = copy.deepcopy(config)
+        for _ in range(2):
+            app = Application(config)
+            self.assertEqual(list(app.workflow("test", ["x"])), ["Hi x"])
+            self.assertEqual(config, expected)
+
+    def testFactoryConfig(self):
+        """
+        Test that factories preserve task, argument and stream configuration
+        """
+
+        def action(values, suffix):
+            return [value + suffix for value in values]
+
+        args = ["!"]
+        task = {"action": action, "args": args}
+        config = {"tasks": [dict(task, task="")], "stream": {"task": "stream", "action": iter, "batch": True}}
+        expected = copy.deepcopy(config)
+        for _ in range(2):
+            self.assertEqual(TaskFactory.create(task, "")(["x"]), ["x!"])
+            self.assertIs(task["action"], action)
+            self.assertIs(task["args"], args)
+            self.assertEqual(list(WorkflowFactory.create(config, "test")(["x"])), ["x!"])
+            self.assertEqual(config, expected)
+
     def testBaseWorkflow(self):
         """
         Test a basic workflow
@@ -524,6 +563,36 @@ class TestWorkflow(unittest.TestCase):
 
         self.assertEqual(len(results), 22)
 
+    def testStorageTaggedWorkflow(self):
+        """
+        Test a storage task with tagged (id, data, tag) inputs (#1328)
+        """
+
+        task = StorageTask()
+
+        # matches() accepts untagged and tagged storage urls
+        self.assertIsNotNone(task.matches("local://" + Utils.PATH))
+        self.assertIsNotNone(task.matches(("id", "local://" + Utils.PATH, "tag")))
+        self.assertIsNone(task.matches("test string"))
+
+        workflow = Workflow([StorageTask()])
+
+        # Tagged inputs list the container and re-tag each file with the input (id, tag)
+        url = "local://" + Utils.PATH
+        results = list(workflow([("id", url, "tag")]))
+        self.assertEqual(len(results), 21)
+        for result in results:
+            self.assertEqual(len(result), 3)
+            self.assertEqual(result[0], "id")
+            # normpath: libcloud joins listing paths with os.path.join, which uses
+            # backslashes on Windows while Utils.PATH uses forward slashes
+            self.assertTrue(os.path.normpath(result[1]).startswith(os.path.normpath(Utils.PATH)))
+            self.assertEqual(result[2], "tag")
+
+        # Non-storage inputs pass through unmodified
+        results = list(workflow([("id2", "test string", "tag2")]))
+        self.assertEqual(results, [("id2", "test string", "tag2")])
+
     def testTemplateInput(self):
         """
         Test template task input