New features

README.md

@@ -181,6 +181,23 @@ name: together-conf
 parallel: true
 ```
 
+You can update an existing config using the `pz update` command (also aliased as `pz uc`):
+```bash
+$ pz update --name default --settings parallel=true,pdfprocessor=pdfplumber
+Updated config: default
+
+$ pz config
+--- default ---
+filecachedir: /some/local/filepath
+llmservice: anthropic
+name: default
+parallel: true
+pdfprocessor: pdfplumber
+```
+
+The `--name` parameter specifies which config to update. `--settings` specifies all the parameter name and value pairs in the format `param_name=param_value`, separated by commas.
+
+
 Finally, you can delete a config with the `pz rm-config` command (also aliased as `pz rmc`):
 ```bash
 $ pz rmc --name together-conf

demos/askem-var.py

@@ -33,18 +33,26 @@ class Variable(Schema):
     value = Field(desc="The value of the variable, optional, set 'null' if not found")
 
 
+dict_of_excerpts = [
+    {"id": 0, "text": "ne of the few states producing detailed daily reports of COVID-19 confirmed cases, COVID-19 related cumulative hospitalizations, intensive care unit (ICU) admissions, and deaths per county. Likewise, Ohio is a state with marked variation of demographic and geographic attributes among counties along with substantial differences in the capacity of healthcare within the state. Our aim is to predict the spatiotemporal dynamics of the COVID-19 pandemic in relation with the distribution of the capacity of healthcare in Ohio. 2. Methods 2.1. Mathematical model We developed a spatial mathematical model to simulate the transmission dynamics of COVID-19 disease infection and spread. The spatially-explicit model incorporates geographic connectivity information at county level. The Susceptible-Infected-Hospitalized-Recovered- Dead (SIHRD) COVID-19 model classified the population into susceptibles (S), confirmed infections (I), hospitalized and ICU admitted (H), recovered (R) and dead (D). Based on a previous study that identified local air hubs and main roads as important geospatial attributes lio residing in the county. In the second scenario, we used the model to generate projections of the impact of potential easing on the non-pharmaceutical interventions in the critical care capacity of each county in Ohio. We assessed the impact of 50% reduction on the estimated impact of non-pharmaceutical interventions in reducing the hazard rate of infection. Under this scenario we calculated the proportion of ICU \n'"},
+    {"id": 1, "text": "t model incorporates geographic connectivity information at county level. The Susceptible-Infected-Hospitalized-Recovered- Dead (SIHRD) COVID-19 model classified the population into susceptibles (S), confirmed infections (I), hospitalized and ICU admitted (H), recovered (R) and dead (D). Based on a previous study that identified local air hubs and main roads as important geospatial attributes linked to differential COVID-19 related hospitalizations and mortality (Correa-Agudelo et a"}
+]
+
+list_of_strings = ["I have a variable a, the value is 1", "I have a variable b, the value is 2"]
+list_of_numbers = [1, 2, 3, 4, 5]
+
 if __name__ == "__main__":
     run_pz = True
     dataset = "askem"
+    file_path = "testdata/askem-tiny/"
 
     if run_pz:
         # reference, plan, stats = run_workload()
-        excerpts = Dataset(dataset, schema=TextFile)
+        df_input = pd.DataFrame(dict_of_excerpts)
+        excerpts = Dataset(df_input)
         output = excerpts.convert(
-            Variable, desc="A variable used or introduced in the paper snippet", cardinality=Cardinality.ONE_TO_MANY
-        )
-
-        engine = StreamingSequentialExecution
+            Variable, desc="A variable used or introduced in the context", cardinality=Cardinality.ONE_TO_MANY
+        ).filter("The value name is 'a'", depends_on="name")
         policy = MaxQuality()
         engine = StreamingSequentialExecution(
             policy=policy,
@@ -55,8 +63,7 @@ class Variable(Schema):
             allow_bonded_query=True,
         )
         engine.generate_plan(output, policy)
-
-        print(engine.plan)
+        print("Generated plan:\n", engine.plan)
         with st.container():
             st.write("### Executed plan: \n")
             # st.write(" " + str(plan).replace("\n", "  \n "))
@@ -66,9 +73,9 @@ class Variable(Schema):
                 st.write(strop)
 
         input_records = engine.get_input_records()
-        input_df = DataRecord.as_df(input_records)
+        input_df = DataRecord.as_df(input_records, fields_in_schema=True)
         print(input_df)
-        
+
         variables = []
         statistics = []
         start_time = time.time()
@@ -81,9 +88,9 @@ class Variable(Schema):
                 total_plan_time = time.time() - start_time
                 engine.plan_stats.finalize(total_plan_time)
 
-            record_time = time.time()
             statistics.append(engine.plan_stats)
-
+            intermediate_vars = DataRecord.as_df(vars, fields_in_schema=True)
+            print(intermediate_vars)
             for var in vars:
                 # ref.key = ref.first_author.split()[0] + ref.title.split()[0] + str(ref.year)
                 try:
