init

prompt/prompt.py

@@ -46,7 +46,30 @@ def handle_general_rules(data:dict):
 
 prompt_parse_flow = """
 # 角色
-请把图中的流程图转换为python代码
+你是一个专业的Python代码生成器，能够根据给定的流程图和函数内容，用Python语言生成完整的流程执行代码。
+
+给定的函数内容:
+{func}
+函数与节点对应关系:
+{node_func_map}
+请把图中的流程转换为python代码，流程中的各个节点参考给定的函数内容，并构造测试数据data，代码样式如下。
+
+from function1 import function1
+from function2 import function2
+from function3 import function3
+
+def main(data: dict):
+    流程逻辑
+    
+if __name__ == "__main__":
+    data = 测试数据
+    print(main(data))
+    
+返回结果要求：
+1、function1、function2、function3指代给定的函数，请替换为具体的函数名称。
+2、只返回函数代码，不要多余的输出。
+3、代码逻辑应严格按照图中的逻辑，不要自行添加多余的逻辑。
+4、代码的语法要符合python的语法规范，返回的代码应该是可执行的。
 """
 
 
@@ -54,5 +77,10 @@ def f_get_prompt_parse_node(rules_name: str, rules: str):
     return prompt_parse_node.replace("{rules_name}", rules_name).replace("{rules}", rules)
 
 
-def f_get_prompt_parse_flow():
-    return prompt_parse_flow
+def f_get_prompt_parse_flow(node_list:list):
+    func = ""
+    node_func_map = ""
+    for node_name, func_name, code in node_list:
+        func = f"{func}{code}\n\n"
+        node_func_map = f"{node_func_map}{node_name}: {func_name}\n"
+    return prompt_parse_flow.replace("{func}", func).replace("{node_func_map}", node_func_map)

strategy_parse.py

@@ -7,6 +7,7 @@
 import json
 import re
 import time
+from tqdm import tqdm
 
 import pandas as pd
 from PIL import Image
@@ -17,14 +18,15 @@ from enums import ResultCodesEnum
 from prompt import f_get_prompt_parse_node, f_get_prompt_parse_flow
 
 
-def _f_parse_flow(ws):
+def _f_parse_flow(ws, node_list: list):
     image = ws._images[0]
     img = Image.open(image.ref).convert("RGB")
     save_path = f_get_save_path("流程图.png")
     img.save(save_path)
     time.sleep(1)
     file_id = f_file_upload(save_path)
-    prompt = f_get_prompt_parse_flow()
+    prompt = f_get_prompt_parse_flow(node_list)
+    print(prompt)
     prompt = [
         {
             "type": "text",
@@ -36,15 +38,18 @@ def _f_parse_flow(ws):
         }
     ]
     prompt = json.dumps(prompt, ensure_ascii=False)
-    print(prompt)
     llm_answer = call_llm(prompt, "object_string")
     print(llm_answer)
+    code = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
     save_path = f_get_save_path("flow.py")
     with open(save_path, mode="w", encoding="utf8") as f:
-        f.write(llm_answer)
+        f.write(code)
 
+    save_path = f_get_save_path("__init__.py")
+    with open(save_path, mode="w", encoding="utf8") as f:
+        f.write("")
 
-def _f_parse_node(df: pd.DataFrame, sheet_name):
+def _f_parse_node(df: pd.DataFrame, node_name):
     rules = ""
     for idx, row in df.iterrows():
         var_name = row["变量"]
@@ -53,15 +58,16 @@ def _f_parse_node(df: pd.DataFrame, sheet_name):
         rule_out = row["输出"]
         rules = f"{rules}规则{idx + 1}: 变量:{var_name} 逻辑:{rule_content} 输出:{rule_out}\n"
 
-    prompt = f_get_prompt_parse_node(sheet_name, rules)
+    prompt = f_get_prompt_parse_node(node_name, rules)
     print(prompt)
     llm_answer = call_llm(prompt)
-    llm_answer = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
-    func_name = re.findall(r"def (.*)\(data", llm_answer)[0]
-    save_path = f_get_save_path(f"{func_name}.py", "node")
-    print(llm_answer)
+    code = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
+    func_name = re.findall(r"def (.*)\(data", code)[0]
+    save_path = f_get_save_path(f"{func_name}.py")
+    print(code)
     with open(save_path, mode="w", encoding="utf8") as f:
-        f.write(llm_answer)
+        f.write(code)
+    return func_name, code
 
 
 def f_parse_strategy(file_path):
@@ -70,12 +76,14 @@ def f_parse_strategy(file_path):
     sheet_names = excel.sheet_names
     if "流程图" not in sheet_names:
         GeneralException(ResultCodesEnum.NOT_FOUND, message=f"sheet【流程图】不存在")
-    for sheet_name in sheet_names:
-        if sheet_name == "流程图":
-            _f_parse_flow(wb["流程图"])
+    node_list = []
+    for node_name in tqdm(sheet_names):
+        if node_name == "流程图":
             continue
-        df = excel.parse(sheet_name=sheet_name)
-        _f_parse_node(df, sheet_name)
+        df = excel.parse(sheet_name=node_name)
+        func_name, code = _f_parse_node(df, node_name)
+        node_list.append((node_name, func_name, code))
+    _f_parse_flow(wb["流程图"], node_list)
     wb.close()
     excel.close()