单阶段：本地检索经验，调用模型完成应用选择和任务描述生成。
(task_description, use_graphrag=False, device_type="Android", use_experience=False)
| 887 | return None |
| 888 | |
| 889 | def get_app_package_name(task_description, use_graphrag=False, device_type="Android", use_experience=False): |
| 890 | """单阶段：本地检索经验，调用模型完成应用选择和任务描述生成。""" |
| 891 | current_file_path = Path(__file__).resolve() |
| 892 | current_dir = current_file_path.parent |
| 893 | default_template_path = current_dir.parent.parent / "utils" /"experience" / "templates-new.json" |
| 894 | logging.debug("Using template path: %s", default_template_path) |
| 895 | |
| 896 | # 本地检索经验 |
| 897 | search_engine = PromptTemplateSearch(default_template_path) |
| 898 | experience_content = search_engine.get_experience(task_description, 1) |
| 899 | logging.debug("检索到的相关经验:\n%s", experience_content) |
| 900 | if device_type == "Android": |
| 901 | planner_prompt_template = load_prompt("planner_oneshot.md") |
| 902 | elif device_type == "Harmony": |
| 903 | planner_prompt_template = load_prompt("planner_oneshot_harmony.md") |
| 904 | |
| 905 | # 检索用户偏好 |
| 906 | user_preferences = {} |
| 907 | if preference_extractor and preference_extractor.mem: |
| 908 | user_preferences = retrieve_user_preferences( |
| 909 | task_description, |
| 910 | preference_extractor.mem, |
| 911 | use_graphrag=use_graphrag |
| 912 | ) |
| 913 | if user_preferences: |
| 914 | print(f"检索到的用户偏好 (使用{'GraphRAG' if use_graphrag else '向量检索'}):\n{user_preferences}") |
| 915 | else: |
| 916 | print("未找到相关用户偏好") |
| 917 | # 结合上下文 |
| 918 | enhanced_context = combine_context(experience_content, user_preferences) |
| 919 | # 构建Prompt |
| 920 | prompt = planner_prompt_template.format( |
| 921 | task_description=task_description, |
| 922 | experience_content=enhanced_context |
| 923 | ) |
| 924 | response_str = planner_client.chat.completions.create( |
| 925 | model = planner_model, |
| 926 | messages=[ |
| 927 | { |
| 928 | "role": "user", |
| 929 | "content": [{"type": "text", "text": prompt}], |
| 930 | } |
| 931 | ], |
| 932 | ).choices[0].message.content |
| 933 | logging.info(f"Planner 响应: \n{response_str}") |
| 934 | response_json = parse_planner_response(response_str) |
| 935 | if response_json is None: |
| 936 | logging.error("无法解析模型响应为 JSON。") |
| 937 | logging.error(f"原始响应内容: {response_str}") |
| 938 | raise ValueError("无法解析模型响应为 JSON。") |
| 939 | app_name = response_json.get("app_name") |
| 940 | package_name = response_json.get("package_name") |
| 941 | final_desc = response_json.get("final_task_description", task_description) |
| 942 | return app_name, package_name, final_desc |
| 943 | |
| 944 | |
| 945 | def execute_single_task(task_description, device, data_dir, use_experience, use_graphrag, current_device_type, use_qwen3_model): |
no test coverage detected