Skip to content

[DRAFT] Pipeline Statefulness for Open World Challenges #277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ygefen wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[DRAFT] Pipeline Statefulness for Open World Challenges #277

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8464205
Adds working_state history to pipeline ADM, with reset_history hook
ygefen May 21, 2026
8b898e3
Revert to using hasattr for backwards compatibility failover
ygefen May 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions align_system/algorithms/abstracts.py
View file
Open in desktop
Copy link
Copy Markdown
Contributor Author

@ygefen ygefen May 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addes a reset_history default NOOP for backwards compatibility and so that we do not have to check hasattr in the code.
In fact I would push to do this same pattern for output_conflict_resolver to avoid the hasattr check for that too.

Copy link
Copy Markdown
Contributor

@dmjoy dmjoy May 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I agree with you that doing the same thing with output_conflict_resolver would tidy some things up; however the way it's set up currently it's configurable at the hydra config level, but this change would only make it configurable at the code level (I think)? FWIW we don't use this pattern very often but here's one config where it's used: https://github.com/ITM-Kitware/align-system/blob/main/align_system/configs/adm/phase2_pipeline_fewshot_comparative_regression_swap_average.yaml#L53

Copy link
Copy Markdown
Contributor Author

@ygefen ygefen May 21, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cool, so in the ABC it would look like:

def output_conflict_resolver(*args, **kwargs):
    return *args, **kwargs

i.e. a Callable type noop, and the hydra in your example would override it in the subclass with the injected callable.

Copy link
Copy Markdown
Contributor

@dmjoy dmjoy May 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, can you override members like that in the hydra config? I was under the impression that you can only set / pass initialization args with hydra instantiation. If the config I linked works without modification with this ABC update you're suggesting then I'm in favor of this change.

Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ def choose_action(self,
**kwargs) -> Union[Action, tuple[Action, dict]]:
pass

def reset_history(self):
pass

class StructuredInferenceEngine(ABC):
@abstractmethod
Expand All @@ -38,3 +40,6 @@ def run_returns(self) -> Union[str, Iterable[str]]:
returns expect from the `run` method
'''
pass

def reset_history(self):
pass
19 changes: 14 additions & 5 deletions align_system/algorithms/pipeline_adm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from collections.abc import Iterable
from collections.abc import Iterable, deque
from timeit import default_timer as timer

from align_system.algorithms.abstracts import ActionBasedADM, ADMComponent
Expand All @@ -8,8 +8,9 @@


class PipelineADM(ActionBasedADM):
def __init__(self, steps: list[ADMComponent]):
def __init__(self, steps: list[ADMComponent], history_window=None):
self.steps = steps
self.history = deque(maxlen=history_window)
Comment thread
ygefen marked this conversation as resolved.

def choose_action(self,
scenario_state,
Expand All @@ -21,15 +22,17 @@ def choose_action(self,
'actions': available_actions,
'alignment_target': alignment_target,
**kwargs}


per_step_timing_stats = []

for i, step in enumerate(self.steps):
step_returns = step.run_returns()

start_time = timer()
# Run the step
run_output = call_with_coerced_args(step.run, working_output)
# Run the step, temporarily adding historical working outputs to working_output
working_output_with_history = {**{"history": list(self.history)}, **working_output}
Comment thread
ygefen marked this conversation as resolved.
run_output = call_with_coerced_args(step.run, working_output_with_history)
end_time = timer()

per_step_timing_stats.append(
Expand Down Expand Up @@ -73,5 +76,11 @@ def choose_action(self,

working_output.setdefault('choice_info', {})['per_step_timing_stats'] =\
per_step_timing_stats

self.history.append(working_output)
return working_output['chosen_action'], working_output

def reset_history(self):
self.history.clear()
for step in self.steps:
if hasattr(step, 'reset_history'):
step.reset_history()