refactor: CodeParser.parse_code

metagpt/actions/debug_error.py

@@ -70,6 +70,6 @@ class DebugError(Action):
         prompt = PROMPT_TEMPLATE.format(code=code_doc.content, test_code=test_doc.content, logs=output_detail.stderr)
 
         rsp = await self._aask(prompt)
-        code = CodeParser.parse_code(block="", text=rsp)
+        code = CodeParser.parse_code(text=rsp)
 
         return code

metagpt/actions/di/write_analysis_code.py

@@ -30,7 +30,7 @@ class WriteAnalysisCode(Action):
         )
 
         rsp = await self._aask(reflection_prompt, system_msgs=[REFLECTION_SYSTEM_MSG])
-        reflection = json.loads(CodeParser.parse_code(block=None, text=rsp))
+        reflection = json.loads(CodeParser.parse_code(text=rsp))
 
         return reflection["improved_impl"]
 
@@ -57,7 +57,7 @@ class WriteAnalysisCode(Action):
             code = await self._debug_with_reflection(context=context, working_memory=working_memory)
         else:
             rsp = await self.llm.aask(context, system_msgs=[INTERPRETER_SYSTEM_MSG], **kwargs)
-            code = CodeParser.parse_code(block=None, text=rsp)
+            code = CodeParser.parse_code(text=rsp)
 
         return code
 
@@ -69,5 +69,5 @@ class CheckData(Action):
         code_written = "\n\n".join(code_written)
         prompt = CHECK_DATA_PROMPT.format(code_written=code_written)
         rsp = await self._aask(prompt)
-        code = CodeParser.parse_code(block=None, text=rsp)
+        code = CodeParser.parse_code(text=rsp)
         return code

metagpt/actions/di/write_plan.py

@@ -47,7 +47,7 @@ class WritePlan(Action):
             context="\n".join([str(ct) for ct in context]), max_tasks=max_tasks, task_type_desc=task_type_desc
         )
         rsp = await self._aask(prompt)
-        rsp = CodeParser.parse_code(block=None, text=rsp)
+        rsp = CodeParser.parse_code(text=rsp)
         return rsp
 
 

metagpt/actions/write_code.py

@@ -87,7 +87,7 @@ class WriteCode(Action):
     @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
     async def write_code(self, prompt) -> str:
         code_rsp = await self._aask(prompt)
-        code = CodeParser.parse_code(block="", text=code_rsp)
+        code = CodeParser.parse_code(text=code_rsp)
         return code
 
     async def run(self, *args, **kwargs) -> CodingContext:

metagpt/actions/write_code_review.py

@@ -132,7 +132,7 @@ class WriteCodeReview(Action):
         # if LBTM, rewrite code
         rewrite_prompt = f"{context_prompt}\n{cr_rsp}\n{REWRITE_CODE_TEMPLATE.format(filename=filename)}"
         code_rsp = await self._aask(rewrite_prompt)
-        code = CodeParser.parse_code(block="", text=code_rsp)
+        code = CodeParser.parse_code(text=code_rsp)
         return result, code
 
     async def run(self, *args, **kwargs) -> CodingContext:

metagpt/actions/write_test.py

@@ -45,7 +45,7 @@ class WriteTest(Action):
         code_rsp = await self._aask(prompt)
 
         try:
-            code = CodeParser.parse_code(block="", text=code_rsp)
+            code = CodeParser.parse_code(text=code_rsp)
         except Exception:
             # Handle the exception if needed
             logger.error(f"Can't parse the code: {code_rsp}")

metagpt/provider/openai_api.py

@@ -220,7 +220,7 @@ class OpenAILLM(BaseLLM):
             # The response content is `code``, but it appears in the content instead of the arguments.
             code_formats = "```"
             if message.content.startswith(code_formats) and message.content.endswith(code_formats):
