promptflow

1
import subprocess
2
from pathlib import Path
3
import hashlib
4
from jinja2 import Environment, FileSystemLoader, Template
5
from .telemetry_obj import Telemetry
6

7

8
class Step:
9
    """
10
    StepType in workflow
11
    """
12

13
    Environment = None
14

15
    @staticmethod
16
    def init_jinja_loader() -> Environment:
17
        jinja_folder_path = (
18
            Path(ReadmeStepsManage.git_base_dir())
19
            / "scripts"
20
            / "readme"
21
            / "ghactions_driver"
22
            / "workflow_steps"
23
        )
24
        Step.Environment = Environment(
25
            loader=FileSystemLoader(jinja_folder_path.resolve())
26
        )
27

28
    def __init__(self, name: str) -> None:
29
        self.workflow_name = name
30

31
    def get_workflow_step(self) -> str:
32
        # virtual method for override
33
        return ""
34

35
    @staticmethod
36
    def get_workflow_template(step_file_name: str) -> Template:
37
        # virtual method for override
38
        if Step.Environment is None:
39
            Step.init_jinja_loader()
40
        template = Step.Environment.get_template(step_file_name)
41
        return template
42

43

44
class AzureLoginStep(Step):
45
    def __init__(self) -> None:
46
        Step.__init__(self, "Azure Login")
47

48
    def get_workflow_step(self) -> str:
49
        template = Step.get_workflow_template("step_azure_login.yml.jinja2")
50
        return template.render(
51
            {
52
                "step_name": self.workflow_name,
53
            }
54
        )
55

56

57
class LoginAgainStep(Step):
58
    def __init__(self) -> None:
59
        Step.__init__(self, "Azure Login Again")
60

61
    def get_workflow_step(self) -> str:
62
        template = Step.get_workflow_template("step_login_again.yml.jinja2")
63
        return template.render(
64
            {
65
                "step_name": self.workflow_name,
66
            }
67
        )
68

69

70
class InstallDependenciesStep(Step):
71
    def __init__(self) -> None:
72
        Step.__init__(self, "Prepare requirements")
73

74
    def get_workflow_step(self) -> str:
75
        template = Step.get_workflow_template("step_install_deps.yml.jinja2")
76
        return template.render(
77
            {
78
                "step_name": self.workflow_name,
79
                "working_dir": ReadmeSteps.working_dir,
80
            }
81
        )
82

83

84
class InstallDevDependenciesStep(Step):
85
    def __init__(self) -> None:
86
        Step.__init__(self, "Prepare dev requirements")
87

88
    def get_workflow_step(self) -> str:
89
        template = Step.get_workflow_template("step_install_dev_deps.yml.jinja2")
90
        return template.render(
91
            {
92
                "step_name": self.workflow_name,
93
                "working_dir": ReadmeSteps.working_dir,
94
            }
95
        )
96

97

98
class CreateAoaiFromYaml(Step):
99
    def __init__(self, yaml_name: str) -> None:
100
        Step.__init__(self, "Create AOAI Connection from YAML")
101
        self.yaml_name = yaml_name
102

103
    def get_workflow_step(self) -> str:
104
        template = Step.get_workflow_template("step_yml_create_aoai.yml.jinja2")
105
        return template.render(
106
            {
107
                "step_name": self.workflow_name,
108
                "yaml_name": self.yaml_name,
109
            }
110
        )
111

112

113
class ExtractStepsAndRun(Step):
114
    def __init__(self) -> None:
115
        Step.__init__(self, f"Extract Steps {ReadmeSteps.readme_name}")
116

117
    def get_workflow_step(self) -> str:
118
        template = Step.get_workflow_template("step_extract_steps_and_run.yml.jinja2")
119
        return template.render(
120
            {
121
                "step_name": self.workflow_name,
122
                "working_dir": ReadmeSteps.working_dir,
123
                "readme_name": ReadmeSteps.readme_name,
124
            }
125
        )
126

127

128
class ExtractStepsAndRunGPTFour(Step):
129
    def __init__(self) -> None:
130
        Step.__init__(self, f"Extract Steps {ReadmeSteps.readme_name}")
131

132
    def get_workflow_step(self) -> str:
133
        template = Step.get_workflow_template(
134
            "step_extract_steps_and_run_gpt4.yml.jinja2"
135
        )
136
        return template.render(
137
            {
138
                "step_name": self.workflow_name,
139
                "working_dir": ReadmeSteps.working_dir,
140
                "readme_name": ReadmeSteps.readme_name,
141
            }
142
        )
143

144

145
class ExecuteCommand(Step):
146
    def __init__(self) -> None:
147
        Step.__init__(self, "Execute Command")
148

149
    def get_workflow_step(self) -> str:
150
        template = Step.get_workflow_template(
151
            "step_execute_command.yml.jinja2"
152
        )
