删除 src/code_executor.py

This commit is contained in:
2025-06-24 10:18:59 +08:00
parent 5a1c74b356
commit de04e1ca9b
-65
View File
@@ -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}"