@@ -119,8 +126,8 @@ class Variable(Schema):
                     st.write(" **value:** ", var.value, "\n")
 
         # write variables to a json file with readable format
-        with open(f"askem-variables-{dataset}.json", "w") as f:
-            json.dump(variables, f, indent=4)
+        # with open(f"askem-variables-{dataset}.json", "w") as f:
+        #     json.dump(variables, f, indent=4)
         vars_df = pd.DataFrame(variables)
 
     # G = nx.DiGraph()
@@ -150,7 +157,7 @@ class Variable(Schema):
     #
     # nx.write_gexf(G, "demos/bdf-usecase3.gexf")
 
-    print("References:", vars_df)
+    # print("References:", vars_df)
     # st.write(table.title, table.author, table.abstract)
     # endTime = time.time()
     # print("Elapsed time:", endTime - startTime)

demos/paper-demo.py

@@ -5,19 +5,15 @@
 
 import gradio as gr
 import numpy as np
-from palimpzest.constants import Cardinality, OptimizationStrategy
+from palimpzest.constants import Cardinality
 from palimpzest.core.data.datasources import UserSource
 from palimpzest.core.elements.records import DataRecord
 from palimpzest.core.lib.fields import BooleanField, Field, ImageFilepathField, ListField, NumericField, StringField
 from palimpzest.core.lib.schemas import Schema, Table, TextFile, XLSFile
 from palimpzest.datamanager.datamanager import DataDirectory
 from palimpzest.policy import MaxQuality, MinCost, MinTime
-from palimpzest.query import (
-    Execute,
-    NoSentinelPipelinedParallelExecution,
-    NoSentinelPipelinedSingleThreadExecution,
-    NoSentinelSequentialSingleThreadExecution,
-)
+from palimpzest.query.processor.config import QueryProcessorConfig
+from palimpzest.query.processor.query_processor_factory import QueryProcessorFactory
 from palimpzest.sets import Dataset
 from palimpzest.utils.udfs import xls_to_tables
 from PIL import Image
@@ -210,18 +206,6 @@ def get_item(self, idx: int):
         print("Policy not supported for this demo")
         exit(1)
 
-    execution_engine = None
-    executor = args.executor
-    if executor == "sequential":
-        execution_engine = NoSentinelSequentialSingleThreadExecution
-    elif executor == "pipelined":
-        execution_engine = NoSentinelPipelinedSingleThreadExecution
-    elif executor == "parallel":
-        execution_engine = NoSentinelPipelinedParallelExecution
-    else:
-        print("Executor not supported for this demo")
-        exit(1)
-
     if os.getenv("OPENAI_API_KEY") is None and os.getenv("TOGETHER_API_KEY") is None:
         print("WARNING: Both OPENAI_API_KEY and TOGETHER_API_KEY are unset")
 
@@ -261,15 +245,24 @@ def get_item(self, idx: int):
         plan = plan.filter("The rows of the table contain the patient age")
         plan = plan.convert(CaseData, desc="The patient data in the table", cardinality=Cardinality.ONE_TO_MANY)
 
-    # execute pz plan
-    records, execution_stats = Execute(
-        plan,
-        policy,
-        nocache=True,
-        optimization_strategy=OptimizationStrategy.PARETO,
-        execution_engine=execution_engine,
-        verbose=verbose,
-    )
+
+    config = QueryProcessorConfig(nocache=True, policy=policy, max_workers=10)
+    # # Option1: Create a basic processor
+    # # We could pass this process around to different service if needed.
+    # processor = QueryProcessorFactory.create_processor(
+    #     datasource=plan,
+    #     processing_strategy="no_sentinel",
+    #     execution_strategy="sequential",
+    #     optimizer_strategy="pareto", 
+    #     config=config
+    # )
+    # records, execution_stats = processor.execute()
+
+    # Option2: Use the new interface
+    records, execution_stats = plan.run(config, 
+                                        optimizer_strategy="pareto", 
+                                        execution_strategy="sequential", 
+                                        processing_strategy="no_sentinel")
 
     # save statistics
     if profile:

src/cli/cli_main.py

@@ -325,6 +325,53 @@ def set_config(name: str) -> None:
     _print_msg(f"Set config: {name}")
 
 
