Convert incoming messages to vLLM format and inject DeepAnalyze template: - Always wrap user message with "# Instruction" heading - Optionally append workspace file info under "# Data"
(
messages: List[Dict[str, Any]],
workspace_dir: str,
)
| 89 | |
| 90 | |
| 91 | def prepare_vllm_messages( |
| 92 | messages: List[Dict[str, Any]], |
| 93 | workspace_dir: str, |
| 94 | ) -> List[Dict[str, str]]: |
| 95 | """ |
| 96 | Convert incoming messages to vLLM format and inject DeepAnalyze template: |
| 97 | - Always wrap user message with "# Instruction" heading |
| 98 | - Optionally append workspace file info under "# Data" |
| 99 | """ |
| 100 | vllm_messages: List[Dict[str, str]] = [] |
| 101 | for msg in messages: |
| 102 | role = msg.get("role") if isinstance(msg, dict) else None |
| 103 | raw_content = msg.get("content") if isinstance(msg, dict) else None |
| 104 | content = _normalize_openai_message_content(raw_content) |
| 105 | if role: |
| 106 | vllm_messages.append({"role": role, "content": content}) |
| 107 | |
| 108 | # Locate last user message |
| 109 | last_user_idx: Optional[int] = None |
| 110 | for idx in range(len(vllm_messages) - 1, -1, -1): |
| 111 | if vllm_messages[idx].get("role") == "user": |
| 112 | last_user_idx = idx |
| 113 | break |
| 114 | |
| 115 | workspace_file_info = collect_file_info(workspace_dir) |
| 116 | |
| 117 | if last_user_idx is not None: |
| 118 | user_content = str(vllm_messages[last_user_idx].get("content", "")).strip() |
| 119 | instruction_body = user_content if user_content else "# Instruction" |
| 120 | if workspace_file_info: |
| 121 | vllm_messages[last_user_idx]["content"] = ( |
| 122 | f"# Instruction\n{instruction_body}\n\n# Data\n{workspace_file_info}" |
| 123 | ) |
| 124 | else: |
| 125 | vllm_messages[last_user_idx]["content"] = f"# Instruction\n{instruction_body}" |
| 126 | |
| 127 | return vllm_messages |
| 128 | |
| 129 | |
| 130 | def execute_code_safe( |
no test coverage detected