Python 如何在超时的情况下同步运行Matplolib服务器端该过程随机挂起

我正在try 复制ChatGPT代码解释器功能，其中LLM通过执行代码按需创建数字.

不幸的是，Matplotlib有20%的时间挂起，我还不明白为什么.

我希望实现:

• 服务器的其余部分不被阻止
• 如果代码太长而无法执行，则有超时

我进行了第一个实现:

import asyncio
import platform
import psutil


TIMEOUT = 5


async def exec_python(code: str) -> str:
    """Execute Python code.

    Args:
        code (str): Python code to execute.

    Returns:
        dict: A dictionary containing the stdout and the stderr from executing the code.
    """
    code = preprocess_code(code)
    stdout = ""
    stderr = ""
    try:
        stdout, stderr = await run_with_timeout(code, TIMEOUT)
    except asyncio.TimeoutError:
        stderr = "Execution timed out."
    return {"stdout": stdout, "stderr": stderr}


async def run_with_timeout(code: str, timeout: int) -> str:
    proc = await run(code)

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        return stdout.decode().strip(), stderr.decode().strip()
    except asyncio.TimeoutError:
        kill_process(proc.pid)
        raise


async def run(code: str):
    timeout_command = 'gtimeout' if platform.system() == 'Darwin' else 'timeout'
    sanity_timeout = TIMEOUT + 1
    command = f'{timeout_command} {sanity_timeout} python -c "{code}"'
    return await asyncio.create_subprocess_shell(
        command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )


def kill_process(pid: int):
    try:
        parent = psutil.Process(pid)
        for child in parent.children(recursive=True):
            child.kill()
        parent.kill()
        print(f"Killing Process {pid} (timed out)")
    except psutil.NoSuchProcess:
        print("Process already killed.")


PLT_OVERRIDE_PREFIX = """
import matplotlib
import asyncio
matplotlib.use('Agg') # non-interactive backend
import matplotlib.pyplot as plt
import io
import base64

def custom_show():
    buf = io.BytesIO()
    plt.gcf().savefig(buf, format='png')
    buf.seek(0)
    image_base64 = base64.b64encode(buf.getvalue()).decode('utf-8')
    print('[BASE_64_IMG]', image_base64)
    plt.clf()

plt.show = custom_show
"""


def preprocess_code(code: str) -> str:
    override_prefix = ""
    code_lines = code.strip().split("\n")
    if not code_lines:
        return code  # Return original code if it's empty
    if "import matplotlib.pyplot as plt" in code:
        override_prefix = PLT_OVERRIDE_PREFIX + "\n"
        code_lines = [
            line for line in code_lines if line != "import matplotlib.pyplot as plt"
        ]

    last_line = code_lines[-1]
    # Check if the last line is already a print statement
    if last_line.strip().startswith("print"):
        return "\n".join(code_lines)

    try:
        compile(last_line, "<string>", "eval")
        # If it's a valid expression, wrap it with print
        code_lines[-1] = f"print({last_line})"
    except SyntaxError:
        # If it's not an expression, check if it's an assignment
        if "=" in last_line:
            variable_name = last_line.split("=")[0].strip()
            code_lines.append(f"print({variable_name})")

    return override_prefix + "\n".join(code_lines)

我已经try 过了，但没有成功:

• ipython而不是Python
• 使用线程而不是进程
• 将图像保存在磁盘上而不是缓冲区上

极其奇怪的是，我无法使用上面的代码重现这个错误.而且它经常在prod和我的机器上看到错误.