diff --git a/src/code_executor.py b/src/code_executor.py deleted file mode 100644 index 51351b3..0000000 --- a/src/code_executor.py +++ /dev/null @@ -1,65 +0,0 @@ -import logging -from langchain_core.output_parsers import StrOutputParser -from langchain_experimental.utilities import PythonREPL -from langchain_core.tools import Tool -from langchain_experimental.tools import PythonREPLTool -from project_implementation import ProjectBuilder - -logger = logging.getLogger("BoweiAgent.CodeExecutor") - -class CodeExecutor: - def __init__(self, prompts, llm_client, max_retries=3): - self.llm_client = llm_client - self.prompts = prompts - self.max_retries = max_retries - self.output_parser = StrOutputParser() - - def generate_code(self, user_request: str, context: str = '', bowei_api_docs: str = '') -> str: - logger.info(f"开始生成代码,访问请求:{user_request}") - prompt = self.prompts.code_gen_prompt.format_prompt(user_request=user_request, context=context, bowei_api_docs=bowei_api_docs) - - response = self.llm_client.invoke(prompt.to_messages()) - code = self.output_parser.parse(response) - logger.debug(f"生成的代码内容:\n{code}") - return code - - def fix_code(self, code: str, error: str) -> str: - logger.warning(f"代码执行出错,开始修复。错误信息:{error}") - prompt = self.prompts.code_fix_prompt.format_prompt(code=code, error=error) - response = self.llm_client.invoke(prompt.to_messages()) - fixed_code = self.output_parser.parse(response) - logger.debug(f"修复后的代码内容:\n{fixed_code}") - return fixed_code - - def generate_and_run_code(self, user_request: str, context: str = '', bowei_api_docs: str = '') -> str: - code = self.generate_code(user_request, context, bowei_api_docs) - logger.info("开始执行生成的代码") - - for attempt in range(self.max_retries): - try: - # 自定义命名空间 - namespace = { - "ProjectBuilder": ProjectBuilder, - "project_implementation": __import__("project_implementation"), - "project": __import__("project"), - } - - # 初始化 REPL 并传入命名空间 - python_repl = PythonREPL(globals=namespace) - repl_tool = Tool( - name="python_repl", - description="...", - func=python_repl.run, - ) - - result = repl_tool.func(code.content) # 代码将在自定义命名空间中执行 - logger.info(f"代码执行成功,返回结果长度:{len(result)}") - return result - except Exception as e: - error_msg = str(e) - logger.error(f"代码执行异常,尝试第 {attempt+1} 次修复。异常信息:{error_msg}") - - code = self.fix_code(code, error_msg) - - logger.error(f"代码执行失败,超过最大重试次数 {self.max_retries}") - return f"代码执行失败,超过最大重试次数 {self.max_retries}。\n最后一次错误信息:\n{error_msg}"