| 6 | import json |
| 7 | |
| 8 | class lowCodeLLM: |
| 9 | def __init__(self, PLLM_temperature=0.4, ELLM_temperature=0): |
| 10 | self.PLLM = planningLLM(PLLM_temperature) |
| 11 | self.ELLM = executingLLM(ELLM_temperature) |
| 12 | |
| 13 | def get_workflow(self, task_prompt): |
| 14 | return self.PLLM.get_workflow(task_prompt) |
| 15 | |
| 16 | def extend_workflow(self, task_prompt, current_workflow, step=''): |
| 17 | ''' generate a sub-workflow for one of steps |
| 18 | - input: the current workflow, the step needs to extend |
| 19 | - output: sub-workflow ''' |
| 20 | workflow = self._json2txt(current_workflow) |
| 21 | return self.PLLM.extend_workflow(task_prompt, workflow, step) |
| 22 | |
| 23 | def execute(self, task_prompt,confirmed_workflow, history, curr_input): |
| 24 | ''' chat with the workflow-equipped low-code LLM ''' |
| 25 | prompt = [{'role': 'system', "content": 'The overall task you are facing is: '+task_prompt+ |
| 26 | '\nThe standard operating procedure(SOP) is:\n'+self._json2txt(confirmed_workflow)}] |
| 27 | history = prompt + history |
| 28 | response = self.ELLM.execute(curr_input, history) |
| 29 | return response |
| 30 | |
| 31 | def _json2txt(self, workflow_json): |
| 32 | ''' convert the json workflow to text''' |
| 33 | def json2text_step(step): |
| 34 | step_res = "" |
| 35 | step_res += step["stepId"] + ": [" + step["stepName"] + "]" |
| 36 | step_res += "[" + step["stepDescription"] + "][" |
| 37 | for jump in step["jumpLogic"]: |
| 38 | step_res += "[[" + jump["Condition"] + "][" + jump["Target"] + "]]," |
| 39 | step_res += "]\n" |
| 40 | return step_res |
| 41 | |
| 42 | workflow_txt = "" |
| 43 | for step in json.loads(workflow_json): |
| 44 | workflow_txt += json2text_step(step) |
| 45 | for substep in step['extension']: |
| 46 | workflow_txt += json2text_step(substep) |
| 47 | return workflow_txt |
no outgoing calls