i did something but i need to make workflows now

app.py

@@ -7,6 +7,537 @@ import pandas as pd
 import streamlit as st
 
 from txtai.embeddings import Documents, Embeddings
-from txtai.pipeline import Summary, Tabular, Textractor, Translation
+from txtai.pipeline import Segmentation, Summary, Tabular, Textractor, Translation
 from txtai.workflow import ServiceTask, Task, UrlTask, Workflow
 
+class Process:
+
+    @staticmethod
+    @st.cache(ttl=60 * 60, max_entries=3, allow_output_mutation=True, show_spinner=False)
+    def get(components, data):
+        """
+        Lookup or creates a new workflow process instance
+        """
+
+        process = Process(data)
+
+        with st.spinner("Building workflow...."):
+            process.build(components)
+
+        return process
+    
+    def __init__(self, data):
+        """
+        Create new Process
+        """
+
+        self.components = {}
+
+        self.pipelines = {}
+
+        self. workflow = []
+
+        self.embeddings = None
+        self.documents = None
+        self.data = data
+
+    def build(self, components):
+        """
+        Builds a workflow using components
+        """
+
+        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 == "translation":
+                self.pipelines[wtype] = Translation()
+                tasks.append(Task(lambda x: self.pipelines["translation"](x, **self.components["translation"])))
+
+        self.workflow = Workflow(tasks)
+
+    def run(self, data):
+        """
+        Runs a workflow using data as input
+        """
+
+        if data and self.workflow:
+            # Builds tuples for embedding index
+            if self.documents:
+                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)
+
+            # Build embedding index
+            if self.documents:
+                # Cache data
+                self.data = list(self.documents)
+
+                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
+
+    def search(self, query):
+        """
+        Runs a search for query
+        """
+        if self.embeddings and query:
+            st.markdown(
+                """
+            <style>
+            table td:nth-child(1) {
+                display: none
+            }
+            table th:nth-child(1) {
+                display: none
+            }
+            table {text-align: left !important}
+            </style>
+            """,
+                unsafe_allow_html=True,
+            )
+
+            limit = min(5, len(self.data))
+
+            results = []
+            for result in self.embeddings.search(query, limit):
+                # Tuples are returned when an index doesn't have stored content
+                if isinstance(result, tuple):
+                    uid, score = result
+                    results.append({"text": self.find(uid), "score": f"{score:.2}"})
+                else:
+                    if "id" in result and "text" in result:
+                        result["text"] = self.content(result.pop("id"), result["text"])
+                    if "score" in result and result["score"]:
+                        result["score"] = f'{result["score"]:.2}'
+
+                    results.append(result)
+
+            df = pd.DataFrame(results)
+            st.write(df.to_html(escape=False), unsafe_allow_html=True)
+
+    def find(self, key):
+        """
+        Lookup record from cached data by uid key
+        """
+
+        # Lookup text by id
+        text = [text for uid, text, _ in self.data if uid == key][0]
+        return self.content(key, text)
+    
+    def content(self, uid, text):
+        """
+        Builds a content reference for uid and text
+        """
+
+        if uid and uid.lower().startswith("http"):
+            return f"<a href='{uid}' rel='noopener noreferrer' target='blank'>{text}</a>"
+        
+        return text
+    
+class Application:
+    """
+    Main application
+    """
+
+    def __init__(self, directory):
+        """
+        Creates a new application
+        """
+
+        # Workflow configuration directory
+        self.directory = directory
+
+    def default(self, names):
+        """
+        Gets default workflow index
+        """
+
+        # Gets names as lowercase to match case sensitive
+        lnames = [name.lower() for name in names]
+
+        # Get default workflow param
+        params = st.experimental_get_query_params()
+        index = params.get("default")
+        index = index[0].lower() if index else 0
+
+        # Lookup index of workflow name, add 1 to account for "--"
+        if index and index in lnames:
+            return lnames.index(index) + 1
+        
+        # Workflow not found, default to index 0
+        return 0
+    
+    def load(self, components):
+        """
+        Load an existing workflow file
+        """
+
+        with open(os.path.join(self.directory, "config.yml"), encoding="utf-8") as f:
+            config = yaml.safe_load(f)
+
+        names = [row["name"] for row in config]
+        files = [row["file"] for row in config]
+
+        selected = st.selectbox("Load workflow", ["--"] + names, self.default(names))
+        if selected != "--":
+            index = [x for x, name in enumerate(names) if name == selected][0]
+            with open(os.path.join(self.directory, files[index]), encoding="utf-8") as f:
+                workflow = yaml.safe_load(f)
+
+            st.markdown("---")
+
+            # Get tasks for first workflow
+            tasks = list(workflow["workflow"].values())[0]["tasks"]
+            selected = []
+
+            for task in tasks:
+                name = task.get("action", task.get("task"))
+                if name in components:
+                    selected.append(name)
+                elif name in ["index", "upsert"]:
+                    selected.append("embeddings")
+
+            return (selected, workflow)
+        
+        return (None, None)
+    
+    def state(self, key):
+        """
+        Lookup a session state variable
+        """
+
+        if key in st.session_state:
+            return st.session_state[key]
+        
+        return None
+    
+    def appsetting(self, workflow, name):
+        """
+        Looks up an application configuration setting
+        """
+
+        if workflow:
+            config = workflow.get("app")
+            if config:
+                return config.get(name)
+            
+        return None
+    
+    def setting(self, config, name, default=None):
+        """
+        Looks up a component configuration settings
+        """
+
+        return config.get(name, default) if config else default
+    
+    def text(self, label, component, config, name, default=None):
+        """
+        Create a new text input field
+        """
+
+        default = self.setting(config, name, default)
+        if not default:
+            default = ""
+        elif isinstance(default, list):
+            default = ",".join(default)
+        elif isinstance(default, dict):
+            default = ",".join(default.keys())
+
+        st.caption(label)
+        st.code(default, language="yaml")
+        return default
+    
+    def number(self, label, component, config, name, default=None):
+        """
+        Creates a new numeric input field
+        """
+
+        value = self.text(label, component, config, name, default)
+        return int(value) if value else None
+    
+    def boolean(self, label, component, config, name, default=None):
+        """
+        Creates a new checkbox field
+        """
+
+        default = self.setting(config, name, default)
+
+        st.caption(label)
+        st.markdown(":white_check_mark:" if default else ":white_large_square:")
+        return default
+    
+    def select(self, label, component, config, name, options, default=0):
+        """
+        Creates a new select box field
+        """
+
+        index = self.setting(config, name)
+        index = [x for x, option in enumerate(options) if option == default]
+
+        # Derive default index
+        default = index[0] if index else default
+
+        st.caption(label)
+        st.code(options[default], language="yaml")
+        return options[default]
+    
+    def split(self, text):
+        """
+        Splits text on commas and returns a list
+        """
+
+        return [x.strip() for x in text.split(",")]
+    
+    def options(self, component, workflow, index):
+        """
+        Extracts component settings into a component configuration dict
+        """
+
+        options = {"type": component}
+
+        config = None
+        if workflow:
+            if component in ["service", "translation"]:
+                tasks = list(workflow["workflow"].values())[0]["tasks"]
+                tasks = [task for task in tasks if task.get("task") == component or task.get("action") == component]
+                if tasks:
+                    config = tasks[0]
+            else:
+                config = workflow.get(component)
+
+        if component == "embeddings":
+            st.markdown(f"** {index + 1}.) Embeddings Index**  \n*Index workflow output*")
+            options["path"] = self.text("Embeddings model path", component, config, "path", "sentence-transformers/nli-mpnet-base-v2")
+            options["upsert"] = self.boolean("Upsert", component, config, "upsert")
+            options["content"] = self.boolean("Content", component, config, "content")
+
+        elif component in ("segmentation", "textractor"):
+            if component == "segmentation":
+                st.markdown(f"** {index + 1}.) Segment**  \n*Split text into semantic units*")
+            else:
+                st.markdown(f"** {index + 1}.) Textract**  \n*Extract text from documents")
+
+            options["sentences"] = self.boolean("Split sentences", component, config, "sentences")
+            options["lines"] = self.boolean("Split lines", component, config, "lines")
+            options["paragraphs"] = self.boolean("Split paragraphs", component, config, "paragraphs")
+            options["joint"] = self.boolean("Join tokenized", component, config, "join")
+            options["minlength"] = self.number("Min section length", component, config, "minlength")
+
+        elif component == "service":
+            st.markdown(f"** {index + 1}.) Service**  \n*Extract data from an API*")
+            options["url"] = self.text("URL", component, config, "url")
+            options["method"] = self.select("Method", component, config, "method", ["get", "post"], 0)
+            options["params"] = self.text("URL parameters", component, config, "params")
+            options["batch"] = self.boolean("Run as batch", component, config, "batch", True)
+            options["extract"] = self.text("Subsection(s) to extract", component, config, "extract")
+
+            if options["params"]:
+                options["params"] = {key: None for key in self.split(options["params"])}
+            if options["extract"]:
+                options["extract"] = self.split(options["extract"])
+
+        elif component == "summary":
+            st.markdown(f"** {index + 1}.) Summary**  \n*Abstractive text summarization*")
+            options["path"] = self.text("Model", component, config, "path", "sshleifer/distilbart-cnn-12-6")
+            options["minlength"] = self.number("Min length", component, config, "minlength")
+            options["maxlength"] = self.number("Max length", component, config, "maxlength")
+
+        elif component == "tabular":
+            st.markdown(f"** {index + 1}.) Tabular**  \n*Split tabular data into rows and columns*")
+            options["idcolumn"] = self.text("Id columns", component, config, "idcolumn")
+            options["textcolumns"] = self.text("Text columns", component, config, "textcolumns")
+            options["content"] = self.text("Content", component, config, "content")
+
+            if options["textcolumns"]:
+                options["textcolumns"] = self.split(options["textcolumns"])
+
+            if options["content"]:
+                options["content"] = self.split(options["content"])
+                if len(options["content"]) == 1 and options["content"][0] == "1":
+                    options["content"] = options["content"][0]
+
+        elif component == "translation":
+            st.markdown(f"** {index + 1}.) Translate**  \n*Machine translation*")
+            options["target"] = self.text("Target language code", component, config, "args", "en")
+
+        st.markdown("---")
+
+        return options
+    
+    def yaml(self, components):
+        """
+        Builds yaml string for components
+        """
+
+        data = {"app": {"data": self.state("data"), "query": self.state("query")}}
+        tasks = []
+        name = None
+
+        for component in components:
+            component = dict(component)
+            name = wtype = component.pop("type")
+
+            if wtype == "embeddings":
+                upsert = component.pop("upsert")
+
+                data[wtype] = component
+                data["writable"] = True
+
+                name = "index"
+                tasks.append({"action": "upsert" if upsert else "index"})
+
+            elif wtype == "segmentation":
+                data[wtype] = component
+                tasks.append({"action": wtype})
+
+            elif wtype == "service":
+                config = dict(**component)
+                config["task"] = wtype
+                tasks.append(config)
+
+            elif wtype == "summary":
+                data[wtype] = {"path": component.pop("path")}
+                tasks.append({"action": wtype})
+
+            elif wtype == "tabular":
+                data[wtype] = component
+                tasks.append({"action": wtype})
+
+            elif wtype == "textractor":
+                data[wtype] = component
+                tasks.append({"action": wtype, "tasks": "url"})
+
+            elif wtype == "translation":
+                data[wtype] = component
+                tasks.append({"action": wtype, "args": list(component.values())})
+
+        # Add in workflow
+        data["workflow"] = {name: {"tasks": tasks}}
+
+        return (name, yaml.dump(data))
+    
+    def data(self, workflow):
+        """
+        Gets input data
+        """
+
+        # Get default data setting
+        data = self.appsetting(workflow, "data")
+        if not self.appsetting(workflow, "query"):
+            data = st.text_input("Input", value=data)
+
+        # Save data state
+        st.session_state["data"] = data
+
+        # Wrap data as list for workflow processing
+        return [data]
+    
+    def query(self, workflow, index):
+        """
+        Gets input query
+        """
+
+        default = self.appsetting(workflow, "query")
+        default = default if default else ""
+
+        # Get query if this is an indexing workflow
+        query = st.text_input("Query", value=default) if index else None
+
+        # Save query state
+        st.session_state["query"] = query
+
+        return query
+    
+    def process(self, workflow, components, index):
+        """
+        Processes the current application action
+        """
+
+        # Get input data and initialize query
+        data = self.data(workflow)
+        query = self.query(workflow, index)
+
+        # Get workflow process
+        process = Process.get(components, data if index else None)
+
+        # Run workflow process
+        process.run(data)
+
+        # Run search
+        if index:
+            process.search(query)
+
+    def run(self):
+        """
+        Runs Streamlit application
+        """
+
+        with st.sidebar:
+            st.markdown("# Workflow builder for Station  \n*Build and apply workflows to data about articles*  ")
+            st.markdown("This is a demo for Station and the data used is from [Hugging Face](https://huggingface.co/datasets/ag_news/viewer/default/train).")
+            st.markdown("---")
+
+            # Component configuration
+            components = ["embeddings", "segmentation", "service", "summary", "tabular", "textractor", "translation"]
+
+            selected, workflow = self.load(components)
+            if selected:
+                # Get selected options
+                components = [self.options(component, workflow, x) for x, component in enumerate(selected)]
+
+        if selected:
+            # Process current action
+            self.process(workflow, components, "embeddings" in selected)
+
+            with st.sidebar:
+                # Generate export button after workflow is complete
+                _, config = self.yaml(components)
+                st.download_button("Export", config, file_name="workflow.yaml", help="Export the API workflow as YAML")
+        else:
+            st.info("Selected a workflow from the sidebar")
+
+if __name__ == "__main__":
+    os.environ["TOKENIZERS_PARALLELISM"] = "false"
+
+    try:
+        nltk.sent_tokenize("This is a test. Split")
+    except:
+        nltk.download("punkt")
+
+    # Create and run application
+    app = Application("workflows")
+    app.run()