Add Codeforces Flows.

.gitignore

@@ -0,0 +1,149 @@
+### macOS ###
+# General
+.DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit tests / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+docs/build/
+docs/docs/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# dotenv
+.env
+
+# virtualenv
+.venv
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+
+# datafiles
+.xml
+.pkl
+
+# misc
+.idea/
+.iml
+.dropbox
+
+# media files
+.png
+.jpg
+.pdf
+
+
+# auto-generated by flows, all synced modules will be ignored by default
+FLOW_MODULE_ID

CF_Code.py

@@ -0,0 +1,6 @@
+from flows.application_flows import OpenAIChatAtomicFlow
+
+
+class CF_Code(OpenAIChatAtomicFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

CF_Code.yaml

@@ -0,0 +1,82 @@
+name: "Code_Flow"
+description: |2- 
+  Given a problem description, generate code directly.
+
+# ~~~ Input interface specification ~~~
+input_interface_non_initialized:  # Applied when constructing the first user message.
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+
+input_interface_initialized:  # Applied when constructing all subsequent user messages.
+  - "query"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "api_output"
+
+# ~~~ Flow specification ~~~
+model_name: "gpt-4"
+
+generation_parameters:
+  n: 1
+  max_tokens: 2000
+  temperature: 0.3
+
+  model_kwargs:
+    top_p: 0.2
+    frequency_penalty: 0
+    presence_penalty: 0
+
+system_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    Your goal is to provide executable Python code that solves a competitive programming problem. The code should correctly handle all corner cases in order to pass the hidden test cases, which are used to evaluate the correctness of the solution.
+
+    The user will specify the problem by providing you with:
+      - the problem statement
+      - input description
+      - output description
+      - example test cases
+      - (optional) explanation of the test cases
+
+    The user will provide you with a task and an output format that you will strictly follow.
+  input_variables: []
+  template_format: jinja2
+
+human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: "{{query}}"
+  input_variables:
+    - "query"
+  template_format: jinja2
+
+init_human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    # Problem statement
+    {{problem_description}}
+
+    # Input description
+    {{input_description}}
+
+    # Output description
+    {{output_description}}
+
+    {{io_examples_and_explanation}}
+
+
+    The input should be read from the standard input and the output should be passed to the standard output.
+    Return Python code that solves the problem. Reply in the following format:
+    ```python
+    {{code_placeholder}}
+    ```
+  input_variables:
+    - "problem_description"
+    - "input_description"
+    - "output_description"
+    - "io_examples_and_explanation"
+  partial_variables:
+    code_placeholder: "{{python_code}}"
+  template_format: jinja2

CF_CodeCollab.py

@@ -0,0 +1,6 @@
+from flows.base_flows import GeneratorCriticFlow
+
+
+class CF_CodeCollab(GeneratorCriticFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

CF_CodeCollab.yaml

@@ -0,0 +1,93 @@
+name: "CodeCollab_Flow"
+description: |2-
+  Given a problem description, alternate between a step in which code is generated, and a step in which the produced code is evaluated and useful feedback is provided.
+
+
+max_rounds: 4
+
+input_interface:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+output_interface:
+  - "code"
+
+subflows_config:
+  CodeGenerator:
+    _target_: .CF_Code.instantiate_from_default_config
+    name: "CodeGenerator"
+    human_message_prompt_template:
+      _target_: langchain.PromptTemplate
+      template: |2-
+        # Feedback on the last proposed solution
+        {{code_feedback}}
+
+
+        Consider the original problem statement, the last proposed solution and the provided feedback. Does the solution need to be updated? If so, provide the corrected version of the code in the following format:
+        ```python
+        {{code_placeholder}}
+        ```
+        otherwise, reply:
+        "Final answer."
+      input_variables:
+        - code_feedback
+      partial_variables:
+        code_placeholder: "{{python_code}}"
+    input_interface_initialized:
+      - "code_feedback"
+  CodeCritic:
+    _target_: .CF_CodeCritic.instantiate_from_default_config
+
+topology:
+  # ~~~ Code Generator ~~~
+  - goal: "Generate/refine a solution."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeGenerator
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.RegexFirstOccurrenceExtractor
+          regex: '(?<=```python)([\s\S]*?)(?=```)'
+          regex_fallback: '(?<=```)([\s\S]*?)(?=```)'
+          input_key: "api_output"
+          output_key: "code"
+          strip: True
+          assert_unique: True
+        - _target_: flows.data_transformations.EndOfInteraction
+          end_of_interaction_string: "Final answer"
+          input_key: "api_output"
+          output_key: "end_of_interaction"
+        - _target_: flows.data_transformations.PrintPreviousMessages
+    reset: false
+
+  # ~~~ Code Critic ~~~
+  - goal: "Provide feedback for the candidate solution."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeCritic
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      keys_to_rename:
+        api_output: "code_feedback"
+
+    reset: true
+
+early_exit_key: "end_of_interaction"

CF_CodeCritic.py

@@ -0,0 +1,6 @@
+from flows.application_flows import OpenAIChatAtomicFlow
+
+
+class CF_CodeCritic(OpenAIChatAtomicFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

CF_CodeCritic.yaml

@@ -0,0 +1,87 @@
+name: "CodeCritic_Flow"
+description: |2-
+  Given a problem description and a solution candidate, provide useful feedback concerning the correctness of the solution candidate. 
+
+# ~~~ Input interface specification ~~~
+input_interface_non_initialized:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+  - "code"
+
+input_interface_initialized:
+  - "query"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "api_output"
+
+# ~~~ Flow specification ~~~
+model_name: "gpt-4"
+
+generation_parameters:
+  n: 1
+  max_tokens: 3000
+  temperature: 0.3
+
+  model_kwargs:
+    top_p: 0.2
+    frequency_penalty: 0
+    presence_penalty: 0
+
+system_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    Your goal is to identify potential issues with a competitive programming solution attempt.
+
+    The user will specify the problem by providing you with:
+      - the problem statement
+      - input description
+      - output description
+      - example test cases
+      - (optional) explanation of the test cases
+      - a Python solution attempt
+
+    Crucially, your goal is to correctly identify potential issues with the solution attempt, and not to provide the code implementation yourself.
+    The user will provide you with a task and an output format that you will strictly follow.
+  input_variables: []
+  template_format: jinja2
+
+human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: "{{query}}"
+  input_variables:
+    - "query"
+  template_format: jinja2
+
+init_human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    # Problem statement
+    {{problem_description}}
+
+    # Input description
+    {{input_description}}
+
+    # Output description
+    {{output_description}}
+
+    {{io_examples_and_explanation}}
+
+    # Python solution attempt:
+    ```python
+    {{code}}
+    ```
+
+
+    Consider the problem statement and the solution attempt. Are there any issues with the proposed solution or it is correct? Explain your reasoning very concisely, and do not provide code.
+  input_variables:
+    - "problem_description"
+    - "input_description"
+    - "output_description"
+    - "io_examples_and_explanation"
+    - "code"
+  template_format: jinja2
+
+

CF_CodeCriticWrongAttempt.py

@@ -0,0 +1,6 @@
+from flows.application_flows import OpenAIChatAtomicFlow
+
+
+class CF_CodeCriticWrongAttempt(OpenAIChatAtomicFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

CF_CodeCriticWrongAttempt.yaml

@@ -0,0 +1,90 @@
+name: "CodeCriticWrongAttempt_Flow"
+description: |2-
+  Given a problem description and an incorrect solution candidate, provide useful feedback for correcting the mistakes in the solution. 
+
+# ~~~ Input interface specification ~~~
+input_interface_non_initialized:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+  - "testing_results_summary"
+  - "code"
+
+input_interface_initialized:
+  - "query"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "api_output"
+
+# ~~~ Flow specification ~~~
+model_name: "gpt-4"
+
+generation_parameters:
+  n: 1
+  max_tokens: 3000
+  temperature: 0.3
+
+  model_kwargs:
+    top_p: 0.2
+    frequency_penalty: 0
+    presence_penalty: 0
+
+system_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    Your goal is to identify the issues with an incorrect competitive programming solution attempt.
+  
+    The user will specify the problem by providing you with:
+      - the problem statement
+      - input description
+      - output description
+      - example test cases
+      - (optional) explanation of the test cases
+      - an incorrect Python solution attempt and a description of its issue
+  
+    Crucially, your goal is to consider all aspects of the problem and pinpoint the issues with the solution attempt, and not to provide the code implementation yourself.
+    Some aspects to consider: Is the input correctly parsed? Is the output correctly formatted? Are the corner cases correctly handled? Is there a logical mistake with the algorithm itself?
+    Use the code execution results provided in the issue description to guide your reasoning/debugging.
+  input_variables: []
+  template_format: jinja2
+
+human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: "{{query}}"
+  input_variables:
+    - "query"
+  template_format: jinja2
+
+init_human_message_prompt_template:
+  _target_: langchain.PromptTemplate
+  template: |2-
+    # Problem statement
+    {{problem_description}}
+
+    # Input description
+    {{input_description}}
+
+    # Output description
+    {{output_description}}
+
+    {{io_examples_and_explanation}}
+
+    # Solution attempt to be fixed
+    ```python
+    {{code}}
+    ```
+
+    {{testing_results_summary}}
+
+
+    Consider the problem statement, the solution attempt and the issue. Why is the solution attempt incorrect? How should it be fixed? Explain your reasoning very concisely, and do not provide code.
+  input_variables:
+    - "problem_description"
+    - "input_description"
+    - "output_description"
+    - "io_examples_and_explanation"
+    - "code"
+    - "testing_results_summary"
+  template_format: jinja2

CF_CodeDebug.py

@@ -0,0 +1,12 @@
+from flows.base_flows import GeneratorCriticFlow
+
+
+class CF_CodeDebug(GeneratorCriticFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def _early_exit(self):
+        if self.flow_state.get("all_tests_passed", False):
+            return True
+
+        return super()._early_exit()

CF_CodeDebug.yaml

@@ -0,0 +1,158 @@
+name: "CodeDebug_Flow"
+description: |2-
+  Given a problem description, generate code, test and refine it until all tests pass or a maximum number of rounds is reached.
+
+# ~~~ Input interface specification ~~~
+input_interface:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+  - "public_tests_individual_io"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "code"
+
+# ~~~ Flow specification ~~~
+max_rounds: 4
+
+### Subflows specification
+subflows_config:
+  CodeGenerator:
+    _target_: .CF_Code.instantiate_from_default_config
+    name: "CodeGenerator"
+    model_name: "gpt-4"
+    human_message_prompt_template:
+      template: |2-
+        {{testing_results_summary}}
+
+
+        Consider the problem statement, the last proposed solution, and its issue. Provide a corrected version of the code that solves the original problem and resolves the issue, without any explanation, in the following format:
+        ```python
+        {{code_placeholder}}
+        ```
+      input_variables:
+        - testing_results_summary
+      partial_variables:
+        code_placeholder: "{{python_code}}"
+    input_interface_initialized:
+      - "testing_results_summary"
+  CodeTestingCritic:
+    _target_: .CF_CodeTesting.instantiate_from_default_config
+
+### Topology specification (specifies how the sequence of messages will flow from one of the subflows to another)
+topology:
+  # ~~~ Code Generator ~~~
+  - goal: "Generate/refine a solution."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeGenerator
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.RegexFirstOccurrenceExtractor
+          regex: '(?<=```python)([\s\S]*?)(?=```)'
+          regex_fallback: '(?<=```)([\s\S]*?)(?=```)'
+          input_key: "api_output"
+          output_key: "code"
+          strip: True
+          assert_unique: True
+        - _target_: flows.data_transformations.PrintPreviousMessages
+      keys_to_select:
+        - "code"
+
+    ### Reset flag
+    reset: false
+
+  # ~~~ Code Testing Critic ~~~
+  - goal: "Test the code on the public tests and provide a results summary."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeTestingCritic
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: martinjosifoski.CC_flows.src.data_transformations.CorrectnessFlag
+          input_key: "public_tests_results"
+          output_key: "all_tests_passed"
+        - _target_: martinjosifoski.CC_flows.src.data_transformations.TestingResultsSummaryGeneration
+          output_key: "testing_results_summary"
+
+          single_test_error_message: True
+
+          no_error_template: |2-
+            ${.issue_title}
+            All of the executed tests passed.
+
+          compilation_error_template: |2-
+            ${.issue_title}
+            The execution resulted in a compilation error.
+            ## Compilation error message:
+            {{error_message}}
+          timeout_error_template: |2-
+            ${.issue_title}
+            The execution timed out, the solution is not efficient enough.
+          runtime_error_template: |2-
+            ${.issue_title}
+            The execution resulted in a runtime error on the following test.
+            ## [Failed test] Input
+            ```
+            {{test_input}}
+            ```
+            ## [Failed test] Runtime error message
+            {{error_message}}
+          single_test_error_template: |2-
+            ${.issue_title}
+            The Python code does not solve the problem in the problem description due to logical errors. It fails the following test:
+            ## [Failed test] Input
+            ```
+            {{test_input}}
+            ```
+            ## [Failed test] Expected output
+            ```
+            {{expected_output}}
+            ```
+            ## [Failed test] Generated output
+            ```
+            {{generated_output}}
+            ```
+          all_tests_header: |2-
+            ${.issue_title}
+            The Python code does not solve the problem in the problem description due to logical errors. It fails on the following tests.
+          test_error_template: |2-
+            ## [Failed test {{idx}}]
+            ### [Failed test {{idx}}] Input
+            ```
+            {{test_input}}
+            ```
+            ### [Failed test {{idx}}] Expected output
+            ```
+            {{expected_output}}
+            ```
+            ### [Failed test {{idx}}] Generated output
+            ```
+            {{generated_output}}
+            ```
+          tests_separator: "\n\n"
+
+          issue_title: "# Issue with the last proposed solution"
+
+    ### Reset flag
+    reset: true

CF_CodeDebugCollab.py

@@ -0,0 +1,12 @@
+from flows.base_flows import GeneratorCriticFlow
+
+
+class CF_CodeDebugCollab(GeneratorCriticFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def _early_exit(self):
+        if self.flow_state.get("all_tests_passed", False):
+            return True
+
+        return super()._early_exit()

CF_CodeDebugCollab.yaml

@@ -0,0 +1,89 @@
+name: "CodeDebugCollab_Flow"
+description: |2-
+  Given a problem description, alternate between a step in which code is generated, and a step in which the produced code is evaluated and useful feedback is provided. 
+
+# ~~~ Input interface specification ~~~
+input_interface:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+  - "public_tests_individual_io"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "code"
+
+# ~~~ Flow specification ~~~
+max_rounds: 4
+
+subflows_config:
+  CodeGenerator:
+    _target_: .CF_Code.instantiate_from_default_config
+    name: "CodeGenerator"
+    model_name: "gpt-4"
+    human_message_prompt_template:
+      _target_: langchain.PromptTemplate
+      template: |2-
+        {{testing_results_summary}}
+      
+        {{code_feedback}}
+
+
+        Consider the problem statement, the last proposed solution, its issue and the provided feedback. Return a corrected version of the code that solves the original problem and resolves the issue, without any explanation, in the following format:
+        ```python
+        {{code_placeholder}}
+        ```
+      input_variables:
+        - code_feedback
+        - testing_results_summary
+      partial_variables:
+        code_placeholder: "{{python_code}}"
+    input_interface_initialized:
+      - "code_feedback"
+      - "testing_results_summary"
+  CodeDebugCritic:
+    _target_: .CF_CodeDebugCritic.instantiate_from_default_config
+
+topology:
+  # ~~~ Code Generator ~~~
+  - goal: "Generate/refine a solution."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeGenerator
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.RegexFirstOccurrenceExtractor
+          regex: '(?<=```python)([\s\S]*?)(?=```)'
+          regex_fallback: '(?<=```)([\s\S]*?)(?=```)'
+          input_key: "api_output"
+          output_key: "code"
+          strip: True
+          assert_unique: True
+      keys_to_select:
+        - "code"
+
+    reset: false
+
+  # ~~~ Code Critic Grounded in Tests ~~~
+  - goal: "Provide feedback for the candidate solution that is grounded in test results."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeDebugCritic
+
+    reset: true

CF_CodeDebugCritic.py

@@ -0,0 +1,13 @@
+from flows.base_flows import SequentialFlow
+
+
+class CF_CodeDebugCritic(SequentialFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def _early_exit(self):
+        if self.flow_state.get("all_tests_passed", False):
+            self.flow_state["code_feedback"] = None
+            return True
+
+        return super()._early_exit()

CF_CodeDebugCritic.yaml

@@ -0,0 +1,131 @@
+name: "CodeDebugCritic_Flow"
+description: "Given a problem description and a candidate solution, test the code and provide useful feedback concerning the correctness of the solution and the potential mistakes."
+
+# ~~~ Input interface specification ~~~
+input_interface:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+  - "public_tests_individual_io"
+  - "code"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "testing_results_summary"
+  - "all_tests_passed"
+  - "code_feedback"
+
+# ~~~ Flow specification ~~~
+public_tests_key: "public_tests_individual_io"
+
+subflows_config:
+  CodeTestingCritic:
+    _target_: .CF_CodeTesting.instantiate_from_default_config
+  CodeCriticWrongAttempt:
+    _target_: .CF_CodeCriticWrongAttempt.instantiate_from_default_config
+
+topology:
+  # ~~~ Code Testing Critic ~~~
+  - goal: "Test the code on the public tests and provide a results summary."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeTestingCritic
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: martinjosifoski.CC_flows.src.data_transformations.CorrectnessFlag
+          input_key: "public_tests_results"
+          output_key: "all_tests_passed"
+        - _target_: martinjosifoski.CC_flows.src.data_transformations.TestingResultsSummaryGeneration
+          output_key: "testing_results_summary"
+
+          single_test_error_message: True
+
+          no_error_template: |2-
+            ${.issue_title}
+            All of the executed tests passed.
+
+          compilation_error_template: |2-
+            ${.issue_title}
+            The execution resulted in a compilation error.
+            ## Compilation error message:
+            {{error_message}}
+          timeout_error_template: |2-
+            ${.issue_title}
+            The execution timed out, the solution is not efficient enough.
+          runtime_error_template: |2-
+            ${.issue_title}
+            The execution resulted in a runtime error on the following test.
+            ## [Failed test] Input
+            ```
+            {{test_input}}
+            ```
+            ## [Failed test] Runtime error message
+            {{error_message}}
+          single_test_error_template: |2-
+            ${.issue_title}
+            The Python code does not solve the problem in the problem description due to logical errors. It fails the following test:
+            ## [Failed test] Input
+            ```
+            {{test_input}}
+            ```
+            ## [Failed test] Expected output
+            ```
+            {{expected_output}}
+            ```
+            ## [Failed test] Generated output
+            ```
+            {{generated_output}}
+            ```
+          all_tests_header: |2-
+            ${.issue_title}
+            The Python code does not solve the problem in the problem description due to logical errors. It fails on the following tests.
+          test_error_template: |2-
+            ## [Failed test {{idx}}]
+            ### [Failed test {{idx}}] Input
+            ```
+            {{test_input}}
+            ```
+            ### [Failed test {{idx}}] Expected output
+            ```
+            {{expected_output}}
+            ```
+            ### [Failed test {{idx}}] Generated output
+            ```
+            {{generated_output}}
+            ```
+          tests_separator: "\n\n"
+
+          issue_title: "# Issue with the last proposed solution"
+
+  # ~~~ Feedback Generator ~~~
+  - goal: "Generate feedback grounded in the test results summary."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+    ### Flow Specification
+    flow: CodeCriticWrongAttempt
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyRename
+          old_key2new_key:
+            api_output: "code_feedback"
+    
+    reset: true
+

CF_CodeReflect.py

@@ -0,0 +1,6 @@
+from flows.base_flows import GeneratorCriticFlow
+
+
+class CF_CodeReflect(GeneratorCriticFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

CF_CodeReflect.yaml

@@ -0,0 +1,85 @@
+name: "CodeReflect_Flow"
+description: "Given a problem description, generate code, reflecting on it and improving it until a message suggesting that the code seems correct or a maximum number of rounds is reached."
+
+# ~~~ Input interface specification ~~~
+input_interface:
+  - "problem_description"
+  - "input_description"
+  - "output_description"
+  - "io_examples_and_explanation"
+
+# ~~~ Output interface specification ~~~
+output_interface:
+  - "code"
+
+# ~~~ Flow specification ~~~
+max_rounds: 4
+
+### Subflows specification
+subflows_config:
+  CodeGenerator:
+    _target_: .CF_Code.instantiate_from_default_config
+  CodeReflectCritic:
+    _target_: .FixedReply_CodeReflect.instantiate_from_default_config
+
+### Topology specification (specifies how the sequence of messages will flow from one of the subflows to another)
+topology:
+  # ~~~ Code Generator ~~~
+  - goal: "Generate/refine a solution."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+      keys_to_rename:
+        code_reflect_message: "query"
+
+    ### Flow Specification
+    flow: CodeGenerator
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.RegexFirstOccurrenceExtractor
+          regex: '(?<=```python)([\s\S]*?)(?=```)'
+          regex_fallback: '(?<=```)([\s\S]*?)(?=```)'
+          input_key: "api_output"
+          output_key: "code"
+          strip: True
+          assert_unique: True
+        - _target_: flows.data_transformations.EndOfInteraction
+          end_of_interaction_string: "Final answer"
+          input_key: "api_output"
+          output_key: "end_of_interaction"
+        - _target_: flows.data_transformations.PrintPreviousMessages
+      keys_to_select:
+        - "code"
+        - "end_of_interaction"
+
+    ### Reset flag
+    reset: false
+
+  - goal: "Generate a message that encourages reflection."
+
+    ### Input Interface
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+
+
+    ### Flow Specification
+    flow: CodeReflectCritic
+
+    ### Output Interface
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      keys_to_rename:
+        fixed_reply: "code_reflect_message"
+
+    ### Reset flag
+    reset: true
+
+early_exit_key: "end_of_interaction"

CF_CodeTesting.py

@@ -0,0 +1,52 @@
+from typing import Any, Dict
+
+from flows import logging
+from flows.utils.general_helpers import validate_parameters
+from .src.evaluation import testing_utils_codeforces
+from .CodeTesting import CodeTesting
+
+log = logging.get_logger(__name__)
+
+
+class CF_CodeTesting(CodeTesting):
+    REQUIRED_KEYS_CONFIG = []
+    REQUIRED_KEYS_CONSTRUCTOR = []
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @classmethod
+    def _validate_parameters(cls, kwargs):
+        validate_parameters(cls, kwargs)
+
+        if "public_tests_key" not in kwargs["flow_config"] and "hidden_tests_key" not in kwargs["flow_config"]:
+            raise ValueError("At least one of 'public_tests_key' "
+                             "and 'hidden_tests_key' must be specified in the config.")
+
+    def _get_test_data(self, input_data: Dict):
+        """This function retrieves (or generates) input-output pairs that will be used to test the implementation."""
+        test_data = {"public_tests_io": None, "hidden_tests_io": None}
+
+        if "public_tests_key" in self.flow_config:
+            test_data["public_tests_io"] = input_data[self.flow_config["public_tests_key"]]
+
+        if "hidden_tests_key" in self.flow_config:
+            test_data["hidden_tests_io"] = input_data[self.flow_config["hidden_tests_key"]]
+
+        return test_data
+
+    def _run_tests(self, input_data: Dict, test_data: Dict) -> Dict[str, Any]:
+        testing_results = testing_utils_codeforces.evaluate_solution_for_problem(
+            candidate_solution=input_data["code"],
+            **test_data
+        )
+
+        if "public_tests_results" in testing_results:
+            for test_output in testing_results["public_tests_results"]:
+                test_output["input"] = "\n".join(test_output["input"])
+
+        if "hidden_tests_results" in testing_results:
+            for test_output in testing_results["hidden_tests_results"]:
+                test_output["input"] = "\n".join(test_output["input"])
+
+        return testing_results

CF_CodeTesting.yaml

@@ -0,0 +1,12 @@
+name: "CF_CodeTesting"
+description: "Given code and test specified by input-output pairs, executes the code with the specific input, compares it with the output, and returns an informative message."
+
+input_interface:
+  - "code"
+  - "public_tests_individual_io"
+
+output_interface:
+  - "all_tests_passed"
+  - "testing_results_summary"
+
+public_tests_key: "public_tests_individual_io"

CodeTesting.py

@@ -0,0 +1,30 @@
+from typing import Any, Dict
+
+from flows import logging
+from flows.base_flows import AtomicFlow
+
+log = logging.get_logger(__name__)
+
+
+class CodeTesting(AtomicFlow):
+    REQUIRED_KEYS_CONFIG = []
+    REQUIRED_KEYS_CONSTRUCTOR = []
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def _get_test_data(self, input_data: Dict):
+        """This function retrieves (or generates) input-output pairs that will be used to test the implementation."""
+        raise NotImplementedError()
+
+    def _run_tests(self, input_data: Dict, test_data: Dict) -> Dict[str, Any]:
+        raise NotImplementedError()
+
+    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        # ~~~ Retrieve the test data ~~~
+        test_data = self._get_test_data(input_data)
+
+        # ~~~ Run tests ~~~
+        response: Dict[str, Any] = self._run_tests(input_data, test_data)
+
+        return response

FixedReply_CodeReflect.py

@@ -0,0 +1,6 @@
+from flows.base_flows import FixedReplyFlow
+
+
+class FixedReply_CodeReflect(FixedReplyFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

FixedReply_CodeReflect.yaml

@@ -0,0 +1,18 @@
+name: "CodeReflectCritic"
+description: "A flow that prompts the caller to reflect on the generated code and provide a corrected version if necessary."
+
+input_interface: []
+output_interface:
+  - fixed_reply
+
+fixed_reply: |2-
+  Consider the problem statement and the last proposed solution. Are you sure that the solution is provided in the requested format, and crucially, solves the problem?
+  If that is not the case, provide the corrected version of the code in the following format:
+  ```python
+  {{python_code}}
+  ```
+  otherwise, reply:
+  "Final answer."
+
+
+

__init__.py

@@ -0,0 +1,20 @@
+# ~~~ Codeforces ~~~
+# code
+from .CF_Code import CF_Code
+
+# code_reflect
+from .FixedReply_CodeReflect import FixedReply_CodeReflect
+from .CF_CodeReflect import CF_CodeReflect
+
+# code_collab
+from .CF_CodeCritic import CF_CodeCritic
+from .CF_CodeCollab import CF_CodeCollab
+
+# code_debug
+from .CF_CodeTesting import CF_CodeTesting
+from .CF_CodeDebug import CF_CodeDebug
+
+# cf-code_debug_collab
+from .CF_CodeCriticWrongAttempt import CF_CodeCriticWrongAttempt
+from .CF_CodeDebugCritic import CF_CodeDebugCritic
+from .CF_CodeDebugCollab import CF_CodeDebugCollab

src/data_transformations/__init__.py

@@ -0,0 +1,2 @@
+from .correctness_flag import CorrectnessFlag
+from .testing_results_summary_generation import TestingResultsSummaryGeneration

src/data_transformations/correctness_flag.py

@@ -0,0 +1,14 @@
+from typing import Dict, Any
+
+from flows.data_transformations.abstract import DataTransformation
+
+
+class CorrectnessFlag(DataTransformation):
+    def __init__(self, output_key, input_key):
+        super().__init__(output_key)
+        self.input_key = input_key
+
+    def __call__(self, data_dict: Dict[str, Any], **kwargs) -> Dict[str, Any]:
+        all_tests_passed = all([test_result["status"] for test_result in data_dict[self.input_key]])
+        data_dict[self.output_key] = all_tests_passed
+        return data_dict

src/data_transformations/testing_results_summary_generation.py

@@ -0,0 +1,107 @@
+from typing import Dict, Any
+
+import jinja2
+
+from flows.data_transformations.abstract import DataTransformation
+from flows.utils.general_helpers import unflatten_dict
+
+
+class TestingResultsSummaryGeneration(DataTransformation):
+    def __init__(self, output_key, **kwargs):
+        super().__init__(output_key)
+        self.params = kwargs
+        if "test_results_key" not in self.params:
+            self.params["test_results_key"] = "public_tests_results"
+        if "tests_passed_key" not in self.params:
+            self.params["tests_passed_key"] = "all_tests_passed"
+
+    def __call__(self, data_dict: Dict[str, Any], **kwargs) -> Dict[str, Any]:
+        if data_dict[self.params["tests_passed_key"]]:
+            # the execution did not result in any errors
+            data_dict[self.output_key] = self.params["no_error_template"]
+            return data_dict
+
+        test_data = unflatten_dict(data_dict)
+
+        if not test_data["compilation_status"]:
+            # compilation error occurred
+            kwargs = {
+                "error_message": test_data["compilation_error_message"].strip(),
+            }
+
+            message_content = (
+                jinja2.Environment(loader=jinja2.BaseLoader())
+                .from_string(self.params["compilation_error_template"])
+                .render(**kwargs)
+            )
+        elif test_data["timeout_error"]:
+            # timeout error occurred
+
+            message_content = self.params["timeout_error_template"]
+        else:
+            # code compiled successfully without timeouts
+
+            # retrieve the failed tests
+            failed_tests = [
+                test_result
+                for test_result in test_data[self.params["test_results_key"]]
+                if not test_result["status"]
+            ]
+
+            runtime_error_test = None
+            for test_result in failed_tests:
+                if test_result["generated_output"] is None:
+                    # runtime error occurred
+                    runtime_error_test = test_result
+
+            if runtime_error_test:
+                # construct the error message for the runtime error
+                kwargs = {
+                    "test_input": runtime_error_test["input"],
+                    "error_message": runtime_error_test["error_message"].strip(),
+                }
+
+                message_content = (
+                    jinja2.Environment(loader=jinja2.BaseLoader())
+                    .from_string(self.params["runtime_error_template"])
+                    .render(**kwargs)
+                )
+            else:
+                # construct the error message corresponding to a logical error
+
+                if self.params["single_test_error_message"]:
+                    # construct the error message for a single (the first) failed test
+                    first_failed_test = failed_tests[0]
+
+                    kwargs = {
+                        "test_input": first_failed_test["input"],
+                        "expected_output": first_failed_test["expected_output"],
+                        "generated_output": first_failed_test["generated_output"],
+                    }
+
+                    message_content = (
+                        jinja2.Environment(loader=jinja2.BaseLoader())
+                        .from_string(self.params["single_test_error_template"])
+                        .render(**kwargs)
+                    )
+                else:
+                    # construct the error message covering all failed tests
+                    parts = [self.params["all_tests_header"]]
+
+                    for idx, test_result in enumerate(failed_tests):
+                        kwargs = {
+                            "idx": idx + 1,
+                            "test_input": test_result["input"],
+                            "expected_output": test_result["expected_output"],
+                            "generated_output": test_result["generated_output"],
+                        }
+
+                        parts.append(
+                            jinja2.Environment(loader=jinja2.BaseLoader())
+                            .from_string(self.params["test_error_template"])
+                            .render(**kwargs)
+                        )
+
+                    message_content = self.params["tests_separator"].join(parts)
+        data_dict[self.output_key] = message_content
+        return data_dict

src/datasets/schema.py

@@ -0,0 +1,74 @@
+from typing import List, Tuple, Dict
+
+
+def assert_test_format_codeforces(tests: List[Tuple[List[str], str]]):
+    assert isinstance(tests, list) or tests is None
+    if tests is None:
+        return
+    for test in tests:
+        assert isinstance(test, list)
+        assert len(test) == 2
+        inputs, outputs = test
+        assert isinstance(inputs, list)
+        assert isinstance(outputs, str)
+        for input in inputs:
+            assert isinstance(input, str)
+
+
+def assert_entry_format_codeforces(obj: Dict):
+    # each data point must follow the same schema
+    assert isinstance(obj["id"], str)  # contest + problem_name = id, will not change when formatting changes
+    assert isinstance(obj["id_hash"], str)  # hashsum of all entries, any change to obj will change this
+    assert isinstance(obj["contest"], int)
+    assert isinstance(obj["problem_name"], str)
+    assert isinstance(obj["problem_url"], str)
+    assert isinstance(obj["solution_url"], str)
+
+    assert isinstance(obj["header"], str)
+    assert isinstance(obj["problem_description"], str)
+    assert isinstance(obj["input_description"], str)
+    assert isinstance(obj["output_description"], str)
+    assert isinstance(obj["note"], str) or obj["note"] is None
+
+    assert isinstance(obj["difficulty"], int)
+    assert isinstance(obj["tags"], list)
+    assert isinstance(obj["working_solution"], str)  # can be empty
+
+    assert_test_format_codeforces(obj["public_tests_io"])
+    assert_test_format_codeforces(obj["public_tests_individual_io"])
+    assert_test_format_codeforces(obj["hidden_tests_io"])
+
+
+def assert_test_format_leetcode(tests: List[Tuple[List[str], str]]):
+    pass
+    # ToDo: Uncomment after the test format is updated
+    # assert isinstance(tests, list)
+    # for test in tests:
+    #     assert isinstance(test, tuple)
+    #     assert len(test) == 2
+    #     x, y = test
+    #     assert isinstance(x, str)
+    #     assert isinstance(y, str)
+
+
+def assert_entry_format_leetcode(obj: Dict):
+    # each data point must follow the same schema
+    assert isinstance(obj["id"], str)  # contest + problem_name = id, will not change when formatting changes
+    assert isinstance(obj["id_hash"], str)  # hashsum of all entries, any change to obj will change this
+    assert isinstance(obj["index"], int)
+    assert isinstance(obj["problem_name"], str)
+    assert isinstance(obj["problem_url"], str)
+
+    assert isinstance(obj["problem_description"], str)
+    assert isinstance(obj["constraints"], str)
+    assert isinstance(obj["python_stub"], str)
+    assert isinstance(obj["difficulty"], str) and obj["difficulty"] in {"easy", "medium", "hard"}
+
+    # ToDo: Should be added
+    # assert isinstance(obj['tags'], list)
+    # assert isinstance(obj['solution_url'], str)
+    # assert isinstance(obj['working_solution'], str)  # can be empty
+
+    # ToDo: Uncomment after the test format is updated
+    # assert_test_format_leetcode(obj['public_tests_io'])
+    # assert_test_format_leetcode(obj['hidden_tests_io'])

src/evaluation/testing_utils_codeforces.py

@@ -0,0 +1,449 @@
+# This is based heavily on the huggingface APPS metric
+import re
+
+# to run the solution files we're using a timing based approach
+import signal
+import sys
+
+# for capturing the stdout
+from io import StringIO
+from typing import List, Tuple
+
+# used for testing the code that reads from input
+from unittest.mock import patch, mock_open
+
+import numpy as np
+from pyext import RuntimeModule
+from wrapt_timeout_decorator import timeout as wrapt_timeout
+import threading
+
+from ..datasets.schema import assert_test_format_codeforces
+
+from flows import logging
+
+log = logging.get_logger(__name__)
+lock = threading.Lock()
+
+
+def evaluate_solution_for_problem(
+    candidate_solution,
+    hidden_tests_io=None,
+    public_tests_io=None,
+    timeout=10,
+    debug=False,
+    add_extra_imports=False,
+    allow_truncated_io=False,
+):
+    """See the readme for the output format of this function."""
+    if hidden_tests_io is None:
+        hidden_tests_io = []
+    if public_tests_io is None:
+        public_tests_io = []
+
+    if candidate_solution is None:
+        results_dict = {
+            "compilation_status": False,
+            "compilation_error_message": "No code was provided.",
+            "timeout_error": False,
+            "hidden_tests_results": [
+                {
+                    "status": False,
+                    "error_message": "No code was provided.",
+                    "generated_output": None,
+                    "input": test[0],
+                    "expected_output": test[1],
+                }
+                for test in hidden_tests_io
+            ],
+            "public_tests_results": [
+                {
+                    "status": False,
+                    "error_message": "No code was provided.",
+                    "generated_output": None,
+                    "input": test[0],
+                    "expected_output": test[1],
+                }
+                for test in public_tests_io
+            ],
+        }
+        return results_dict
+
+    @wrapt_timeout(timeout, use_signals=False)
+    def run_tests():
+        hidden_tests_results = check_correctness(
+            candidate_solution, hidden_tests_io, timeout, debug, add_extra_imports, allow_truncated_io
+        )
+        public_tests_results = check_correctness(
+            candidate_solution, public_tests_io, timeout, debug, add_extra_imports, allow_truncated_io
+        )
+
+        return hidden_tests_results, public_tests_results
+
+    try:
+        lock.acquire()
+        hidden_tests_results, public_tests_results = run_tests()
+        timeout_error_occurred = False
+        lock.release()
+    except BaseException as e:
+        lock.release()
+        log.info(e)
+        hidden_tests_results = {}
+        public_tests_results = {}
+
+        hidden_tests_results["compilation_status"] = True
+        public_tests_results["compilation_status"] = True
+        timeout_error_occurred = True
+        hidden_tests_results["error_message"] = "Timeout error."
+
+        hidden_tests_results["results"] = [
+            {
+                "status": False,
+                "error_message": hidden_tests_results["error_message"],
+                "generated_output": None,
+                "input": test[0],
+                "expected_output": test[1],
+            }
+            for test in hidden_tests_io
+        ]
+        public_tests_results["results"] = [
+            {
+                "status": False,
+                "error_message": hidden_tests_results["error_message"],
+                "generated_output": None,
+                "input": test[0],
+                "expected_output": test[1],
+            }
+            for test in public_tests_io
+        ]
+
+    # the compilation status shouldn't depend on the tests
+    assert hidden_tests_results["compilation_status"] == public_tests_results["compilation_status"]
+
+    results_dict = {
+        "compilation_status": hidden_tests_results["compilation_status"],
+        "compilation_error_message": hidden_tests_results["error_message"],
+        "timeout_error": timeout_error_occurred,
+        "hidden_tests_results": hidden_tests_results["results"],
+        "public_tests_results": public_tests_results["results"],
+    }
+
+    return results_dict
+
+
+def check_correctness(
+    candidate_solution: str,
+    tests: List[Tuple[List[str], str]],
+    timeout: int = 6000,
+    debug=True,
+    add_extra_imports=False,
+    allow_truncated_io=True,
+):
+    """
+    wrapping the testing code in a global timeout, based on huggingface code
+    """
+
+    assert_test_format_codeforces(tests)
+    inputs, outputs = [], []
+    if len(tests) > 0:
+        inputs, outputs = zip(*tests)
+
+    compilation_error, results = run_test(
+        candidate_solution, inputs, outputs, timeout, debug, add_extra_imports, allow_truncated_io
+    )
+
+    assert len(results) == len(inputs)
+
+    for result in results:
+        assert isinstance(result["generated_output"], str) or result["generated_output"] is None
+        assert isinstance(result["status"], bool)
+        assert isinstance(result["error_message"], str) or result["error_message"] is None
+        assert isinstance(result["input"], list)
+        assert isinstance(result["expected_output"], str)
+
+    compilation_status = compilation_error == ""
+    if compilation_status:
+        compilation_error = None
+
+    return {"compilation_status": compilation_status, "error_message": compilation_error, "results": results}
+
+
+class TimeoutException(Exception):
+    pass
+
+
+def timeout_handler(signum, frame):
+    log.info("Alarm went off")
+    # return
+    raise TimeoutException
+
+
+signal.signal(signal.SIGALRM, timeout_handler)
+
+
+# used to capture stdout as a list
+# from https://stackoverflow.com/a/16571630/6416660
+# alternative use redirect_stdout() from contextlib
+class Capturing(list):
+    def __enter__(self):
+        self._stdout = sys.stdout
+        sys.stdout = self._stringio = StringIO()
+        # Make closing the StringIO a no-op
+        self._stringio.close = lambda x: 1
+        return self
+
+    def __exit__(self, *args):
+        self.extend(self._stringio.getvalue().splitlines())
+        del self._stringio  # free up some memory
+        sys.stdout = self._stdout
+
+
+def run_test(code, inputs, outputs, timeout: int = 10, debug=True, add_extra_imports=False, allow_truncated_io=True):
+    """
+    runs the code and tries to match inputs and outputs
+    the scraped testcases may be incomplete
+    if allow_truncated_io==True, then we ignore an EOF exception at the end of the generated output
+    """
+    # Disable functionalities that can make destructive changes to the test.
+
+    results = []
+
+    if isinstance(code, list):
+        tmp_test = code
+    elif isinstance(code, str):
+        tmp_test = code.split("\n")
+    else:
+        raise AssertionError("code must be provided as list of lines or string with \\n linebreaks.")
+
+    # parse the code into code and imports
+    import_lines = []
+    future_import_lines = []
+    code_lines = []
+    for x in tmp_test:
+        if (not x.startswith("from ")) and (not x.startswith("import ")):
+            code_lines.append("\t" + x + "\n")
+        else:
+            if "__future__" in x:
+                future_import_lines.append(x + "\n")
+            else:
+                import_lines.append(x + "\n")
+
+    # assemble a new solution snippet which wraps the generated solution in a function code()
+    new_test = "stdin = sys.stdin\nstdout = sys.stdout\n"
+    new_test += '__name__="__main__"\n'
+    new_test += "def code():\n"
+    new_test += "\tstdin = sys.stdin\n\tstdout = sys.stdout\n"
+
+    for line in code_lines:
+        new_test += line
+
+    sol = "\n".join(future_import_lines)
+    sol += "import sys\n"
+    if add_extra_imports:
+        sol += "import time\nimport itertools\nfrom itertools import accumulate, product, permutations, combinations\nimport collections\nfrom collections import Counter, OrderedDict, deque, defaultdict, ChainMap\nfrom functools import lru_cache\nimport math\nfrom math import sqrt, sin, cos, tan, ceil, fabs, floor, gcd, exp, log, log2\nimport fractions\nfrom typing import List, Tuple\nimport numpy as np\nimport random\nimport heapq\nfrom heapq import *\n"
+    sol += "\n".join(import_lines) + "\n" + new_test
+
+    if debug:
+        log.info(f"sol = {sol}")
+    method_name = "code"
+    signal.alarm(timeout)
+
+    # convert the solution snippet into a pyext runtime module
+    sol_module = None
+    try:
+        sol_module = RuntimeModule.from_string("tmp_sol", "", sol)
+        signal.alarm(0)
+    except Exception as e:
+        signal.alarm(0)
+        if debug:
+            log.info(f"type 1 compilation error = {e}")
+        for inp, out in zip(inputs, outputs):
+            # consider all inputs failed
+            results.append(
+                {
+                    "status": False,
+                    "input": inp,
+                    "expected_output": out,
+                    "generated_output": None,
+                    "error_message": repr(e),
+                }
+            )
+        return repr(e), results
+
+    assert sol_module is not None
+    signal.alarm(0)
+
+    try:
+        method = getattr(sol_module, method_name)  # get_attr second arg must be str
+    except:
+        signal.alarm(0)
+        e = sys.exc_info()
+        log.info(f"unable to get function error = {e}")
+
+        for inp, out in zip(inputs, outputs):
+            # consider all inputs failed
+            results.append(
+                {
+                    "status": False,
+                    "input": inp,
+                    "expected_output": out,
+                    "generated_output": None,
+                    "error_message": repr(e),
+                }
+            )
+        return repr(e), results
+
+    # go through all tests, call our runtime module with the inputs
+    # then compare with the reference output
+    for index, (test_input, reference_output) in enumerate(zip(inputs, outputs)):
+
+        result_object = {
+            "input": test_input,
+            "expected_output": reference_output,
+        }
+
+        # if the last token of the input is truncated and marked with "..." we delete it
+        input_truncated = False
+        if "".join(test_input).strip().endswith("...") and allow_truncated_io:
+            test_input = test_input[:-1]
+            input_truncated = True
+
+        # sometimes the last input token is ""
+        # if len(test_input)>0:
+        #    if test_input[-1]=="":
+        #        test_input = test_input[:-1]
+
+        error_code = None
+        with Capturing() as generated_output:
+            try:
+                call_method(method, test_input)
+                # reset the alarm
+                signal.alarm(0)
+            except Exception as e:
+                # runtime error or took too long
+                signal.alarm(0)
+                error_code = e
+                if debug:
+                    log.info(f"Call-based runtime error or time limit exceeded error = {repr(e)}{e}")
+            signal.alarm(0)
+
+        # in some cases we run into truncated tests
+        # in such cases we expect the error code to be None, EOFError or ValueError
+        if (
+            (input_truncated or reference_output.strip().endswith("..."))
+            and allow_truncated_io
+            and (error_code is None or isinstance(error_code, EOFError) or isinstance(error_code, ValueError))
+        ):
+
+            generated_output = generated_output[:-1]
+            reference_output = reference_output.rstrip("...")
+            if len(generated_output) == 0:
+                # no output left, we pass by default
+                result_object.update(
+                    **{
+                        "status": True,
+                        "generated_output": "\n".join(generated_output),
+                        "error_message": None,
+                    }
+                )
+                results.append(result_object)
+            else:
+                result_object.update(
+                    **{
+                        "status": string_compare(generated_output, reference_output, True),
+                        "generated_output": "\n".join(generated_output),
+                        "error_message": None,
+                    }
+                )
+                results.append(result_object)
+
+        # if the input and output are not truncated, we don't allow any errors
+        elif error_code is not None:
+            result_object.update(**{"status": False, "generated_output": None, "error_message": repr(error_code)})
+            results.append(result_object)
+        # finally, if there are no errors, we expect the output to match the reference output
+        else:
+            # the execution went well, let's compare the outputs
+            result_object.update(
+                **{
+                    "status": string_compare(generated_output, reference_output, False),
+                    "generated_output": "\n".join(generated_output),
+                    "error_message": None,
+                }
+            )
+            results.append(result_object)
+
+    return "", results
+
+
+def string_compare(candidate, correct, truncate_output=False, floating_point_accuracy=0.01):
+    candidate = [o.strip().lower() for o in candidate]
+    correct = correct.strip().lower()
+
+    # normalize whitespace
+    candidate = "\n".join(candidate)
+    candidate = re.sub("\s+", " ", candidate).strip()
+    correct = re.sub("\s+", " ", correct).strip()
+
+    # split into individual tokens
+    candidate = candidate.split(" ")
+    correct = correct.split(" ")
+
+    # some tests may be truncated, if we allow this we don't enforce equal length of inputs/outputs
+    if not truncate_output:
+        if not len(candidate) == len(correct):
+            return False
+
+    # if we allow truncated io, the last token of the output may have been corrupted
+    if truncate_output:
+        correct = correct[:-1]
+
+    # when zip is used for lists of unequal length it will give as many pairs as there are items in the shorter list
+    for left, right in zip(candidate, correct):
+        if left == right:
+            continue
+
+        try:
+            int_left = int(left)
+            int_right = int(right)
+            if int_left == int_right:
+                continue
+        except ValueError:
+            pass
+
+        try:
+            float_left = float(left)
+            float_right = float(right)
+            if np.abs(float_left - float_right) < floating_point_accuracy:
+                continue
+        except ValueError:
+            pass
+
+        return False
+
+    return True
+
+
+def call_method(method, inputs):
+    if isinstance(inputs, list):
+        inputs = "\n".join(inputs)
+
+    inputs_line_iterator = iter(inputs.split("\n"))
+
+    # sys.setrecursionlimit(10000)
+
+    # @patch('builtins.input', side_effect=inputs.split("\n"))
+    @patch("builtins.open", mock_open(read_data=inputs))
+    @patch("sys.stdin", StringIO(inputs))
+    @patch("sys.stdin.readline", lambda *args: next(inputs_line_iterator))
+    @patch("sys.stdin.readlines", lambda *args: inputs.split("\n"))
+    @patch("sys.stdin.read", lambda *args: inputs)
+    # @patch('sys.stdout.write', print)
+    def _inner_call_method(_method):
+        try:
+            return _method()
+        except SystemExit as e:
+            pass
+        finally:
+            pass
+
+    return _inner_call_method(method)

src/evaluation/testing_utils_leetcode.py

@@ -0,0 +1,258 @@
+# This is based heavily on the huggingface APPS metric
+# to run the solution files we're using a timing based approach
+# for capturing the stdout
+# used for testing the code that reads from input
+import logging
+import re
+from subprocess import Popen, PIPE, TimeoutExpired
+from typing import List, Tuple
+import threading
+
+log = logging.getLogger(__name__)
+lock = threading.Lock()
+
+def evaluate_solution_for_problem(
+    candidate_solution,
+    python_stub,
+    hidden_tests_io=None,
+    public_tests_io=None,
+    timeout=10,
+    debug=False,
+    add_extra_imports=False,
+):
+    with lock:
+        """See the readme for the output format of this function."""
+        if hidden_tests_io is None:
+            hidden_tests_io = []
+        if public_tests_io is None:
+            public_tests_io = []
+
+        if candidate_solution is None:
+            results_dict = {
+                "compilation_status": False,
+                "compilation_error_message": "No code was provided.",
+                "timeout_error": False,
+                "hidden_tests_results": [
+                    {
+                        "status": False,
+                        "error_message": "No code was provided.",
+                        "generated_output": None,
+                        "input": test[0],
+                        "expected_output": test[1],
+                    }
+                    for test in hidden_tests_io
+                ],
+                "public_tests_results": [
+                    {
+                        "status": False,
+                        "error_message": "No code was provided.",
+                        "generated_output": None,
+                        "input": test[0],
+                        "expected_output": test[1],
+                    }
+                    for test in public_tests_io
+                ],
+            }
+            return results_dict
+
+        hidden_tests_results = check_correctness(
+            candidate_solution, python_stub, hidden_tests_io, timeout, debug, add_extra_imports
+        )
+        public_tests_results = check_correctness(
+            candidate_solution, python_stub, public_tests_io, timeout, debug, add_extra_imports
+        )
+
+        # the compilation status shouldn't depend on the tests
+        if len(hidden_tests_io) > 0 and len(public_tests_io) > 0:
+            assert hidden_tests_results["compilation_status"] == public_tests_results["compilation_status"]
+
+        compilation_status = True
+        error_message = None
+        timeout_error = False
+
+        if len(hidden_tests_io) > 0:
+            compilation_status = compilation_status and hidden_tests_results["compilation_status"]
+            error_message = hidden_tests_results["error_message"]
+            timeout_error = timeout_error or hidden_tests_results["timeout_error"]
+
+        if len(public_tests_io) > 0:
+            compilation_status = compilation_status and public_tests_results["compilation_status"]
+            error_message = public_tests_results["error_message"]
+            timeout_error = timeout_error or public_tests_results["timeout_error"]
+
+        results_dict = {
+            "compilation_status": compilation_status,
+            "compilation_error_message": error_message,
+            "timeout_error": timeout_error,
+            "hidden_tests_results": hidden_tests_results["results"],
+            "public_tests_results": public_tests_results["results"],
+        }
+
+        return results_dict
+
+
+def check_correctness(
+    candidate_solution: str,
+    python_stub: str,
+    tests: List[Tuple[List[str], str]],
+    timeout: int = 6000,
+    debug=True,
+    add_extra_imports=False,
+):
+    compilation_status = True
+    compilation_error = None
+    results = []
+    timeout_occurred = False
+
+    for idx, test in enumerate(tests):
+        inp, out, expl = test
+        result = one_test(
+            candidate_solution, python_stub, inp, out, timeout=timeout, debug=debug, add_extra_imports=add_extra_imports
+        )
+        error_message = result["error_message"]
+
+        if error_message is not None:
+            if "syntaxerror" in error_message.lower():
+                compilation_status = False
+                compilation_error = error_message
+            if "timeout" in error_message.lower():
+                timeout_occurred = True
+        results.append(result)
+
+        if timeout_occurred:
+            break
+
+    if timeout_occurred:
+        return {
+            "compilation_status": True,
+            "timeout_error": True,
+            "error_message": "Timeout error.",
+            "results": results,
+        }
+
+    return {
+        "compilation_status": compilation_status,
+        "timeout_error": False,
+        "error_message": compilation_error,
+        "results": results,
+    }
+
+
+def one_test(candidate_solution, python_stub, inp, out, timeout=10, debug=False, add_extra_imports=False):
+    python_stub = python_stub.strip()
+    candidate_solution = candidate_solution.strip()
+
+    out = out.replace("null", "None").replace("true", "True").replace("false", "False")
+
+    # reformat the solution and parse class and method name
+    class_def, signature = python_stub.split("    def ")
+    class_name = class_def.split("class ")[1].strip().rstrip(":")
+    func_name, _ = signature.split("(")
+
+    # reformatting the input
+    first_param = r"^\w+\s\=\s"
+    later_params = r",\s\w+\s\=\s"
+
+    inp = re.sub(first_param, "", inp)
+    inp = re.sub(later_params, ", ", inp)
+
+    # we add custom code to invoke the solution
+    before_output = "AFTER THIS COMES OUR OWN GENERATED OUTPUT !@#!@!"
+    after_output = "AFTER THIS COMES OUR VERDICT !@#!@!"
+
+    if add_extra_imports:
+        sol = f"""
+from collections import *
+from math import *
+import math
+from functools import *
+from heapq import *
+import heapq
+import itertools
+from itertools import *
+import bisect
+from bisect import *
+"""
+    else:
+        sol = ""
+
+    sol += f"""
+from typing import List, Tuple, Optional
+{candidate_solution}
+sfohsdfdsfjhsdkfjhsdkjfh = {class_name}()
+res = sfohsdfdsfjhsdkfjhsdkjfh.{func_name}({inp})
+
+def nested_list_convert(inp):
+    try:
+        try:
+            inp = list(inp)
+        except BaseException as e:
+            return inp
+        out = []
+        for i in inp:
+            out.append(nested_list_convert(i))
+    except BaseException as e:
+        return inp
+    return out
+
+matching = False
+matching = matching or res == {out}
+matching = matching or nested_list_convert(res) == {out}
+matching = matching or nested_list_convert(res) == nested_list_convert({out})
+matching = matching or str({out})==str(res).replace("{{","[").replace("(","[").replace("}}","]").replace(")","]")
+matching = matching or str({out})==str(res).replace("{{","[").replace("(","[").replace("}}","]").replace(")","]")
+print("res: ", res)
+print("out: ", {out})
+print("{before_output}")
+print(res)
+print("{after_output}")
+print(matching)
+"""
+
+    cmd = "python3"
+
+    proc = Popen([cmd, "-c", sol], stdin=PIPE, stdout=PIPE, stderr=PIPE)
+
+    result_object = {"input": inp, "expected_output": out.strip('"')}
+
+    try:
+        stdout, stderr = proc.communicate("", timeout=timeout)
+    except TimeoutExpired as e:
+        if debug:
+            log.info(f"Timeout error, timeout={timeout}")
+        result_object.update({"status": False, "error_message": "Timeout error.", "generated_output": None})
+        return result_object
+
+    finally:
+        proc.kill()
+
+    stdout = stdout.decode()
+    stderr = stderr.decode().lower()
+
+    if stderr == "":
+        # No compilation or runtime error
+        stderr = None
+    else:
+        # Runtime or compilation error (distinction is made by the presence of "syntaxerror" in the error message)
+        result_object.update(**{"status": False, "error_message": stderr, "generated_output": None})
+        return result_object
+
+    try:
+        generated_output = stdout.split(before_output)[1]
+        generated_output, verdict = generated_output.split(after_output)
+        result_object.update(
+            **{
+                "status": verdict.strip() == "True",
+                "error_message": stderr,
+                "generated_output": generated_output.strip(),
+            }
+        )
+        return result_object
+    except IndexError as e:
+        raise Exception(f"An unexpected error has occurred while parsing the following generated output: {stdout}")
+        # Used in debugging
+        # log.info(e)
+        # result_object.update(
+        #     **{"status": False, "error_message": "The output couldn't be parsed", "generated_output": None}
+        # )
+        # return result_object