-                code = CodeParser.parse_code(None, message.content)
+                code = CodeParser.parse_code(text=message.content)
                 return {"language": "python", "code": code}
             # reponse is message
             return {"language": "markdown", "code": self.get_choice_text(rsp)}

metagpt/roles/di/data_interpreter.py

@@ -73,7 +73,7 @@ class DataInterpreter(Role):
 
         prompt = REACT_THINK_PROMPT.format(user_requirement=user_requirement, context=context)
         rsp = await self.llm.aask(prompt)
-        rsp_dict = json.loads(CodeParser.parse_code(block=None, text=rsp))
+        rsp_dict = json.loads(CodeParser.parse_code(text=rsp))
         self.working_memory.add(Message(content=rsp_dict["thoughts"], role="assistant"))
         need_action = rsp_dict["state"]
         self._set_state(0) if need_action else self._set_state(-1)

metagpt/roles/di/team_leader.py

@@ -93,7 +93,7 @@ class TeamLeader(Role):
         context = self.llm.format_msg(self.get_memories(k=10) + [Message(content=prompt, role="user")])
 
         rsp = await self.llm.aask(context)
-        self.commands = json.loads(CodeParser.parse_code(block=None, text=rsp))
+        self.commands = json.loads(CodeParser.parse_code(text=rsp))
         self.rc.memory.add(Message(content=rsp, role="assistant"))
 
         return True

metagpt/schema.py

@@ -18,7 +18,6 @@ from __future__ import annotations
 import asyncio
 import json
 import os.path
-import re
 import uuid
 from abc import ABC
 from asyncio import Queue, QueueEmpty, wait_for
@@ -49,7 +48,7 @@ from metagpt.const import (
 )
 from metagpt.logs import ToolLogItem, log_tool_output, logger
 from metagpt.repo_parser import DotClassInfo
-from metagpt.utils.common import any_to_str, any_to_str_set, import_class
+from metagpt.utils.common import CodeParser, any_to_str, any_to_str_set, import_class
 from metagpt.utils.exceptions import handle_exception
 from metagpt.utils.serialize import (
     actionoutout_schema_to_mapping,
@@ -337,11 +336,8 @@ class Message(BaseModel):
         return_format += '- a "reason" key explaining why;\n'
         instructions = ['Lists all the resources contained in the "Original Requirement".', return_format]
         rsp = await llm.aask(msg=content, system_msgs=instructions)
-        pattern = r"```json\s*({[\s\S]*?})\s*```"
-        matches = re.findall(pattern, rsp)
-        if not matches:
-            return {}
-        m = json.loads(matches[0])
+        json_data = CodeParser.parse_code(text=rsp, lang="json")
+        m = json.loads(json_data)
         m["resources"] = [Resource(**i) for i in m.get("resources", [])]
         return m
 

metagpt/strategy/tot.py

@@ -62,7 +62,7 @@ class ThoughtSolverBase(BaseModel):
             current_state=current_state, **{"n_generate_sample": self.config.n_generate_sample}
         )
         rsp = await self.llm.aask(msg=state_prompt + "\n" + OUTPUT_FORMAT)
-        thoughts = CodeParser.parse_code(block="", text=rsp)
+        thoughts = CodeParser.parse_code(text=rsp)
         thoughts = eval(thoughts)
         # fixme 避免不跟随，生成过多nodes
         # valid_thoughts = [_node for idx, _node in enumerate(thoughts) if idx < self.n_generate_sample]

metagpt/tools/libs/gpt_v_generator.py

@@ -89,7 +89,7 @@ class GPTvGenerator:
         webpages_path.mkdir(parents=True, exist_ok=True)
 
         index_path = webpages_path / "index.html"
-        index_path.write_text(CodeParser.parse_code(block=None, text=webpages, lang="html"))
+        index_path.write_text(CodeParser.parse_code(text=webpages, lang="html"))
 
         extract_and_save_code(folder=webpages_path, text=webpages, pattern="styles?.css", language="css")
 
