A workflow implementation that executes tasks in a sequential chain.
| 10 | |
| 11 | |
| 12 | class ChainWorkflow(Workflow): |
| 13 | """A workflow implementation that executes tasks in a sequential chain.""" |
| 14 | |
| 15 | WORKFLOW_NAME = 'ChainWorkflow' |
| 16 | |
| 17 | def build_workflow(self): |
| 18 | if not self.config: |
| 19 | return |
| 20 | |
| 21 | has_next = set() |
| 22 | start_task = None |
| 23 | for task_name, task_config in self.config.items(): |
| 24 | if 'next' in task_config: |
| 25 | next_tasks = task_config['next'] |
| 26 | if isinstance(next_tasks, str): |
| 27 | has_next.add(next_tasks) |
| 28 | else: |
| 29 | assert len( |
| 30 | next_tasks |
| 31 | ) == 1, 'ChainWorkflow only supports one next task' |
| 32 | has_next.update(next_tasks) |
| 33 | |
| 34 | for task_name in self.config.keys(): |
| 35 | if task_name not in has_next: |
| 36 | start_task = task_name |
| 37 | break |
| 38 | |
| 39 | if start_task is None: |
| 40 | raise ValueError('No start task found') |
| 41 | |
| 42 | result = [] |
| 43 | current_task = start_task |
| 44 | |
| 45 | while current_task: |
| 46 | result.append(current_task) |
| 47 | next_task = None |
| 48 | task_config = self.config[current_task] |
| 49 | if 'next' in task_config: |
| 50 | next_tasks = task_config['next'] |
| 51 | if isinstance(next_tasks, str): |
| 52 | next_task = next_tasks |
| 53 | else: |
| 54 | next_task = next_tasks[0] |
| 55 | |
| 56 | current_task = next_task |
| 57 | self.workflow_chains = result |
| 58 | |
| 59 | async def run(self, inputs, **kwargs): |
| 60 | """ |
| 61 | Execute the chain of tasks sequentially. |
| 62 | |
| 63 | For each task in the built workflow chain: |
| 64 | - Determine the agent type and instantiate it. |
| 65 | - Run the agent with the provided inputs. |
| 66 | - Pass the result as input to the next agent. |
| 67 | |
| 68 | Args: |
| 69 | inputs (Any): Initial input data for the first task in the chain. |