Skip to content
This repository has been archived by the owner on Mar 16, 2024. It is now read-only.

Commit

Permalink
Feature/add reflection rebased merged (#477)
Browse files Browse the repository at this point in the history 
* Add Reflexion

* cleanup

* Add problem solver

* cleanup reflexion impl

* further cleanup reflection workflow

* further cleanup reflection workflow

* finish addition

* 'Refactored by Sourcery' (#476)

Co-authored-by: Sourcery AI <>

* cleanup for merge

* update data

* fix indices

---------

Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
  • Loading branch information
@emrgnt-cmplxty @sourcery-ai
emrgnt-cmplxty and sourcery-ai[bot] authored Aug 10, 2023
1 parent 49707fa commit 156604c
Showing 1 changed file with 118 additions and 112 deletions.
230 changes: 118 additions & 112 deletions research/study_leetcode/run_automata_problem_solver.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,99 +101,50 @@ def main(): # sourcery skip: docstrings-for-functions
embedding_provider = OpenAIEmbeddingProvider()
test_stand = LeetCodeTestStand(loader=loader)

for index in [solver.indices[20]]:
# try:
problem_context = loader.get_problem_context(index)
agent_message_buffer = []

print(
f"Running w/ problem at index {index} and context:\n\n{problem_context}"
)

solutions_finder = LeetCodeSolutionsFinder(
embedding_provider,
max_entry_id=loader.get_frontend_problem_id(
index
), # Solutions are indexed along frontend problem id
max_num_examples=args.max_num_examples,
num_examples_to_screen=args.num_examples_to_screen,
solutions_data_path=args.solutions_data_path,
lowest_difficulty=args.lowest_difficulty_supported,
)

formatted_instructions = SOLVER_INSTRUCTIONS.format(
PROBLEM_STATEMENT=problem_context,
SHORTENED_PROBLEM_STATEMENT=f"{problem_context[:200]}...",
)

# Construcs an agent that will provide a solution to the
# given LeetCode problem when ran

solution_agent = solver.construct_agent(
loader.get_problem_header(index),
formatted_instructions,
solutions_finder,
args.include_leetcode_best_old_solution,
)

# If a related solution was fetched, update the agent message buffer
# so that these messages will be included after reflexion in future iterations
if args.include_leetcode_best_old_solution:
agent_message_buffer = deepcopy(
solution_agent.conversation.messages[-2:]
for index in solver.indices:
try:
problem_context = loader.get_problem_context(index)
agent_message_buffer = []

print(
f"Running w/ problem at index {index} and context:\n\n{problem_context}"
)
print(f"Storing an agent message buffer = {agent_message_buffer}")
result = solution_agent.run()

cleaned_result = (
result.split("```python")[1].split("```")[0].replace("\\n", "\n")
)
print(f"Final Cleaned Result:\n{cleaned_result}")

exception, test_results = test_stand.run_tests_for_example(
index, cleaned_result
)

for i in range(NUM_REFLECTIONS):
reflection_agent = solver.build_reflection_agent(
loader.get_problem_header(index).split("Constraints")[0],
result,
test_results,
exception,
solutions_finder,

solutions_finder = LeetCodeSolutionsFinder(
embedding_provider,
max_entry_id=loader.get_frontend_problem_id(
index
), # Solutions are indexed along frontend problem id
max_num_examples=args.max_num_examples,
num_examples_to_screen=args.num_examples_to_screen,
solutions_data_path=args.solutions_data_path,
lowest_difficulty=args.lowest_difficulty_supported,
)

formatted_instructions = SOLVER_INSTRUCTIONS.format(
PROBLEM_STATEMENT=problem_context,
SHORTENED_PROBLEM_STATEMENT=f"{problem_context[:200]}...",
)

reflection = reflection_agent.run()
# Construcs an agent that will provide a solution to the
# given LeetCode problem when ran

solution_agent = solver.construct_agent(
loader.get_problem_header(index),
formatted_instructions,
solutions_finder,
False,
)

solution_agent.conversation.messages.extend(agent_message_buffer)

query_message = OpenAIChatMessage(
role="assistant",
content="Thoughts:\n Excellent, the previous solution will be helpful in guiding our best solution. Further, we should query the solution history for additional information, this is a query we can only do once.\nAction:\n I will now query the past solution history.",
function_call=FunctionCall(
name="attempted-solution-history",
arguments={
"query": "Past solutions for 'Minimum Reverse Operations'"
},
),
)

solution_agent.conversation.messages.append(query_message)

response_message = OpenAIChatMessage(
role="user",
content=f"Previous Result:\n{result}\n\nTest Results:\n{test_results}\n\nExceptions:\n{exception}\n\nReflections:\n{reflection}",
args.include_leetcode_best_old_solution,
)

solution_agent.conversation.messages.append(response_message)

# If a related solution was fetched, update the agent message buffer
# so that these messages will be included after reflexion in future iterations
if args.include_leetcode_best_old_solution:
agent_message_buffer = deepcopy(
solution_agent.conversation.messages[-2:]
)
print(
f"Storing an agent message buffer = {agent_message_buffer}"
)
result = solution_agent.run()

cleaned_result = (
Expand All @@ -207,35 +158,90 @@ def main(): # sourcery skip: docstrings-for-functions
index, cleaned_result
)

print(
f"~~~Testing~~~\n\nException:\n{exception}\nTest Result:\n{test_results}"
)

def prep_for_leetcode(code: str) -> str:
lines = code.split("\n")
modified_lines = ["class Solution():"]
for line in lines:
if line.startswith("def "):
line = "def " + line[4:].replace("(", "(self, ", 1)
modified_lines.append(f" {line}")
return "\n".join(modified_lines)

lang = ProgrammingLanguage.PYTHON3
sub = LeetCodeSubmission(
code=prep_for_leetcode(cleaned_result),
lang=lang,
question_id=loader.get_backend_problem_id(index),
question_slug=loader.get_problem_slug(index),
)

env = LeetCodeEnv()

status, reward, done, submission_result = env.step(sub)
print(status, reward, done, submission_result)
solver.log_result(index, reward)
# except:
# print("Failed to run example")
# solver.log_result(index, False)
for i in range(NUM_REFLECTIONS):
reflection_agent = solver.build_reflection_agent(
loader.get_problem_header(index).split("Constraints")[0],
result,
test_results,
exception,
solutions_finder,
)

reflection = reflection_agent.run()

solution_agent = solver.construct_agent(
loader.get_problem_header(index),
formatted_instructions,
solutions_finder,
False,
)

solution_agent.conversation.messages.extend(
agent_message_buffer
)

query_message = OpenAIChatMessage(
role="assistant",
content="Thoughts:\n Excellent, the previous solution will be helpful in guiding our best solution. Further, we should query the solution history for additional information, this is a query we can only do once.\nAction:\n I will now query the past solution history.",
function_call=FunctionCall(
name="attempted-solution-history",
arguments={
"query": "Past solutions for 'Minimum Reverse Operations'"
},
),
)

solution_agent.conversation.messages.append(query_message)

response_message = OpenAIChatMessage(
role="user",
content=f"Previous Result:\n{result}\n\nTest Results:\n{test_results}\n\nExceptions:\n{exception}\n\nReflections:\n{reflection}",
)

solution_agent.conversation.messages.append(response_message)

result = solution_agent.run()

cleaned_result = (
result.split("```python")[1]
.split("```")[0]
.replace("\\n", "\n")
)
print(f"Final Cleaned Result:\n{cleaned_result}")

exception, test_results = test_stand.run_tests_for_example(
index, cleaned_result
)

print(
f"~~~Testing~~~\n\nException:\n{exception}\nTest Result:\n{test_results}"
)

def prep_for_leetcode(code: str) -> str:
lines = code.split("\n")
modified_lines = ["class Solution():"]
for line in lines:
if line.startswith("def "):
line = "def " + line[4:].replace("(", "(self, ", 1)
modified_lines.append(f" {line}")
return "\n".join(modified_lines)

lang = ProgrammingLanguage.PYTHON3
sub = LeetCodeSubmission(
code=prep_for_leetcode(cleaned_result),
lang=lang,
question_id=loader.get_backend_problem_id(index),
question_slug=loader.get_problem_slug(index),
)

env = LeetCodeEnv()

status, reward, done, submission_result = env.step(sub)
print(status, reward, done, submission_result)
solver.log_result(index, reward)
except:
print("Failed to run example")
solver.log_result(index, False)


if __name__ == "__main__":
Expand Down

0 comments on commit 156604c

Please sign in to comment.