mirror of
https://github.com/neuml/txtai.git
synced 2026-10-11 21:39:55 +00:00
Preserve workflow and task configuration dictionaries (#1377)
This commit is contained in:
@@ -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"]
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user