@@ -102,5 +102,5 @@ def extract_and_save_code(folder, text, pattern, language):
     word = re.search(pattern, text)
     if word:
         path = folder / word.group(0)
-        code = CodeParser.parse_code(block=None, text=text, lang=language)
+        code = CodeParser.parse_code(text=text, lang=language)
         path.write_text(code, encoding="utf-8")

metagpt/tools/tool_recommend.py

@@ -132,7 +132,7 @@ class ToolRecommender(BaseModel):
             topk=topk,
         )
         rsp = await LLM().aask(prompt, stream=False)
-        rsp = CodeParser.parse_code(block=None, text=rsp)
+        rsp = CodeParser.parse_code(text=rsp)
         ranked_tools = json.loads(rsp)
 
         valid_tools = validate_tool_names(ranked_tools)

metagpt/utils/common.py

@@ -271,7 +271,7 @@ class CodeParser:
         return block_dict
 
     @classmethod
-    def parse_code(cls, block: Optional[str], text: str, lang: str = "") -> str:
+    def parse_code(cls, text: str, lang: str = "", block: Optional[str] = None) -> str:
         if block:
             text = cls.parse_block(block, text)
         pattern = rf"```{lang}.*?\s+(.*?)```"
@@ -287,7 +287,7 @@ class CodeParser:
 
     @classmethod
     def parse_str(cls, block: str, text: str, lang: str = ""):
-        code = cls.parse_code(block, text, lang)
+        code = cls.parse_code(block=block, text=text, lang=lang)
         code = code.split("=")[-1]
         code = code.strip().strip("'").strip('"')
         return code
@@ -295,7 +295,7 @@ class CodeParser:
     @classmethod
     def parse_file_list(cls, block: str, text: str, lang: str = "") -> list[str]:
         # Regular expression pattern to find the tasks list.
-        code = cls.parse_code(block, text, lang)
+        code = cls.parse_code(block=block, text=text, lang=lang)
         # print(code)
         pattern = r"\s*(.*=.*)?(\[.*\])"
 

tests/metagpt/actions/test_write_code.py

@@ -110,7 +110,7 @@ async def test_write_refined_code(context, git_dir):
 
     # old_workspace contains the legacy code
     await context.repo.with_src_path(context.repo.old_workspace).srcs.save(
-        filename="game.py", content=CodeParser.parse_code(block="", text=REFINED_CODE_INPUT_SAMPLE)
+        filename="game.py", content=CodeParser.parse_code(text=REFINED_CODE_INPUT_SAMPLE)
     )
 
     ccontext = CodingContext(

tests/metagpt/actions/test_write_code_plan_and_change_an.py

@@ -45,7 +45,7 @@ async def test_write_code_plan_and_change_an(mocker, context, git_dir):
     await context.repo.docs.task.save(filename="2.json", content=json.dumps(REFINED_TASK_JSON))
 
     await context.repo.with_src_path(context.repo.old_workspace).srcs.save(
-        filename="game.py", content=CodeParser.parse_code(block="", text=REFINED_CODE_INPUT_SAMPLE)
+        filename="game.py", content=CodeParser.parse_code(text=REFINED_CODE_INPUT_SAMPLE)
     )
 
     root = ActionNode.from_children(

tests/metagpt/roles/test_engineer.py

@@ -91,7 +91,7 @@ target_code = """task_list = [
 
 
 def test_parse_code():
-    code = CodeParser.parse_code("Task list", TASKS, lang="python")
+    code = CodeParser.parse_code(block="Task list", text=TASKS, lang="python")
     logger.info(code)
     assert isinstance(code, str)
     assert target_code == code

tests/metagpt/utils/test_code_parser.py

@@ -119,7 +119,7 @@ class TestCodeParser:
         assert "game.py" in result
 
     def test_parse_code(self, parser, text):
-        result = parser.parse_code("Task list", text, "python")
+        result = parser.parse_code(block="Task list", text=text, lang="python")
         print(result)
         assert "game.py" in result