153
        return template.render(
154
            {
155
                "step_name": self.workflow_name,
156
                "working_dir": ReadmeSteps.working_dir,
157
            }
158
        )
159

160

161
class CreateEnv(Step):
162
    def __init__(self) -> None:
163
        Step.__init__(self, "Refine .env file")
164

165
    def get_workflow_step(self) -> str:
166
        template = Step.get_workflow_template("step_create_env.yml.jinja2")
167
        content = template.render(
168
            {"step_name": self.workflow_name, "working_dir": ReadmeSteps.working_dir}
169
        )
170
        return content
171

172

173
class CreateEnvGPTFour(Step):
174
    def __init__(self) -> None:
175
        Step.__init__(self, "Refine .env file")
176

177
    def get_workflow_step(self) -> str:
178
        template = Step.get_workflow_template("step_create_env_gpt4.yml.jinja2")
179
        content = template.render(
180
            {"step_name": self.workflow_name, "working_dir": ReadmeSteps.working_dir}
181
        )
182
        return content
183

184

185
class CreateAoaiFromEnv(Step):
186
    def __init__(self, connection_name: str) -> None:
187
        Step.__init__(self, "Create AOAI Connection from ENV file")
188
        self.connection_name = connection_name
189

190
    def get_workflow_step(self) -> str:
191
        template = Step.get_workflow_template("step_env_create_aoai.yml.jinja2")
192
        content = template.render(
193
            {
194
                "step_name": self.workflow_name,
195
                "working_dir": ReadmeSteps.working_dir,
196
                "connection_name": self.connection_name,
197
            }
198
        )
199
        return content
200

201

202
class CreateRunYaml(Step):
203
    def __init__(self) -> None:
204
        Step.__init__(self, "Create run.yml")
205

206
    def get_workflow_step(self) -> str:
207
        template = Step.get_workflow_template("step_create_run_yml.yml.jinja2")
208
        content = template.render(
209
            {"step_name": self.workflow_name, "working_dir": ReadmeSteps.working_dir}
210
        )
211
        return content
212

213

214
class ReadmeSteps:
215
    """
216
    Static class to record steps, to be filled in workflow templates and Readme
217
    """
218

219
    step_array = []  # Record steps
220
    readme_name = ""  # Record readme name
221
    working_dir = ""  # the working directory of flow, relative to git_base_dir
222
    template = ""  # Select a base template under workflow_templates folder
223
    workflow = ""  # Target workflow name to be generated
224

225
    @staticmethod
226
    def remember_step(step: Step) -> Step:
227
        ReadmeSteps.step_array.append(step)
228
        return step
229

230
    @staticmethod
231
    def get_length() -> int:
232
        return len(ReadmeSteps.step_array)
233

234
    # region steps
235
    @staticmethod
236
    def create_env() -> Step:
237
        return ReadmeSteps.remember_step(CreateEnv())
238

239
    @staticmethod
240
    def create_env_gpt4() -> Step:
241
        return ReadmeSteps.remember_step(CreateEnvGPTFour())
242

243
    @staticmethod
244
    def yml_create_aoai(yaml_name: str) -> Step:
245
        return ReadmeSteps.remember_step(CreateAoaiFromYaml(yaml_name=yaml_name))
246

247
    @staticmethod
248
    def env_create_aoai(connection_name: str) -> Step:
249
        return ReadmeSteps.remember_step(
250
            CreateAoaiFromEnv(connection_name=connection_name)
251
        )
252

253
    @staticmethod
254
    def azure_login() -> Step:
255
        return ReadmeSteps.remember_step(AzureLoginStep())
256

257
    @staticmethod
258
    def login_again() -> Step:
259
        return ReadmeSteps.remember_step(LoginAgainStep())
260

261
    @staticmethod
262
    def install_dependencies() -> Step:
263
        return ReadmeSteps.remember_step(InstallDependenciesStep())
264

265
    @staticmethod
266
    def install_dev_dependencies() -> Step:
267
        return ReadmeSteps.remember_step(InstallDevDependenciesStep())
268

269
    @staticmethod
270
    def create_run_yaml() -> Step:
271
        return ReadmeSteps.remember_step(CreateRunYaml())
272

273
    @staticmethod
274
    def extract_steps_and_run() -> Step:
275
        return ReadmeSteps.remember_step(ExtractStepsAndRun())
276

277
    @staticmethod
278
    def extract_steps_and_run_gpt_four() -> Step:
279
        return ReadmeSteps.remember_step(ExtractStepsAndRunGPTFour())
280

281
    @staticmethod
282
    def execute_command() -> Step:
283
        return ReadmeSteps.remember_step(ExecuteCommand())
284

285
    # endregion steps
286

287
    @staticmethod
288
    def setup_target(
289
        working_dir: str, template: str, target: str, readme_name: str
290
    ) -> str:
291
        """
292
        Used at the very head of jinja template to indicate basic information
293
        """
294
        ReadmeSteps.working_dir = working_dir
