{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaL4","dataSources":[{"sourceId":84795,"databundleVersionId":11281725,"sourceType":"competition"},{"sourceId":221096520,"sourceType":"kernelVersion"},{"sourceId":236932,"sourceType":"modelInstanceVersion","modelInstanceId":202348,"modelId":224071},{"sourceId":256568,"sourceType":"modelInstanceVersion","modelInstanceId":204085,"modelId":225262},{"sourceId":256580,"sourceType":"modelInstanceVersion","modelInstanceId":204048,"modelId":225262},{"sourceId":256581,"sourceType":"modelInstanceVersion","modelInstanceId":204059,"modelId":225262},{"sourceId":257257,"sourceType":"modelInstanceVersion","modelInstanceId":210022,"modelId":231665},{"sourceId":276458,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":236741,"modelId":224053}],"dockerImageVersionId":30887,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true},"papermill":{"default_parameters":{},"duration":22.341371,"end_time":"2024-12-11T03:22:13.479076","environment_variables":{},"exception":null,"input_path":"__notebook__.ipynb","output_path":"__notebook__.ipynb","parameters":{},"start_time":"2024-12-11T03:21:51.137705","version":"2.6.0"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\n# https://www.kaggle.com/competitions/ai-mathematical-olympiad-progress-prize-2/discussion/560682#3113134\nos.environ[\"TRITON_PTXAS_PATH\"] = \"/usr/local/cuda/bin/ptxas\"","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-02-17T17:08:54.229896Z","iopub.execute_input":"2025-02-17T17:08:54.230211Z","iopub.status.idle":"2025-02-17T17:08:54.233493Z","shell.execute_reply.started":"2025-02-17T17:08:54.230162Z","shell.execute_reply":"2025-02-17T17:08:54.232826Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import io\nimport time\nimport shutil\n\nimport pandas as pd\nimport polars as pl\n\nimport kaggle_evaluation.konwinski_prize_inference_server\nfrom typing import List, Tuple, Dict, Optional\n\nstart_time = time.time()\nallowed_time = [start_time + 60 * 60]","metadata":{"trusted":true,"_kg_hide-output":true,"execution":{"iopub.status.busy":"2025-02-17T17:08:54.234426Z","iopub.execute_input":"2025-02-17T17:08:54.234616Z","iopub.status.idle":"2025-02-17T17:09:06.329934Z","shell.execute_reply.started":"2025-02-17T17:08:54.234599Z","shell.execute_reply":"2025-02-17T17:09:06.329281Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"The evaluation API requires that you set up a server which will respond to inference requests. We have already defined the server; you just need write the predict function. When we evaluate your submission on the hidden test set the client defined in `konwinski_prize_gateway` will run in a different container with direct access to the hidden test set and hand off the data.\n\nYour code will always have access to the published copies of the files.","metadata":{"papermill":{"duration":0.002032,"end_time":"2024-12-11T03:22:08.823897","exception":false,"start_time":"2024-12-11T03:22:08.821865","status":"completed"},"tags":[]}},{"cell_type":"code","source":"instance_count: Optional[int] = None\n\n\ndef get_number_of_instances(num_instances: int) -> None:\n    \"\"\"The very first message from the gateway will be the total number of instances to be served.\n    You don't need to edit this function.\n    \"\"\"\n    global instance_count\n    instance_count = num_instances","metadata":{"papermill":{"duration":0.011949,"end_time":"2024-12-11T03:22:08.838279","exception":false,"start_time":"2024-12-11T03:22:08.82633","status":"completed"},"tags":[],"trusted":true,"execution":{"iopub.status.busy":"2025-02-17T17:09:06.331176Z","iopub.execute_input":"2025-02-17T17:09:06.331769Z","iopub.status.idle":"2025-02-17T17:09:06.334942Z","shell.execute_reply.started":"2025-02-17T17:09:06.331746Z","shell.execute_reply":"2025-02-17T17:09:06.334365Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Initialize LLM","metadata":{}},{"cell_type":"code","source":"from vllm import LLM, SamplingParams, RequestOutput\nimport warnings\n\nwarnings.simplefilter(\"ignore\")\n\nos.environ[\"CUDA_VISIBLE_DEVICES\"] = \"0,1,2,3\"\nos.environ[\"TOKENIZERS_PARALLELISM\"] = \"false\"\n\nif os.getenv(\"KAGGLE_KERNEL_RUN_TYPE\") or os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\"):\n    llm_model_pth: str = (\n        \"/kaggle/input/m/huikang/deepseek-r1/transformers/qwen-qwq-32b-awq/1\"\n    )\nelse:\n    llm_model_pth: str = \"/root/volume/KirillR/QwQ-32B-Preview-AWQ\"\n\nBATCH_SIZE: int = 7\nVALIDATION_COPY_COUNT: int = 3\nMAX_TOKENS: int = 4096\n\nMAX_NUM_SEQS: int = 7\nMAX_MODEL_LEN: int = 32_768\n\nllm: LLM = LLM(\n    llm_model_pth,\n    dtype=\"float16\",\n    max_num_seqs=MAX_NUM_SEQS,  # Maximum number of sequences per iteration. Default is 256\n    max_model_len=MAX_MODEL_LEN,  # Model context length\n    trust_remote_code=True,  # Trust remote code (e.g., from HuggingFace) when downloading the model and tokenizer\n    tensor_parallel_size=4,  # The number of GPUs to use for distributed execution with tensor parallelism\n    gpu_memory_utilization=0.95,  # The ratio (between 0 and 1) of GPU memory to reserve for the model\n    seed=2024,\n)","metadata":{"trusted":true,"_kg_hide-output":true,"execution":{"iopub.status.busy":"2025-02-17T17:09:06.336281Z","iopub.execute_input":"2025-02-17T17:09:06.336584Z","iopub.status.idle":"2025-02-17T17:15:03.224846Z","shell.execute_reply.started":"2025-02-17T17:09:06.336562Z","shell.execute_reply":"2025-02-17T17:15:03.222847Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"tokenizer = llm.get_tokenizer()\n\n\ndef count_tokens(text: str) -> int:\n    return len(tokenizer.encode(text))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-02-17T17:15:03.226961Z","iopub.status.idle":"2025-02-17T17:15:03.227236Z","shell.execute_reply":"2025-02-17T17:15:03.227115Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Helper functions","metadata":{}},{"cell_type":"code","source":"import os\n\n\ndef stringify_directory(directory: str) -> str:\n    full_paths: List[str] = []\n\n    for root, dirs, files in os.walk(directory):\n        for file in files:\n            full_path: str = os.path.join(root, file)\n            full_paths.append(full_path)\n    return \"\\n\".join(full_paths)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-02-17T17:15:03.227793Z","execution_failed":"2025-02-17T17:15:03.478Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import re\n\n\ndef extract_file_query(xml_content: str) -> Dict[str, List[str]]:\n    import xml.etree.ElementTree as ET\n\n    # Prepare a data structure to collect results\n    parsed_data: Dict[str, List[str]] = {}\n    pattern: str = r\"<root>(.*?)</root>\"\n    matches: List[str] = re.findall(pattern, xml_content, re.DOTALL)\n\n    for match in matches:\n        try:\n            # Parse the XML\n            root = ET.fromstring(\"<root>\" + match + \"</root>\")\n\n            # Find all <entry> elements\n            for entry in root.findall(\"entry\"):\n                # Extract the <filepath> text\n                filepath = entry.find(\"filepath\")\n                filepath_text: Optional[str] = (\n                    filepath.text.strip()\n                    if filepath is not None and filepath.text is not None\n                    else None\n                )\n\n                # Locate <strings_to_search> container\n                strings_container = entry.find(\"strings_to_search\")\n\n                # Gather each <string_to_search> text\n                search_strings: List[str] = []\n                if strings_container is not None:\n                    for s in strings_container.findall(\"string_to_search\"):\n                        if s.text is not None:\n                            search_strings.append(s.text.strip())\n\n                # Store in a dictionary: { filepath: [search_strings...] }\n                parsed_data[filepath_text] = search_strings  # type: ignore\n        except:\n            print(\"Error parsing output\")\n            print(xml_content)\n            return {}\n\n    return parsed_data","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.478Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"reading_prompt: str = (\n    \"\"\"\nYou will be implementing a git diff patch to solve an issue with the code repository.\nYou will first need to select files in the file directory.\n\nThis is the problem statement.\n\n{problem_statement}\n\nThis is the file directory\n\n<directory>\n{directory_string}\n</directory>\n\nWhich files should be inspected so that we can solve the problem?\nWhen we inspect each file, what strings should be searched?\n\nReturn the strings to search in this format\n\n(explanation)\n\n<root>\n    <entry>\n        <filepath>filepath</filepath>  \n        <strings_to_search>\n            <string_to_search>string_to_search</string_to_search>\n            ...\n            <string_to_search>string_to_search</string_to_search>\n        </strings_to_search>\n    </entry>\n    <entry>\n        <filepath>filepath</filepath>\n        <strings_to_search>\n            <string_to_search>string_to_search</string_to_search>\n            ...\n            <string_to_search>string_to_search</string_to_search>\n        </strings_to_search>\n    </entry>\n    ...\n</root>\n...\n\nNotes:\n- Make sure to encode each entry between <root> and </root>\n- Return the FULL filepath - exactly as specified in <directory> and </directory>\n    - Example: <filepath>repo/path/to/directory/file.py</filepath>\n- If you are searching for a word instead of a substring, maybe add spaces or brackets before and after the string\n    - For example, if you are searching for uses of the function `calculate`, use ` calculate(` as the search string instead of `calculate`\n- Prefer searching longer strings\n    - Avoid searching for strings that might appear in many parts of the codebase\n- Search the test files as well to understand the feature behavior\n    - Also search for the relevant function calls in the test files\n\"\"\".strip()\n)\n\n\ndef get_selection_query(\n    directory_string: str, problem_statement: str\n) -> Tuple[List[str], List[Dict[str, List[str]]]]:\n    sampling_params: SamplingParams = SamplingParams(\n        temperature=0.6,  # randomness of the sampling\n        min_p=0.01,\n        skip_special_tokens=True,  # Whether to skip special tokens in the output\n        max_tokens=MAX_TOKENS,\n    )\n\n    list_of_messages: List[List[Dict[str, str]]] = [\n        [\n            {\n                \"role\": \"user\",\n                \"content\": reading_prompt.format(\n                    problem_statement=problem_statement[:20_000],\n                    directory_string=directory_string[:30_000],\n                ),\n            },\n        ]\n        for _ in range(BATCH_SIZE)\n    ]\n\n    prompt_texts: List[str] = [\n        (\n            tokenizer.apply_chat_template(\n                conversation=messages, tokenize=False, add_generation_prompt=True\n            )  # type: ignore\n        )\n        + \"<think>\\n\"\n        for messages in list_of_messages\n    ]\n    # print(prompt_texts)\n\n    print(\"get_selection_query\", [count_tokens(text) for text in prompt_texts])\n    request_outputs: list[RequestOutput] = llm.generate(\n        prompt_texts, sampling_params=sampling_params\n    )\n    if not request_outputs:\n        return [], []\n    response_texts: List[str] = [\n        request_output.outputs[0].text for request_output in request_outputs\n    ]\n    print(\"get_selection_query\", [count_tokens(text) for text in response_texts])\n\n    completion_texts = [\n        prompt_text + response_text\n        for prompt_text, response_text in zip(prompt_texts, response_texts)\n    ]\n    file_queries: List[Dict[str, List[str]]] = [\n        extract_file_query(response_text) for response_text in response_texts\n    ]\n    return completion_texts, file_queries","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.478Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def setup(\n    repo_archive: io.BytesIO,\n    pip_packages_archive: io.BytesIO,\n    env_setup_cmds_templates: list[str],\n    repo_path: str,\n) -> None:\n    \"\"\"Replace this function with your inference code.\n    Args:\n        problem_statement: The text of the git issue.\n        repo_path: A BytesIO buffer path with a .tar containing the codebase that must be patched. The gateway will make this directory available immediately before this function runs.\n        pip_packages_archive: A BytesIO buffer path with a .tar containing the wheel files necessary for running unit tests.\n        env_setup_cmds_templates: Commands necessary for installing the pip_packages_archive.\n    \"\"\"\n\n    # Unpack the codebase to be patched into a directory that won't be exported when\n    # the notebook is saved.\n    archive_path = \"/tmp/repo_archive.tar\"\n    with open(archive_path, \"wb\") as f:\n        f.write(repo_archive.read())\n    if os.path.exists(repo_path):\n        shutil.rmtree(repo_path)\n    shutil.unpack_archive(archive_path, extract_dir=repo_path)\n    os.remove(archive_path)\n\n    \"\"\"\n    Unpack pip_packages if you want to run unit tests on your patch.\n    Note that editing unit tests with your patch -- even to add valid tests -- can cause your submission to be flagged as a failure.\n    Most of the relevant repos use pytest for running tests. You will almost certainly need to run only a subset of the unit tests to avoid running out of inference time.\n    \"\"\"\n    pip_archive_dir = \"/tmp/pip_packages_archive.tar\"\n    with open(pip_archive_dir, \"wb\") as f:\n        f.write(pip_packages_archive.read())\n    pip_packages_path = \"/path/to/pip_packages\"\n    if os.path.exists(pip_packages_path):\n        shutil.rmtree(pip_packages_path)\n    shutil.unpack_archive(pip_archive_dir, extract_dir=pip_packages_path)\n    os.remove(pip_archive_dir)\n\n    # Get env setup cmds by setting the pip_packages_path\n    env_setup_cmds = [\n        cmd.format(pip_packages_path=pip_packages_path)\n        for cmd in env_setup_cmds_templates\n    ]\n\n    # Run env setup for the repo\n    subprocess.run(\n        \"\\n\".join(env_setup_cmds),\n        shell=True,\n        executable=\"/bin/bash\",\n        cwd=repo_path,\n    )","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"REPO_PATH: str = \"repo\"\n\n\ndef fetch_file_contents(\n    files_to_search: Dict[str, List[str]], context_lines: int = 12, max_gap: int = 0\n) -> str:\n    from io import StringIO\n    from typing import Tuple\n\n    def find_lines_in_files_with_context(\n        search_map: Dict[str, List[str]], context_lines: int = context_lines\n    ) -> List[List[List[Tuple[int, str]]]]:\n        \"\"\"\n        Given a dictionary mapping file paths to a list of search terms,\n        open each file and gather *snippets* of lines that contain any\n        of those search terms, including 'context_lines' before and after.\n\n        Returns a list of lists:\n        [\n          [  # For file1\n             [ (line_number, text), (line_number, text), ... ],\n             [ ... ],\n          ],\n          [  # For file2\n             ...\n          ],\n          ...\n        ]\n        \"\"\"\n        all_matches_per_file: List[List[List[Tuple[int, str]]]] = []\n\n        for path, terms in search_map.items():\n            if not os.path.isfile(path):\n                # If the file is not found, record an empty list\n                all_matches_per_file.append([])\n                continue\n\n            with open(path, \"r\", encoding=\"utf-8\", errors=\"replace\") as f:\n                lines = f.readlines()\n\n            file_snippets: List[List[Tuple[int, str]]] = []\n            num_lines: int = len(lines)\n\n            for i, line in enumerate(lines, start=1):\n                if any(t in line for t in terms):\n                    start_idx: int = max(1, i - context_lines)\n                    end_idx: int = min(num_lines, i + context_lines)\n                    snippet: List[Tuple[int, str]] = []\n                    for snippet_no in range(start_idx, end_idx + 1):\n                        text_content: str = lines[snippet_no - 1].rstrip(\"\\n\")\n                        snippet.append((snippet_no, text_content))\n                    file_snippets.append(snippet)\n\n            all_matches_per_file.append(file_snippets)\n\n        return all_matches_per_file\n\n    # ---------------------------------------------------------\n    # 3. MERGE OVERLAPPING/ADJACENT SNIPPETS\n    # ---------------------------------------------------------\n\n    def merge_file_snippets(\n        file_snippets: List[List[Tuple[int, str]]], gap: int = 0\n    ) -> List[List[Tuple[int, str]]]:\n        \"\"\"\n        Merge overlapping or nearly adjacent snippets in a single file’s snippet list.\n        \"\"\"\n        intervals: List[Tuple[int, int, List[Tuple[int, str]]]] = []\n        for snippet in file_snippets:\n            if snippet:\n                start_line: int = snippet[0][0]\n                end_line: int = snippet[-1][0]\n                intervals.append((start_line, end_line, snippet))\n\n        intervals.sort(key=lambda x: x[0])  # sort by start line\n\n        merged: List[Tuple[int, int, List[Tuple[int, str]]]] = []\n        for start, end, snippet in intervals:\n            if not merged:\n                merged.append((start, end, snippet))\n                continue\n\n            prev_start, prev_end, prev_snippet = merged[-1]\n            if start <= prev_end + gap:\n                new_end: int = max(end, prev_end)\n                combined_dict: Dict[int, str] = {}\n                for ln, txt in prev_snippet:\n                    combined_dict[ln] = txt\n                for ln, txt in snippet:\n                    combined_dict[ln] = txt\n                merged_snippet: List[Tuple[int, str]] = [\n                    (ln, combined_dict[ln]) for ln in sorted(combined_dict)\n                ]\n                merged[-1] = (prev_start, new_end, merged_snippet)\n            else:\n                merged.append((start, end, snippet))\n\n        # Extract just the merged snippet portion\n        return [x[2] for x in merged]\n\n    def merge_all_snippets(\n        all_files_snips: List[List[List[Tuple[int, str]]]], gap: int = 0\n    ) -> List[List[List[Tuple[int, str]]]]:\n        \"\"\"\n        Merge snippet blocks within each file.\n        all_files_snips is a list-of-lists:\n          [\n            [ snippetA, snippetB, ... ],  # file 1\n            [ snippetC, snippetD, ... ],  # file 2\n          ]\n        \"\"\"\n        merged: List[List[List[Tuple[int, str]]]] = []\n        for snips in all_files_snips:\n            merged.append(merge_file_snippets(snips, gap=gap))\n        return merged\n\n    # ---------------------------------------------------------\n    # 4. RUN LOGIC: generate files, search, merge, and BUILD A STRING\n    # ---------------------------------------------------------\n\n    has_any_matches: bool = False\n\n    # 1) Gather snippets around each match\n    context_snippets: List[List[List[Tuple[int, str]]]] = (\n        find_lines_in_files_with_context(files_to_search, context_lines=context_lines)\n    )\n\n    # 2) Merge overlapping snippets\n    merged_snips: List[List[List[Tuple[int, str]]]] = merge_all_snippets(\n        context_snippets, gap=max_gap\n    )\n\n    # 3) Build a string (instead of printing)\n    output = StringIO()\n\n    # Header\n    output.write(\"Sample files created successfully.\\n\\n\")\n    output.write(\"Search Results (by file, merging any overlapping context):\\n\\n\")\n\n    # For each file\n    for (filepath, terms), snippet_list in zip(files_to_search.items(), merged_snips):\n        output.write(f\"[file name]: {filepath[len(REPO_PATH) + 1:]}\\n\")\n        terms_searched_as_str = \"\\n\".join(terms)\n        output.write(f\"[terms searched]:\\n{terms_searched_as_str}\\n\")\n        output.write(\"[file content begin]\\n\")\n        if not snippet_list:\n            output.write(\"  No matches found.\\n\")\n        else:\n            has_any_matches = True\n            for snippet_idx, snippet in enumerate(snippet_list, start=1):\n                snippet_start: int = snippet[0][0]\n                snippet_end: int = snippet[-1][0]\n                output.write(\n                    f\"\\nMatch #{snippet_idx}, lines {snippet_start} to {snippet_end}:\\n\"\n                )\n                for line_no, text in snippet:\n                    output.write(f\"  {line_no:3d} | {text}\\n\")\n                output.write(\"\\n\")\n        output.write(\"[file content end]\\n\\n\")\n\n    file_content_string: str = output.getvalue()\n\n    if has_any_matches:\n        return file_content_string\n    return \"\"","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.478Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import re\n\ndef extract_patch_string(text: str) -> Optional[str]:\n    pattern: str = r\"\\n```diff\\n(.*?)\\n```\"\n    matches: List[str] = re.findall(pattern, text, re.DOTALL)\n    if not matches:\n        return None\n    return matches[-1] + \"\\n\"","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.478Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"patching_prompt: str = (\n    \"\"\"\nYou will be implementing a git diff patch to solve an issue with the code repository.\nThis is the problem statement.\n\n{problem_statement}\n\nThese are the files that is thought to be relevant\n\n{file_content_string}\n\nWrite a git diff within ```diff and ``` that fully fixes the problem.\nThe git diff should not cause other tests to fail.\n\nExample:\n\n```diff\n--- a/first.txt\n+++ b/first.txt\n@@ -1,3 +1,3 @@\n start\n-first change\n+new first change\n middle\n@@ -7,4 +7,4 @@\n some content\n-second change\n+new second change\n more content\n--- a/second.txt\n+++ b/second.txt\n@@ -1,3 +1,3 @@\n beginning\n-old line\n+new line\n end\n```\n\nReminder\n- Put your diff within ```diff and ``` and make sure the diff is valid.\n- Only the last diff printed will be considered.\n\"\"\".strip()\n)\n\nimport re\n\n\ndef get_patch_string(\n    problem_statement: str, file_content_strings: List[str]\n) -> Tuple[List[str], List[Optional[str]]]:\n    sampling_params: SamplingParams = SamplingParams(\n        temperature=0.7,  # randomness of the sampling\n        min_p=0.01,\n        skip_special_tokens=True,  # Whether to skip special tokens in the output\n        max_tokens=MAX_TOKENS,\n    )\n\n    inference_idx_to_input_idx: list[int] = [\n        input_idx\n        for input_idx, file_content_string in enumerate(file_content_strings)\n        if file_content_string != \"\"\n    ]\n\n    list_of_messages: List[List[Dict[str, str]]] = [\n        [\n            {\n                \"role\": \"user\",\n                \"content\": patching_prompt.format(\n                    problem_statement=problem_statement[:20_000],\n                    file_content_string=file_content_strings[input_idx][:30_000],\n                ),\n            },\n        ]\n        for input_idx in inference_idx_to_input_idx\n    ]\n\n    prompt_texts: List[str] = [\n        (\n            tokenizer.apply_chat_template(\n                conversation=messages, tokenize=False, add_generation_prompt=True\n            )  # type: ignore\n        )\n        + \"<think>\\n\"\n        for messages in list_of_messages\n    ]\n    # print(prompt_texts)\n\n    print(\"get_patch_string\", [count_tokens(text) for text in prompt_texts])\n    request_outputs: list[RequestOutput] = llm.generate(\n        prompt_texts, sampling_params=sampling_params\n    )\n    response_texts_from_inference: List[str] = [\n        request_output.outputs[0].text for request_output in request_outputs\n    ]\n    print(\n        \"get_patch_string\",\n        [count_tokens(text) for text in response_texts_from_inference],\n    )\n    completion_texts_from_inference = [\n        prompt_text + response_text\n        for prompt_text, response_text in zip(\n            prompt_texts, response_texts_from_inference\n        )\n    ]\n    patch_strings_from_inference: List[Optional[str]] = [\n        extract_patch_string(response_text)\n        for response_text in response_texts_from_inference\n    ]\n\n    completion_texts: list[str] = [\"\" for _ in file_content_strings]\n    patch_strings: List[Optional[str]] = [None for _ in file_content_strings]\n    for inference_idx, (completion_text, patch_string) in enumerate(\n        zip(completion_texts_from_inference, patch_strings_from_inference)\n    ):\n        input_idx = inference_idx_to_input_idx[inference_idx]\n        completion_texts[input_idx] = completion_text\n        patch_strings[input_idx] = patch_string\n\n    return completion_texts, patch_strings","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pathlib import Path\n\nverifying_prompt: str = (\n    \"\"\"\nThis is the problem statement.\n\n{problem_statement}\n\nThese are the files that is thought to be relevant, which may not be complete.\n\n{file_content_string}\n\nThis is the proposed patch to fix the problem.\n\n{patch_string}\n\nEvaluate whether the patch works\n- The patch fully fixes the problem described in the problem statement.\n- The patch does not cause side effects and make any other tests fail.\n\nEnd your response with exactly either of\n- <label>Yes</label>, this fixes the problem.\n- <label>No</label>, this does not fix the problem.\n\nReminder\n- Only evaluate, do not provide suggestion on how to fix.\n- Remember to write exactly either of <label>Yes</label> or <label>No</label> in the last line\n\"\"\".strip()\n)\n\n\ndef is_valid_patch_format(patch_string: str) -> bool:\n    \"\"\"\n    A quick check to confirm if a patch could be valid.\n    \"\"\"\n    if not(isinstance(patch_string, str)):\n        return False\n    try:\n        patch_set = unidiff.PatchSet(patch_string)\n        if len(patch_set) == 0:\n            return False\n    except Exception:\n        return False\n    return True\n\n\ndef patch_dry_run_succeeds(patch_string: str, repo_path: str = REPO_PATH, timeout: int = 60) -> bool:\n    \"\"\"\n    A robust check if the patch will proceed without any errors.\n    Should be run after `is_valid_patch_format()`: the patch\n    command can hang if the inputs are sufficiently invalid.\n\n    Args:\n        patch_path: Path to a file containing the patch.\n        repo_path: Path to the directory to be patched.\n        timeout: Number of seconds before the dry run will be cancelled.\n    \"\"\"\n    with open(\"patch.txt\", \"w\") as f:\n        f.write(patch_string)\n    patch_path = \"/kaggle/working/patch.txt\"\n\n    cmd = f\"patch --quiet --dry-run -p1 -i {patch_path} -d {repo_path}\"\n    try:\n        subprocess.run(cmd, shell=True, check=True, timeout=timeout)\n        return True\n    except subprocess.CalledProcessError:\n        return False\n\n\ndef get_verification(\n    problem_statement: str,\n    file_content_strings: List[str],\n    patch_strings: List[Optional[str]],\n    repo_path: str,\n) -> Tuple[List[List[str]], List[List[bool]]]:\n    assert len(file_content_strings) == len(patch_strings)\n    sampling_params: SamplingParams = SamplingParams(\n        temperature=0.3,  # randomness of the sampling\n        min_p=0.01,\n        skip_special_tokens=True,  # Whether to skip special tokens in the output\n        max_tokens=MAX_TOKENS,\n    )\n\n    inference_idx_to_input_idx: list[int] = [\n        input_idx\n        for _ in range(VALIDATION_COPY_COUNT)\n        for input_idx, patch_string in enumerate(patch_strings)\n        if patch_string is not None and is_valid_patch_format(patch_string) # and patch_dry_run_succeeds(patch_string, repo_path)\n    ]\n    print(inference_idx_to_input_idx)\n\n    list_of_messages: List[List[Dict[str, str]]] = [\n        [\n            {\n                \"role\": \"user\",\n                \"content\": verifying_prompt.format(\n                    problem_statement=problem_statement[:20_000],\n                    file_content_string=file_content_strings[input_idx][:30_000],\n                    patch_string=patch_strings[input_idx],\n                ),\n            },\n        ]\n        for input_idx in inference_idx_to_input_idx\n    ]\n\n    prompt_texts: List[str] = [\n        (\n            tokenizer.apply_chat_template(\n                conversation=messages, tokenize=False, add_generation_prompt=True\n            )  # type: ignore\n        )\n        + \"<think>\\n\"\n        for messages in list_of_messages\n    ]\n    # print(prompt_texts)\n\n    print(\"get_verification\", [count_tokens(text) for text in prompt_texts])\n    request_outputs: list[RequestOutput] = llm.generate(\n        prompt_texts, sampling_params=sampling_params\n    )\n    response_texts: List[str] = [\n        request_output.outputs[0].text for request_output in request_outputs\n    ]\n    print(\"get_verification\", [count_tokens(text) for text in response_texts])\n\n    completion_texts = [\n        prompt_text + response_text\n        for prompt_text, response_text in zip(prompt_texts, response_texts)\n    ]\n    judgments_flattened: List[bool] = [\n        \"<label>Yes</label>\" in response_text for response_text in response_texts\n    ]\n    print(judgments_flattened)\n\n    judgments_aggregated: List[List[bool]] = [[] for _ in file_content_strings]\n    completion_text_aggregated: List[List[str]] = [[] for _ in patch_strings]\n    for inference_idx, (completion_text, judgement) in enumerate(\n        zip(completion_texts, judgments_flattened)\n    ):\n        input_idx = inference_idx_to_input_idx[inference_idx]\n        completion_text_aggregated[input_idx].append(completion_text)\n        judgments_aggregated[input_idx].append(judgement)\n    print(judgments_aggregated)\n\n    return completion_text_aggregated, judgments_aggregated","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import unidiff\nimport subprocess\n\n\ndef calculate_patch_score(\n    patch_string: Optional[str],\n    judgments: List[bool],\n    repo_path: str,\n    validation_copy_count: int = VALIDATION_COPY_COUNT\n) -> float:\n    \n    score = 0.0\n\n    if patch_string is None:\n        return -3.0  \n        \n    if not is_valid_patch_format(patch_string):\n        return -2.0  \n\n    if not patch_dry_run_succeeds(patch_string, repo_path):\n        return -1.0  \n\n    judgment_weight = 2.0\n    valid_format_bonus = 1.0\n    dry_run_bonus = 2.0\n    size_penalty_factor = 0.3\n    file_count_penalty = 0.0\n    \n    if judgments.count(True) == 0: # Heavily penalize no \"Yes\" votes\n        return -100.0\n        \n    score += judgments.count(True) ** 2 * judgment_weight\n\n    score += valid_format_bonus\n    score += dry_run_bonus\n\n    #  Non-linear penalty for patch size - inverse function, penalize very long patches more\n    score -= patch_lines_penalty(patch_string, size_penalty_factor) # Using separate function\n\n    patch_set = unidiff.PatchSet(patch_string)\n    num_files_changed = len(patch_set)\n    score -= num_files_changed * file_count_penalty\n\n    return score\n\ndef patch_lines_penalty(patch_string: str, penalty_factor: float) -> float:\n    \"\"\"Calculates a non-linear penalty based on the number of lines in the patch.\n    Uses an inverse function to penalize very long patches more heavily.\n    \"\"\"\n    patch_lines = patch_string.strip().count('\\n') + 1 # Add 1 to avoid division by zero\n    if patch_lines <= 15: # Reduced penalty for very short patches\n        return patch_lines * penalty_factor * 0.5 # Less penalty for short patches\n    return  (100 / patch_lines) * penalty_factor # Inverse penalty - longer patches penalized more sharply\n    \ndef choose_patch_string_optimized(\n    patch_strings: list[Optional[str]], judgments_aggregated: List[List[bool]], repo_path: str, correction_threshold: float, validation_copy_count: int = VALIDATION_COPY_COUNT\n) -> tuple[list[int], Optional[str]]:\n    best_score = -float('inf')\n    best_patch_string = None\n    scores = []\n    \n    all_patches_negative_score: bool = True \n    correction_threshold = 0.0 # Threshold for considering a patch as potentially correct\n    \n    for patch_string, judgments in zip(patch_strings, judgments_aggregated):\n        score = calculate_patch_score(patch_string, judgments, repo_path, validation_copy_count)\n        scores.append(score)\n        \n        if score >= correction_threshold: \n            all_patches_negative_score = False \n            if score > best_score:\n                best_score = score\n                best_patch_string = patch_string\n\n    if all_patches_negative_score or best_score < 1: \n        return scores, None\n    else:\n        return scores, best_patch_string \n\n\ndef ensemble_patch_selection(patch_strings_list, judgments_aggregated_list):\n    best_patch = None\n    max_yes_votes = -1\n\n    for patch_string, judgments in zip(patch_strings_list, judgments_aggregated_list):\n        yes_votes = judgments.count(True)\n        if yes_votes > max_yes_votes:\n            max_yes_votes = yes_votes\n            best_patch = patch_string\n\n    return best_patch","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Predict function","metadata":{}},{"cell_type":"code","source":"# Predict function\ndef predict_inner(problem_statement: str, directory: str) -> Optional[str]:\n    directory_string = stringify_directory(directory)\n\n    selection_completion_texts, file_queries = get_selection_query(\n        directory_string, problem_statement\n    )\n\n    file_content_strings: List[str] = [\n        fetch_file_contents(file_query) for file_query in file_queries\n    ]\n\n    patch_completion_texts, patch_strings = get_patch_string(\n        problem_statement, file_content_strings\n    )\n\n    verification_completion_texts_aggregated, judgments_aggregated = get_verification(\n        problem_statement, file_content_strings, patch_strings, directory\n    )\n\n    scores, patch_string = choose_patch_string_optimized(\n        patch_strings, judgments_aggregated, directory, VALIDATION_COPY_COUNT\n    )\n\n    if not os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\"):\n        data = {\n            \"problem_statement\": [problem_statement] * len(file_queries),\n            \"selection_completion_text\": selection_completion_texts,\n            \"selection_completion_length\": [\n                count_tokens(completion_text)\n                for completion_text in selection_completion_texts\n            ],\n            \"file_query\": file_queries,\n            \"file_content_string\": file_content_strings,\n            \"patch_completion_text\": patch_completion_texts,\n            \"patch_completion_length\": [\n                count_tokens(completion_text)\n                for completion_text in patch_completion_texts\n            ],\n            \"patch_string\": patch_strings, \n        }\n\n        for copy_idx in range(VALIDATION_COPY_COUNT):\n            data[f\"verification_completion_text_{copy_idx}\"] = [\n                completion_texts[copy_idx] if completion_texts else None\n                for completion_texts in verification_completion_texts_aggregated\n            ]\n            data[f\"verification_completion_length_{copy_idx}\"] = [\n                count_tokens(completion_texts[copy_idx]) if completion_texts else None\n                for completion_texts in verification_completion_texts_aggregated\n            ]\n            data[f\"judgment_{copy_idx}\"] = [\n                judgments[copy_idx] if judgments else None\n                for judgments in judgments_aggregated\n            ]\n\n        data[\"judgment_count_true\"] = [judgments.count(True) for judgments in judgments_aggregated] \n        data[\"score\"] = scores\n\n        elapsed_time_int = int(time.time() - start_time)  \n\n        pd.DataFrame(data).to_csv(  \n            f\"{str(elapsed_time_int).zfill(5)}.csv\", index=False  \n        )\n\n    return patch_string","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import io\nfrom typing import Optional, List\n\nskip_prediction: bool = False\n\n\ndef predict(\n    problem_statement: str,\n    repo_archive: io.BytesIO,\n    pip_packages_archive: io.BytesIO,\n    env_setup_cmds_templates: List[str],\n) -> Optional[str]:\n    \"\"\"Replace this function with your inference code.\n    Args:\n        problem_statement: The text of the git issue.\n        repo_archive: A BytesIO buffer path with a .tar containing the codebase that must be patched. The gateway will make this directory available immediately before this function runs.\n    \"\"\"\n    global skip_prediction\n    if skip_prediction:\n        return None\n\n    with open(\"repo_archive.tar\", \"wb\") as f:\n        f.write(repo_archive.read())\n    repo_path: str = REPO_PATH\n    if os.path.exists(repo_path):\n        shutil.rmtree(repo_path)\n    shutil.unpack_archive(\"repo_archive.tar\", extract_dir=repo_path)\n    os.remove(\"repo_archive.tar\")\n\n    patch_string: Optional[str] = None\n    patch_string = predict_inner(\n        problem_statement=problem_statement, directory=repo_path\n    )\n    shutil.rmtree(repo_path)\n\n    if not os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\"):\n        skip_prediction = True\n\n    print(\"submitted patch_string\")\n    print(patch_string)\n\n    if patch_string is None:\n        return None\n\n    return patch_string","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Get predict data without server","metadata":{}},{"cell_type":"code","source":"import os\nimport zipfile\n\n# !mkdir -p /kaggle/tmp/konwinski-prize-alt\nos.makedirs(\"/kaggle/tmp/konwinski-prize-alt\", exist_ok=True)\n\n# !unzip -q -o /kaggle/input/konwinski-prize/data.a_zip -d /kaggle/tmp/konwinski-prize-alt/ 2>/dev/null || true\ntry:\n    with zipfile.ZipFile(\"/kaggle/input/konwinski-prize/data.a_zip\", \"r\") as zip_ref:\n        zip_ref.extractall(\"/kaggle/tmp/konwinski-prize-alt/\")\nexcept:\n    pass","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\n\n\ndef get_problem(problem_index: int) -> Tuple[str, str, io.BytesIO]:\n    df = pd.read_parquet(\"/kaggle/tmp/konwinski-prize-alt/data/data.parquet\")\n\n    problem_statement: str = df[\"problem_statement\"][problem_index]\n    repo_path: str = (\n        f\"/kaggle/tmp/konwinski-prize-alt/data/repos/repo__{df['instance_id'][problem_index]}\"\n    )\n\n    import shutil\n    import tempfile\n\n    with tempfile.TemporaryDirectory() as tmpdir:\n        shutil.make_archive(os.path.join(tmpdir, \"a_repo\"), \"tar\", repo_path)\n        with open(os.path.join(tmpdir, \"a_repo.tar\"), \"rb\") as f:\n            repo_archive = io.BytesIO(f.read())\n\n    return problem_statement, repo_path, repo_archive","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"demo_problem_index: int = 0\n\nif os.getenv(\"KAGGLE_KERNEL_RUN_TYPE\") == \"Interactive\" and not os.getenv(\n    \"KAGGLE_IS_COMPETITION_RERUN\"\n):\n    problem_statement, repo_path, repo_archive = get_problem(\n        problem_index=demo_problem_index\n    )\n\n    print(repo_path)\n    print(problem_statement)\n    print(len(list(repo_archive)))\n    print(len(list(repo_archive)))\n","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if os.getenv(\"KAGGLE_KERNEL_RUN_TYPE\") == \"Interactive\" and not os.getenv(\n    \"KAGGLE_IS_COMPETITION_RERUN\"\n):\n    skip_prediction = False\n    problem_statement, repo_path, repo_archive = get_problem(\n        problem_index=demo_problem_index\n    )\n    patch_string = predict(problem_statement, repo_archive, io.BytesIO(), [])","metadata":{"trusted":true,"_kg_hide-output":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if (\n    os.getenv(\"KAGGLE_KERNEL_RUN_TYPE\") == \"Interactive\"\n    and not os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\")\n    and patch_string is not None\n):\n    import polars as pl\n\n    df = pl.read_parquet(\"/kaggle/tmp/konwinski-prize-alt/data/data.parquet\")\n\n    import kaggle_evaluation.konwinski_prize_gateway\n\n    k_prize_gateway = kaggle_evaluation.konwinski_prize_gateway.KPrizeGateway()\n    k_prize_gateway.unpack_data_paths()\n\n    results = k_prize_gateway._evaluate_instance(\n        instance=df.row(demo_problem_index, named=True),\n        patch=patch_string,\n    )\n\n    from collections import Counter\n    print(\n        demo_problem_index, Counter(result.unit_test_outcome for result in results[1:])\n    )","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if (\n    os.getenv(\"KAGGLE_KERNEL_RUN_TYPE\") == \"Interactive\"\n    and not os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\")\n    and patch_string is not None\n):\n    from kaggle_evaluation.konwinski_prize_gateway import UnitTestOutcome\n\n    for result in results[1:]:\n        if result.unit_test_outcome != UnitTestOutcome.PASSED:\n            print(result.test_name)\n            print(result.fail_description)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Evaluation with inference server","metadata":{}},{"cell_type":"code","source":"skip_prediction = False","metadata":{"trusted":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"inference_server = (\n    kaggle_evaluation.konwinski_prize_inference_server.KPrizeInferenceServer(\n        get_number_of_instances, predict\n    )\n)\n\nif os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\"):\n    inference_server.serve()\nelse:\n    inference_server.run_local_gateway(\n        data_paths=(\n            \"/kaggle/input/konwinski-prize/\",  # Path to the entire competition dataset\n            \"/kaggle/tmp/konwinski-prize/\",  # Path to a scratch directory for unpacking data.a_zip.\n        )  # type: ignore\n    )\n","metadata":{"trusted":true,"_kg_hide-output":true,"execution":{"execution_failed":"2025-02-17T17:15:03.479Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def calculate_score(\n    n_correct: int,\n    n_wrong: int,\n    n_skipped: int,\n    incorrect_score=-1,\n    skip_score=-10**-4\n) -> str:\n    score = (n_correct + n_wrong * incorrect_score + n_skipped * skip_score) / (n_correct + n_skipped + n_wrong)\n    if n_correct == 0:\n        score = incorrect_score\n    return f\"{score:+.20f}\"[:1+1+1+6].lstrip(\"+\")\n\n\ndef calculate_results(score: str) -> list[tuple[int, int, int]]:\n    assert type(score) == str\n    possible_results = []\n    for n_correct in range(72):\n        for n_wrong in range(72 - n_correct):\n            n_skipped = 71 - n_correct - n_wrong\n            if score == calculate_score(n_correct, n_wrong, n_skipped):\n                 possible_results.append((n_correct, n_wrong, n_skipped))\n    return possible_results\n\n\nfor score in [\n    \"-0.309916\",\n    \"-0.295836\",\n    \"-0.126839\",\n    \"-0.253590\",\n    \"-0.112753\",\n    \"-0.197254\",\n    \"-0.873249\",\n    \"-0.971830\",\n    \"1.000000\",\n]:\n    print(score, calculate_results(score), \"\\n\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-07T03:16:32.534825Z","iopub.execute_input":"2025-03-07T03:16:32.535059Z","iopub.status.idle":"2025-03-07T03:16:32.608663Z","shell.execute_reply.started":"2025-03-07T03:16:32.535036Z","shell.execute_reply":"2025-03-07T03:16:32.607418Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}