diff --git a/examples/workflows.py b/examples/workflows.py index 62278557..5bc9e627 100644 --- a/examples/workflows.py +++ b/examples/workflows.py @@ -6,6 +6,7 @@ Requires streamlit to be installed. """ import contextlib +import copy import os import re import tempfile @@ -19,10 +20,7 @@ import pandas as pd import streamlit as st import txtai.api.application - -from txtai.embeddings import Documents, Embeddings -from txtai.pipeline import Segmentation, Summary, Tabular, Textractor, Transcription, Translation -from txtai.workflow import ServiceTask, Task, UrlTask, Workflow +import txtai.app class Server(uvicorn.Server): @@ -68,97 +66,55 @@ class Process: """ @staticmethod - @st.cache(max_entries=1, allow_output_mutation=True, show_spinner=False) - def get(components, data): + @st.cache_resource(show_spinner=False) + def get(name, config): """ Lookup or creates a new workflow process instance. Args: - components: input components - data: initial data, only passed when indexing + name: workflow name + config: application configuration Returns: Process """ - process = Process(data) + process = Process() # Build workflow with st.spinner("Building workflow...."): - process.build(components) + process.build(name, config) return process - def __init__(self, data): + def __init__(self): """ Creates a new Process. + """ + + # Application handle + self.application = None + + # Workflow name + self.name = None + + # Workflow data + self.data = None + + def build(self, name, config): + """ + Builds an application. Args: - data: initial data, only passed when indexing + name: workflow name + config: application configuration """ - # Component options - self.components = {} + # Create application + self.application = txtai.app.Application(config) - # Defined pipelines - self.pipelines = {} - - # Current workflow - self.workflow = [] - - # Embeddings index params - self.embeddings = None - self.documents = None - self.data = data - - def build(self, components): - """ - Builds a workflow using components. - - Args: - components: list of components to add to workflow - """ - - # pylint: disable=W0108 - tasks = [] - for component in components: - component = dict(component) - wtype = component.pop("type") - self.components[wtype] = component - - if wtype == "embeddings": - self.embeddings = Embeddings({**component}) - self.documents = Documents() - tasks.append(Task(self.documents.add, unpack=False)) - - elif wtype == "segmentation": - self.pipelines[wtype] = Segmentation(**self.components[wtype]) - tasks.append(Task(self.pipelines[wtype])) - - elif wtype == "service": - tasks.append(ServiceTask(**self.components[wtype])) - - elif wtype == "summary": - self.pipelines[wtype] = Summary(component.pop("path")) - tasks.append(Task(lambda x: self.pipelines["summary"](x, **self.components["summary"]))) - - elif wtype == "tabular": - self.pipelines[wtype] = Tabular(**self.components[wtype]) - tasks.append(Task(self.pipelines[wtype])) - - elif wtype == "textractor": - self.pipelines[wtype] = Textractor(**self.components[wtype]) - tasks.append(UrlTask(self.pipelines[wtype])) - - elif wtype == "transcription": - self.pipelines[wtype] = Transcription(component.pop("path")) - tasks.append(UrlTask(self.pipelines[wtype], r".\.wav$")) - - elif wtype == "translation": - self.pipelines[wtype] = Translation() - tasks.append(Task(lambda x: self.pipelines["translation"](x, **self.components["translation"]))) - - self.workflow = Workflow(tasks) + # Workflow name + self.name = name def run(self, data): """ @@ -168,27 +124,24 @@ class Process: data: input data """ - if data and self.workflow: + if data and self.application: # Build tuples for embedding index - if self.documents: + if self.application.embeddings: data = [(x, element, None) for x, element in enumerate(data)] # Process workflow - for result in self.workflow(data): - if not self.documents: - st.write(result) + with st.spinner("Running workflow...."): + results = [] + for result in self.application.workflow(self.name, data): + # Store result + results.append(result) - # Build embeddings index - if self.documents: - # Cache data - self.data = list(self.documents) + # Write result if this isn't an indexing workflow + if not self.application.embeddings: + st.write(result) - with st.spinner("Building embedding index...."): - self.embeddings.index(self.documents) - self.documents.close() - - # Clear workflow - self.documents, self.pipelines, self.workflow = None, None, None + # Store workflow results + self.data = results def search(self, query): """ @@ -198,7 +151,7 @@ class Process: query: input query """ - if self.embeddings and query: + if self.application and query: st.markdown( """