1
1
import autogen
2
- from typing import List , Tuple
2
+ from typing import List , Tuple , Dict
3
3
import re
4
4
import tempfile
5
5
import os
6
6
import pytest
7
7
import shutil
8
+ import requests
9
+ import json
10
+ from llm .openrouter import OpenRouterLLM
11
+
12
+ class CustomAssistantAgent (autogen .AssistantAgent ):
13
+ def __init__ (self , name : str , llm_config : dict , openrouter_llm : OpenRouterLLM ):
14
+ super ().__init__ (name = name , llm_config = llm_config )
15
+ self .openrouter_llm = openrouter_llm
16
+
17
+ def generate_reply (self , messages : List [Dict ]):
18
+ """Override the default generate_reply to use OpenRouterLLM"""
19
+ try :
20
+ response = self .openrouter_llm .generate_response (messages )
21
+ return response
22
+ except Exception as e :
23
+ print (f"Error generating reply: { e } " )
24
+ return None
8
25
9
26
class CodeModifier :
10
- def __init__ (self , temperature = 0.7 ):
27
+ def __init__ (self , api_key : str ):
28
+ # Initialize OpenRouterLLM
29
+ self .llm = OpenRouterLLM (api_key )
30
+
11
31
# Configure the assistant and user proxies
12
- self .assistant = autogen . AssistantAgent (
32
+ self .assistant = CustomAssistantAgent (
13
33
name = "assistant" ,
14
34
llm_config = {
15
- "temperature" : temperature ,
16
- "config_list" : [{ "model" : "gpt-4" }] ,
17
- }
35
+ "temperature" : 0.1 ,
36
+ } ,
37
+ openrouter_llm = self . llm
18
38
)
19
-
39
+
20
40
self .user_proxy = autogen .UserProxyAgent (
21
41
name = "user_proxy" ,
22
42
human_input_mode = "NEVER" ,
@@ -30,9 +50,9 @@ def apply_diff(self, file_path: str, diff_block: str) -> bool:
30
50
content = f .read ()
31
51
32
52
# Parse the diff block
33
- search_pattern = re .search (r'<<<<<<< SEARCH\n(.*?)\n=======\n(.*?)\n>>>>>>> REPLACE' ,
53
+ search_pattern = re .search (r'<<<<<<< SEARCH\n(.*?)\n=======\n(.*?)\n>>>>>>> REPLACE' ,
34
54
diff_block , re .DOTALL )
35
-
55
+
36
56
if not search_pattern :
37
57
return False
38
58
@@ -61,13 +81,13 @@ def run_tests(self, test_file: str) -> Tuple[bool, str]:
61
81
62
82
def process_code (self , problem_description : str , target_files : List [str ], max_iterations : int = 5 ) -> bool :
63
83
"""Main workflow to process code modifications and testing."""
64
-
84
+
65
85
# Create a temporary working directory
66
86
with tempfile .TemporaryDirectory () as temp_dir :
67
87
# Copy target files to workspace
68
88
for file_path in target_files :
69
89
shutil .copy (file_path , temp_dir )
70
-
90
+
71
91
iteration = 0
72
92
while iteration < max_iterations :
73
93
# Generate conversation messages
@@ -76,14 +96,14 @@ def process_code(self, problem_description: str, target_files: List[str], max_it
76
96
"role" : "user" ,
77
97
"content" : f"""
78
98
Problem: { problem_description }
79
-
99
+
80
100
Current code files:
81
101
{ self ._read_files (target_files )}
82
-
102
+
83
103
Please generate:
84
104
1. Code modifications in diff format
85
105
2. Pytest test cases
86
-
106
+
87
107
Use this diff format:
88
108
<<<<<<< SEARCH
89
109
(original code)
@@ -154,15 +174,15 @@ def _extract_test_code(self, response: str) -> str:
154
174
155
175
# Example usage
156
176
if __name__ == "__main__" :
157
- modifier = CodeModifier ()
158
-
177
+ api_key = os .getenv ('OPENROUTER_API_KEY' )
178
+ modifier = CodeModifier (api_key = api_key )
179
+
159
180
problem = """
160
- Fix the implementation of the calculate_average function to handle empty lists
181
+ Fix the implementation of the calculate_average function to handle empty lists
161
182
and return the correct average of numbers.
162
183
"""
163
-
184
+
164
185
target_files = ["math_utils.py" ]
165
-
186
+
166
187
success = modifier .process_code (problem , target_files )
167
188
print (f"Code modification { 'succeeded' if success else 'failed' } " )
168
-
0 commit comments