+@cli.command(aliases=["uc", "update"])
+@click.option("--name", type=str, default=None, required=True, help="Name of the config to update.")
+@click.option(
+    "--settings", 
+    type=str, 
+    required=True, 
+    help="Parameters to update in format 'param1=value1,param2=value2'. Example: 'llmservice=openai,parallel=true,pdfprocessor=pdfplumber'"
+)
+def update_config(name: str, settings: str) -> None:
+    """
+    Update multiple parameters in an existing Palimpzest config.
+
+    Parameters
+    ----------
+    name: str
+        Name of the config to update
+    params: str
+        Comma-separated list of parameter=value pairs to update
+    """
+    from palimpzest.config import Config
+    from palimpzest.constants import PZ_DIR
+
+    # check that config exists
+    if not os.path.exists(os.path.join(PZ_DIR, f"config_{name}.yaml")):
+        raise InvalidCommandError(f"Config with name {name} does not exist.")
+
+    # load the specified config
+    config = Config(name)
+
+    # Parse the params string into a dictionary
+    try:
+        param_pairs = settings.split(',')
+        updates = {}
+        for pair in param_pairs:
+            if pair.strip() == "":
+                continue
+            param, value = pair.split('=')
+            updates[param.strip()] = value.strip()
+    except Exception as e:
+        raise InvalidCommandError("Invalid params format. Use: param1=value1,param2=value2") from e
+
+    # Update each parameter
+    for param, value in updates.items():
+        config.set(param, value)
+
+    _print_msg(f"Updated config {name} with: {updates}")
+
 def main():
     """
     Entrypoint for Palimpzest CLI tool implemented using Click.
@@ -339,4 +386,5 @@ def main():
     cli.add_command(create_config)
     cli.add_command(rm_config)
     cli.add_command(set_config)
+    cli.add_command(update_config)
     cli()

src/palimpzest/__init__.py

@@ -1,4 +1,4 @@
-from palimpzest.constants import MAX_ROWS, Cardinality, OptimizationStrategy
+from palimpzest.constants import MAX_ROWS, Cardinality
 
 # Dataset functionality
 #from palimpzest.sets import Dataset
@@ -20,7 +20,6 @@
     # constants
     "MAX_ROWS",
     "Cardinality",
-    "OptimizationStrategy",
     # datamanager
     "DataDirectory",
     # policy

src/palimpzest/constants.py

@@ -36,18 +36,6 @@ class PromptStrategy(str, Enum):
     COT_MOA_AGG = "chain-of-thought-mixture-of-agents-aggregation"
 
 
-class OptimizationStrategy(str, Enum):
-    """
-    OptimizationStrategy determines which (set of) plan(s) the Optimizer
-    will return to the Execution layer.
-    """
-    GREEDY = "greedy"
-    CONFIDENCE_INTERVAL = "confidence-interval"
-    PARETO = "pareto" 
-    SENTINEL = "sentinel"
-    NONE = "none"
-
-
 class AggFunc(str, Enum):
     COUNT = "count"
     AVERAGE = "average"
@@ -81,6 +69,7 @@ class PickOutputStrategy(str, Enum):
 
 # character limit for various IDs
 MAX_ID_CHARS = 10
+MAX_DATASET_ID_CHARS = 16
 
 # retry LLM executions 2^x * (multiplier) for up to 10 seconds and at most 4 times
 RETRY_MULTIPLIER = 2

src/palimpzest/core/data/datasources.py

@@ -15,7 +15,16 @@
 
 from palimpzest import constants
 from palimpzest.core.elements.records import DataRecord
-from palimpzest.core.lib.schemas import File, ImageFile, Number, PDFFile, Schema, TextFile, WebPage, XLSFile
+from palimpzest.core.lib.schemas import (
+    DefaultSchema,
+    File,
+    ImageFile,
+    PDFFile,
+    Schema,
+    TextFile,
+    WebPage,
+    XLSFile,
+)
 from palimpzest.tools.pdfparser import get_text_from_pdf
 
 
@@ -141,10 +150,16 @@ def get_item(self, idx: int) -> DataRecord:
 class MemorySource(DataSource):
     """MemorySource returns multiple objects that reflect contents of an in-memory Python list"""
 
-    def __init__(self, vals: list[int | float], dataset_id: str):
-        # For the moment we assume that we are given a list of floats or ints, but someday it could be strings or something else
-        super().__init__(Number, dataset_id)
-        self.vals = vals
+    def __init__(self, vals: Any, dataset_id: str = "memory_input"):
+        if isinstance(vals, (str, int, float)):
+            self.vals = [vals]
+        elif isinstance(vals, tuple):
+            self.vals = list(vals)
+        else:
+            self.vals = vals
+
+        schema = Schema.from_df(self.vals) if isinstance(self.vals, pd.DataFrame) else DefaultSchema
+        super().__init__(schema, dataset_id)
 
     def copy(self):
         return MemorySource(self.vals, self.dataset_id)
@@ -156,9 +171,14 @@ def get_size(self):
         return sum([sys.getsizeof(self.get_item(idx)) for idx in range(len(self))])
 
     def get_item(self, idx: int) -> DataRecord:
-        value = self.vals[idx]
         dr = DataRecord(self.schema, source_id=idx)
-        dr.value = value
+
+        if isinstance(self.vals, pd.DataFrame):
+            row = self.vals.iloc[idx]
+            for field_name in row.index:
+                setattr(dr, field_name, row[field_name])
+        else:
+            dr.value = self.vals[idx]
 
         return dr