| 361 | sim.makeAction([0], [direction*math.radians(360/len(direction_list))], [0]) |
| 362 | |
| 363 | class Decision_Testing_Experts: |
| 364 | def thought_fusion(self, predictions, thoughts): |
| 365 | matched_dict = dict() |
| 366 | for pred, thought in zip(predictions, thoughts): |
| 367 | if pred not in matched_dict.keys(): |
| 368 | matched_dict[pred] = [] |
| 369 | matched_dict[pred].append(thought) |
| 370 | |
| 371 | for key, value in matched_dict.items(): |
| 372 | multiple_thoughts = "; ".join(["Thought "+str(idx+1)+": "+thought for idx, thought in enumerate(value)]) |
| 373 | prompt = [ |
| 374 | {"role": "system", "content": "You are a thought fusion expert. Your task is to fuse given thought processes \ |
| 375 | into one thought. You need to reserve key information related to actions, landmarks, direction changes. You should only answer fused thought without other words."}, |
| 376 | {"role": "user", "content": f"Can you help me fuse the thoughts leading to the same movement direction? The thoughts are :{multiple_thoughts}, Fused thought: "} |
| 377 | ] |
| 378 | one_thought = gpt_response(prompt, "gpt-4", 1)[0]["message"]["content"] |
| 379 | logger.info(f"Pred: {key} Fused Thought: {one_thought}") |
| 380 | matched_dict[key] = one_thought |
| 381 | |
| 382 | return matched_dict |
| 383 | |
| 384 | def test_decisions(self, fused_pred_thought, observation, instruction): |
| 385 | if len(fused_pred_thought.keys()) == 1: |
| 386 | for key, value in fused_pred_thought.items(): |
| 387 | return key, value |
| 388 | else: |
| 389 | fused_pred_thought_ = "; ".join(["Navigation Viewpoint ID: "+key+" Thought: "+value for key, value in fused_pred_thought.items()]) |
| 390 | decision_prompt = [ |
| 391 | {"role": "system", "content": "You are a decision testing expert. Your task is to evaluate the feasibility of each movement \ |
| 392 | prediction based on thought process and environment. Then, you will make a final decision about navigation viewpoint ID without other words."}, |
| 393 | {"role": "user", "content": f"Can you help me make a final decision? The Observation: {observation}, Navigation Instruction: {instruction}, {fused_pred_thought_}, Final Decision: "} |
| 394 | ] |
| 395 | for _ in range(3): |
| 396 | next_vp = gpt_response(decision_prompt, "gpt-4", 1, temperature=0)[0]["message"]["content"].strip() |
| 397 | if len(next_vp) != 32: |
| 398 | logger.info(f"{next_vp} Length Problem") |
| 399 | continue |
| 400 | if next_vp not in fused_pred_thought.keys(): |
| 401 | logger.info(f"{next_vp} not in the candidates") |
| 402 | continue |
| 403 | break |
| 404 | |
| 405 | return next_vp, fused_pred_thought[next_vp] |
| 406 | |
| 407 | def eval_pred(final_vp, nav_history, shortest_distances, eval_cache, instruction, scanId, des_vp, gt_path): |
| 408 | path = [item["viewpoint"] for item in nav_history]+[final_vp] |