-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor multi agents execution (#376)
* refactor(t2viz): introduced pipeline to run async operations Signed-off-by: Yulong Ruan <[email protected]> * add tests Signed-off-by: Yulong Ruan <[email protected]> * update changelog entry Signed-off-by: Yulong Ruan <[email protected]> * rename Operator -> Task Signed-off-by: Yulong Ruan <[email protected]> * cleanup console.log Signed-off-by: Yulong Ruan <[email protected]> --------- Signed-off-by: Yulong Ruan <[email protected]> (cherry picked from commit fb47ce8) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> # Conflicts: # CHANGELOG.md
- Loading branch information
1 parent
cfc975c
commit 30996ac
Showing
8 changed files
with
346 additions
and
186 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| import { Pipeline } from './pipeline'; | ||
|
|
||
| describe('pipeline', () => { | ||
| it('should run pipeline', (done) => { | ||
| const fn1 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn1')); | ||
| }); | ||
| const fn2 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn2')); | ||
| }); | ||
| const pipeline = new Pipeline([{ execute: fn1 }, { execute: fn2 }]); | ||
| pipeline.getResult$().subscribe((result) => { | ||
| expect(result).toEqual(['input', 'fn1', 'fn2']); | ||
| expect(pipeline.status$.value).toBe('STOPPED'); | ||
| done(); | ||
| }); | ||
|
|
||
| expect(pipeline.status$.value).toBe('STOPPED'); | ||
| pipeline.run('input'); | ||
| expect(pipeline.status$.value).toBe('RUNNING'); | ||
| }); | ||
|
|
||
| it('should run pipeline with the latest input', (done) => { | ||
| const fn1 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn1')); | ||
| }); | ||
| const fn2 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn2')); | ||
| }); | ||
| const pipeline = new Pipeline([{ execute: fn1 }, { execute: fn2 }]); | ||
| pipeline.getResult$().subscribe((result) => { | ||
| expect(result).toEqual(['input2', 'fn1', 'fn2']); | ||
| expect(fn1).toHaveBeenCalledTimes(2); | ||
| // The fn2 should only be called onece because the first pipeline run should already be canceled | ||
| expect(fn2).toHaveBeenCalledTimes(1); | ||
| done(); | ||
| }); | ||
| // the pipeline run twice with different inputs | ||
| // the second pipeline.run should be make it to cancel the first pipeline run | ||
| pipeline.run('input1'); | ||
| pipeline.run('input2'); | ||
| }); | ||
|
|
||
| it('should run pipeline once synchronously', async () => { | ||
| const fn1 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn1')); | ||
| }); | ||
| const fn2 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn2')); | ||
| }); | ||
| const pipeline = new Pipeline([{ execute: fn1 }, { execute: fn2 }]); | ||
| const result = await pipeline.runOnce('input'); | ||
| expect(result).toEqual(['input', 'fn1', 'fn2']); | ||
| }); | ||
|
|
||
| it('should catch error', (done) => { | ||
| const fn1 = jest.fn().mockImplementation((input) => { | ||
| return Promise.resolve(Array<string>().concat(input).concat('fn1')); | ||
| }); | ||
| const fn2 = jest.fn().mockImplementation(() => { | ||
| throw new Error('test'); | ||
| }); | ||
|
|
||
| const pipeline = new Pipeline([{ execute: fn1 }, { execute: fn2 }]); | ||
| pipeline.getResult$().subscribe((result) => { | ||
| expect(result).toEqual({ error: new Error('test') }); | ||
| done(); | ||
| }); | ||
| pipeline.run('input'); | ||
| }); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| /* eslint-disable @typescript-eslint/no-explicit-any */ | ||
|
|
||
| import { BehaviorSubject, Observable, Subject, of } from 'rxjs'; | ||
| import { switchMap, tap, catchError } from 'rxjs/operators'; | ||
|
|
||
| import { Task } from './task'; | ||
|
|
||
| export class Pipeline { | ||
| input$ = new Subject<any>(); | ||
| output$: Observable<any>; | ||
| status$ = new BehaviorSubject<'RUNNING' | 'STOPPED'>('STOPPED'); | ||
|
|
||
| constructor(private readonly tasks: Array<Task<any, any>>) { | ||
| this.output$ = this.input$ | ||
| .pipe(tap(() => this.status$.next('RUNNING'))) | ||
| .pipe( | ||
| switchMap((value) => { | ||
| return this.tasks | ||
| .reduce((acc$, task) => { | ||
| return acc$.pipe(switchMap((result) => task.execute(result))); | ||
| }, of(value)) | ||
| .pipe(catchError((e) => of({ error: e }))); | ||
| }) | ||
| ) | ||
| .pipe(tap(() => this.status$.next('STOPPED'))); | ||
| } | ||
|
|
||
| /** | ||
| * Triggers the pipeline execution by emitting a new input value. | ||
| * This will start the processing of the provided input value through the pipeline's tasks, | ||
| * with each task transforming the input in sequence. The resulting value will be emitted | ||
| * through the `output$` observable. | ||
| */ | ||
| run(input: any) { | ||
| this.input$.next(input); | ||
| } | ||
|
|
||
| /** | ||
| * Synchronously processes the provided input value through the pipeline's tasks in sequence. | ||
| * This method bypasses the reactive pipeline and executes each task one by one, | ||
| * it suitable for use cases where you need a one-time, imperative-style execution. | ||
| */ | ||
| async runOnce(input: any) { | ||
| let nextInput = input; | ||
| for (const task of this.tasks) { | ||
| nextInput = await task.execute(nextInput); | ||
| } | ||
| return nextInput; | ||
| } | ||
|
|
||
| getResult$() { | ||
| return this.output$; | ||
| } | ||
| } |
Oops, something went wrong.