295
        ReadmeSteps.template = template
296
        ReadmeSteps.workflow = target
297
        ReadmeSteps.step_array = []
298
        ReadmeSteps.readme_name = readme_name
299
        return ""
300

301
    @staticmethod
302
    def cleanup() -> None:
303
        ReadmeSteps.working_dir = ""
304
        ReadmeSteps.template = ""
305
        ReadmeSteps.workflow = ""
306
        ReadmeSteps.step_array = []
307

308

309
class ReadmeStepsManage:
310
    """
311
    # Static methods for manage all readme steps
312
    """
313

314
    repo_base_dir = ""
315

316
    @staticmethod
317
    def git_base_dir() -> str:
318
        """
319
        Get the base directory of the git repo
320
        """
321
        if ReadmeStepsManage.repo_base_dir == "":
322
            try:
323
                ReadmeStepsManage.repo_base_dir = (
324
                    subprocess.check_output(["git", "rev-parse", "--show-toplevel"])
325
                    .decode("utf-8")
326
                    .strip()
327
                )
328
                raise Exception("Not in git repo")
329
            except Exception:
330
                ReadmeStepsManage.repo_base_dir = Path(__file__).parent.parent.parent.parent.resolve()
331
                print(ReadmeStepsManage.repo_base_dir)
332
        return ReadmeStepsManage.repo_base_dir
333

334
    @staticmethod
335
    def write_workflow(
336
        workflow_name: str, pipeline_name: str, output_telemetry=Telemetry()
337
    ) -> None:
338
        # Schedule notebooks at different times to reduce maximum quota usage.
339
        name_hash = int(hashlib.sha512(workflow_name.encode()).hexdigest(), 16)
340
        schedule_minute = name_hash % 60
341
        schedule_hour = (name_hash // 60) % 4 + 19  # 19-22 UTC
342

343
        if "tutorials" in workflow_name:
344
            # markdown filename has some exceptions, special handle here
345
            if "chat_with_pdf" in workflow_name:
346
                readme_name = "chat-with-pdf.md"
347
            elif (
348
                "fine_tuning_evaluation_promptflow_quality_improvement" in workflow_name
349
            ):
350
                readme_name = "promptflow-quality-improvement.md"
351
            else:
352
                readme_name = "README.md"
353
            readme_path = (
354
                Path(ReadmeStepsManage.git_base_dir())
355
                / ReadmeSteps.working_dir
356
                / readme_name
357
            )
358
            # local import to avoid circular import
359
            from .resource_resolver import resolve_tutorial_resource
360

361
            path_filter = resolve_tutorial_resource(
362
                workflow_name, readme_path.resolve(), output_telemetry
363
            )
364
        else:
365
            if (
366
                "flow_with_additional_includes" in workflow_name
367
                or "flow_with_symlinks" in workflow_name
368
            ):
369
                # these two flows have dependencies on flow web-classification
370
                # so corresponding workflows should also listen to changes in web-classification
371
                path_filter = (
372
                    f"[ {ReadmeSteps.working_dir}/**, "
373
                    + "examples/*requirements.txt, "
374
                    + "examples/flows/standard/web-classification/**, "
375
                    + f".github/workflows/{workflow_name}.yml ]"
376
                )
377
            else:
378
                path_filter = (
379
                    f"[ {ReadmeSteps.working_dir}/**, "
380
                    + "examples/*requirements.txt, "
381
                    + f".github/workflows/{workflow_name}.yml ]"
382
                )
383
        replacements = {
384
            "steps": ReadmeSteps.step_array,
385
            "workflow_name": workflow_name,
386
            "ci_name": pipeline_name,
387
            "path_filter": path_filter,
388
            "crontab": f"{schedule_minute} {schedule_hour} * * *",
389
            "crontab_comment": f"Every day starting at {schedule_hour - 16}:{schedule_minute} BJT",
390
        }
391
        workflow_template_path = (
392
            Path(ReadmeStepsManage.git_base_dir())
393
            / "scripts"
394
            / "readme"
395
            / "ghactions_driver"
396
            / "workflow_templates"
397
        )
398
        target_path = (
399
            Path(ReadmeStepsManage.git_base_dir())
400
            / ".github"
401
            / "workflows"
402
            / f"{workflow_name}.yml"
403
        )
404
        template = Environment(
405
            loader=FileSystemLoader(workflow_template_path.resolve())
406
        ).get_template(ReadmeSteps.template)
407
        content = template.render(replacements)
408
        with open(target_path.resolve(), "w", encoding="utf-8") as f:
409
            f.write(content)
410
        print(f"Write readme workflow: {target_path.resolve()}")
411
        output_telemetry.workflow_name = workflow_name
412
        output_telemetry.target_path = target_path
413
        output_telemetry.readme_folder = ReadmeSteps.working_dir
414
        output_telemetry.readme_name = ReadmeSteps.readme_name
415
        output_telemetry.path_filter = path_filter
416