React Agent 书写 DockerFile 初试
想尝试使用 Agent 生成 DockerFile，安装指定 PyPI 包，本篇有多处是基于 GPT-5 的指导下完成。
1、Docekrfile 模板
# ====== Dockerfile 模板（非 root，多阶段+小镜像，安装指定 PyPI 包）======
DOCKERFILE_TPL = """\
# syntax=docker/dockerfile:1.7
FROM python:3.11-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
WORKDIR /app
RUN adduser --disabled-password --gecos "" appuser && chown -R appuser:appuser /app
# 依赖层：仅安装目标包（可选额外系统依赖）
FROM base AS deps
# 常见构建依赖（按需补充）
RUN apt-get update && apt-get install -y --no-install-recommends \\
build-essential gcc \\
&& rm -rf /var/lib/apt/lists/*
RUN --mount=type=cache,target=/root/.cache/pip pip install --no-cache-dir {package_spec}
# 运行层
FROM base AS runtime
USER appuser
COPY --from=deps /usr/local /usr/local
WORKDIR /app
{extra_copy}
# 暴露端口（如果是服务型包可以调整）
{expose_line}
# 由 Agent 决定最终 CMD（JSON 形式）
CMD {cmd_json}
"""1️⃣基础层（base）
FROM python:3.11-slim AS base
使用体积小、官方维护的 Python 镜像；slim 兼顾体积与兼容性。
ENV
PYTHONDONTWRITEBYTECODE=1：不生成 .pyc 文件。
PYTHONUNBUFFERED=1：实时日志输出。PIP_NO_CACHE_DIR=1：安装后不留 pip 缓存。
RUN adduser …
创建非 root 用户 appuser，保障容器安全。
WORKDIR /app
统一工作目录，便于部署与维护。
2️⃣依赖层（deps）
FROM base AS deps
单独阶段安装依赖，避免运行镜像被编译文件污染。
RUN apt-get … build-essential gcc
安装编译依赖，仅在构建阶段使用。
RUN –mount=type=cache … pip install {package_spec}
使用 BuildKit 缓存加速安装，{package_spec} 为 PyPI 包名或版本。
3️⃣运行层（runtime）
FROM base AS runtime
轻量运行镜像，不含编译工具。
USER appuser：非 root 启动。
COPY –from=deps /usr/local /usr/local
拷入依赖层安装好的 Python 包与可执行文件。
WORKDIR /app + {extra_copy}
可插入额外文件（如配置、启动脚本）。
{expose_line}：可选 EXPOSE 8000 等端口声明。
CMD {cmd_json}
Agent 生成的启动命令（exec JSON 形式，如 ["python","-m","pkg"]）。
优点：更安全、信号处理正确、无 shell 注入风险。
2、本地执行器
def run_cmd(cmd: List[str], cwd: Path = WORKDIR, timeout: int = 600) -> Tuple[int, str, str]:
# subprocess.Popen(...)启动子进程；stdout/stderr=PIPE 捕获输出；text=True 表示返回字符串（非字节）。
proc = subprocess.Popen(cmd, cwd=str(cwd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
try:
# proc.communicate(timeout=timeout)阻塞等待子进程结束（或直到超时），返回 (out, err)。
out, err = proc.communicate(timeout=timeout)
# 正常返回退出码、标准输出、标准错误。
return proc.returncode, out, err
except subprocess.TimeoutExpired:
# 超时则杀掉子进程，再次 communicate() 把缓冲区读干净。
proc.kill()
out, err = proc.communicate()
# 约定 124 作为超时码（常见约定），若 err 为空则给个 "Timeout"。
return 124, out, err or "Timeout"主要是实现在本地进行指令操作，涉及到的函数只要有
subprocess.Popen：启动子进程。
proc.communicate： 是 subprocess.Popen 对象中最核心的方法之一，它负责与子进程交互、读取输出、等待执行结束。大概流程如下：
的作用相当于执行以下几步的组合操作：
1️⃣向子进程的标准输入 (stdin) 写入数据（如果提供了 input 参数）；
2️⃣从标准输出 (stdout) 和 标准错误 (stderr) 读取全部输出内容；
3️⃣等待子进程结束；
4️⃣返回输出结果（作为一个 (stdout_str, stderr_str) 元组）。
proc.communicate(input=None, timeout=None)
input：可选参数，字符串或字节流。如果想给子进程的标准输入（stdin）传内容，就传它。
timeout：允许的最长等待时间（秒）。超时会抛出 subprocess.TimeoutExpired 异常。
可以安全读取，防止子进程输出很多，导致缓冲区塞满造成死锁。
3、工具函数
# 工具函数
def propose_cmd_local(package: str, previous_error: str) -> Dict[str, Any]:
"""
本地启发式：
- 继续用 candidates 猜 CMD
- 额外：为常见服务型包自动设置 expose 端口
- 额外：基于错误日志，必要时提供 extra_copy（拷贝一个最小示例脚本）
"""
# 1) 先选一个最可能成功的 CMD
candidates: List[List[str]] = []
if not previous_error:
candidates = [
["python", "-m", package],
[package, "--help"],
["python", "-c", f"import {package}; print({package}.__doc__ or 'ok')"]
]
else:
if ("No module named" in previous_error) or ("ModuleNotFoundError" in previous_error):
candidates = [[package, "--help"], ["python", "-m", package]]
elif ("command not found" in previous_error) or ("executable file not found" in previous_error):
candidates = [["python", "-m", package], [package]]
else:
candidates = [["python", "-m", package], [package, "--help"]]
cmd = candidates[0]
# 2) 针对常见“服务型包”设置默认端口（可按需扩充）
default_ports = {
"uvicorn": 8000,
"fastapi": 8000,
"starlette": 8000,
"flask": 5000,
"streamlit": 8501,
"gradio": 7860,
"mlflow": 5000,
"tensorboard": 6006,
"jupyter": 8888,
"jupyterlab": 8888,
"ray": 8265,
}
expose = default_ports.get(package, None)
# 3) 基于错误日志，给出一个最小可跑 demo 的 extra_copy（可选）
extra_copy = ""
# 常见“找不到脚本/文件”的情况，给一个最小示例
if previous_error:
err = previous_error.lower()
# 如果提示找不到 app.py / main.py / hello.py，就提供一个轻量 demo
if any(x in err for x in ["no such file or directory", "file not found", "can't open file"]):
if package in ("uvicorn", "fastapi", "starlette"):
# 给一个极简 FastAPI 应用，并让 CMD 使用它（通常 LLM 会改 CMD，这里仅兜底）
extra_copy = (
'COPY <<EOF /app/app.py\n'
'from fastapi import FastAPI\n'
'app = FastAPI()\n'
'@app.get("/")\n'
'def root():\n'
' return {"msg": "ok"}\n'
'EOF\n'
)
# 如果当前 cmd 看起来没有指向 app:app，可以保守不改 CMD，交给下一轮 LLM 调整
# 也可以在这里直接强改为 uvicorn 启动（按需开启）：
# cmd = ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", str(expose or 8000)]
expose = expose or 8000
elif package == "streamlit":
extra_copy = (
'COPY <<EOF /app/hello.py\n'
'import streamlit as st\n'
'st.title("Hello Streamlit")\n'
'st.write("It works!")\n'
'EOF\n'
)
expose = expose or 8501
# 同理，这里也可考虑改 CMD 为 ["streamlit","run","hello.py","--server.address","0.0.0.0"]
elif package == "gradio":
extra_copy = (
'COPY <<EOF /app/app.py\n'
'import gradio as gr\n'
'def echo(x):\n'
' return x\n'
'demo = gr.Interface(fn=echo, inputs="text", outputs="text")\n'
'if __name__ == "__main__":\n'
' demo.launch(server_name="0.0.0.0", server_port=7860)\n'
'EOF\n'
)
expose = expose or 7860
return {"cmd": cmd, "expose": expose, "extra_copy": extra_copy}这里选择了三条指令作为尝试安装package
1）、python -m pkg：直接执行这个pkg，如过没有main则会失败
2）、pkg --help：尝试运行该包的 CLI（命令行界面），查看它的帮助信息。
3）、["python", "-c", f"import {package}; print({package}.__doc__ or 'ok')"]：启动 Python 解释器；尝试 import <package>；如果能导入成功，打印包的 __doc__ 文档字符串（或“ok”）；退出。
用 -c 参数执行字符串形式的脚本，也就是后面import那些
检查包是否能被导入（验证安装成功）；确认模块名是否与 PyPI 名一致（有的包安装名和导入名不同，例如 bs4 vs beautifulsoup4）；输出结果可用于日志分析。
__doc__ 是模块的文档字符串（模块顶端三引号注释），读取它不影响包执行，输出简短，不会阻塞或报错，如果没有文档，就打印 'ok' 说明导入成功。
脚本的启发式策略是：
-
首先试模块执行（python -m pkg）；
-
不行再试命令行；
-
最后再试最简单的 import 验证。
4）另外两个参数
|
expose |
检测出 Web 服务包时 |
在 Dockerfile 中暴露端口 |
|
extra_copy |
包运行需要额外文件时 |
在 Dockerfile 中添加 COPY 指令 |
这里是在本地进行的简单的修改，没有把参数上传让 LLM 修改。
4、docker 创建与运行
def docker_build(image_tag: str, dockerfile_path: Path) -> Tuple[bool, str]:
# cwd=dockerfile_path.parent,工作目录为 Dockerfile 所在目录
code, out, err = run_cmd(["docker", "build", "-t", image_tag, "-f", str(dockerfile_path), "."], cwd=dockerfile_path.parent, timeout=1800)
ok = (code == 0)
# code=0 时则成功
return ok, out + "\n" + err
def docker_run(image_tag: str, cmd: List[str] = None, timeout: int = 60) -> Tuple[int, str]:
base = ["docker", "run", "--rm"]
if cmd:
base += [image_tag] + cmd
else:
base += [image_tag]
code, out, err = run_cmd(base, timeout=timeout)
return code, out + "\n" + err1）创建Docker的函数
将Docker创建指令与地址传回run_cmd，作为subprocess.Popen的参数，启动子进程。
· Docker 的命令是：docker build -t <镜像名> -f <Dockerfile路径> .
· path.parent的作用：
假设 dockerfile_path = Path("/tmp/pypi_streamlit_1234/Dockerfile")
则 dockerfile_path.parent == Path("/tmp/pypi_streamlit_1234")
2）运行 Docker 的函数
根据是否传入了 CMD 有两种结果
|
传入 cmd |
实际执行命令 |
解释 |
|---|---|---|
|
cmd=None（默认） |
docker run --rm <image_tag> |
不加额外参数，直接运行镜像的默认启动命令（即 Dockerfile 里的 CMD） |
|
cmd=["python","-m","xxx"] |
docker run --rm <image_tag> python -m xxx |
临时覆盖 Dockerfile 里的 CMD，执行新的命令 |
5、System Prompt 设计
SYSTEM_PROMPT = """You are a DevOps agent that generates runnable Dockerfiles for PyPI packages.
Rules:
- Prefer EXEC JSON CMD array, no shell forms.
- Avoid root at runtime (already handled by base image).
- If the package exposes a console_script, use it with --help (non-blocking).
- Otherwise try: python -m <package>.
- If previous run failed, infer a better command from the error/logs.
Return only by calling the tool 'propose_cmd' with {"package": "...", "previous_error": "..."}.
"""1）Prefer EXEC JSON CMD array, no shell forms.
Exec 形式就是 JSON 数组，CMD ["python", "-m", "http.server", "8000"]，相当于执行了python -m http.server 8000。
Exec 形式特点：直接由内核启动程序，不经过shell(/bin/sh)，每个参数都是独立字符串，不会被 shell 解析，信号(SIGTERM、SIGINT)会正确传递到子进程，安全、可靠、可控，是官方推荐形式。
这样设计可以防止在 shell 命令中注入字符串展开风险。
2) Avoid root at runtime (already handled by base image).
运行期避免 root（最小权限）。虽然基础镜像阶段已处理（创建 appuser、USER appuser），但这里再次强调，防止模型建议任何需要升权/写系统目录的做法（如 sudo、写 /root）。
3) If the package exposes a console_script, use it with --help (non-blocking).
优先走 console_script + --help 的短命令。
console_script 是在 Python 包的 setup.py 或 pyproject.toml 中，很多项目会定义一个 “入口点”：
# setup.py
entry_points={
'console_scripts': [
'black = black:patched_main',
'pytest = pytest:main',
'uvicorn = uvicorn.main:main',
],
}这表示当用户 pip install black 后，系统会自动生成一个命令行可执行文件：
-
/usr/local/bin/black
-
/usr/local/bin/pytest
-
/usr/local/bin/uvicorn
这些文件就是 “console_script”。
只需在命令行输入：
black --help
pytest --help
uvicorn --help
就能执行对应包里的主函数。
· Otherwise try: python -m <package>.
退而求其次：模块形式执行。
原因：许多包没有 CLI，但支持 python -m 包名 作为入口（如标准库 http.server 的用法）；
与上一条的关系：形成“CLI 优先、模块次之”的启发式，覆盖面广、尝试成本低。
4） If previous run failed, infer a better command from the error/logs.
如果之前的运行失败，则从错误/日志中推断出更好的命令。
5）Return only by calling the tool 'propose_cmd' with {"package": "...", "previous_error": "..."} .
必须用 Function Calling 工具返回（结构化输出），且仅能调用这个工具，并带上两个参数。
6、主函数，循环逻辑
def main(package: str):
work = Path(tempfile.mkdtemp(prefix=f"pypi_{package}_"))
dockerfile_path = work / "Dockerfile"
print(f"📦 Target PyPI package: {package}")
print(f"🗂 Workdir: {work}")
previous_error = ""
for step in range(1, MAX_STEPS + 1):
print(f"\n===== Attempt {step}/{MAX_STEPS} =====")
# 1) 让 LLM（或本地启发式）给一版 CMD（和可选 expose）
plan = plan_cmd_with_llm(package, previous_error)
# 兜底：如果计划不含 cmd，用模块执行。
cmd = plan.get("cmd", ["python", "-m", package])
expose = plan.get("expose", None)
print(f"🔧 Proposed CMD: {cmd} (expose={expose})")
# 2) 渲染 Dockerfile
dockerfile_text = make_dockerfile(package, cmd, expose=expose)
dockerfile_path.write_text(dockerfile_text, encoding="utf-8")
print("📝 Dockerfile generated.")
# 3) 构建镜像
image_tag = f"{IMAGE_PREFIX}:{package}-{step}"
print(f"🏗 docker build -t {image_tag}")
ok, build_log = docker_build(image_tag, dockerfile_path)
if not ok:
# 构建失败，多数是系统依赖或包名错误；把日志反馈回去迭代
print("❌ Build failed. Will refine.")
previous_error = f"[BUILD FAIL]\n{tail(build_log, 1000)}"
continue
# 4) 运行容器（若 CMD 是阻塞型长跑服务，建议先尝试 --help 或短命令）
print(f"🚀 docker run {image_tag}")
code, run_log = docker_run(image_tag, timeout=45)
if code == 0:
print("✅ Run succeeded!")
print("====== OUTPUT ======")
print(tail(run_log, 1200))
print("====================")
print(f"🎉 Success with image: {image_tag}")
return
else:
print("⚠️ Run failed, try to refine CMD.")
previous_error = f"[RUN FAIL code={code}]\n{tail(run_log, 1500)}"
print("\n⛔ Reached max attempts. Last error/logs:")
print(previous_error)1）cmd = plan.get("cmd", ["python", "-m", package])
expose = plan.get("expose", None)
plan 是返回的一个字典
plan = {"cmd": ["pytest", "--help"], "extra_copy": ""}
plan["cmd"] ——> ["pytest", "--help"] 但如果 cmd 这个键不存在，则会报错KeyError: 'cmd'
所以选择更安全的方式：通过 get 直接获取
plan.get("cmd", ["python", "-m", package]) ——> 如果 cmd 不存在则返回默认值["python", "-m", package]
2）work = Path(tempfile.mkdtemp(prefix=f"pypi_{package}_"))
· tempfile 是 Python 标准库中专门用来创建临时文件或临时目录的模块。mkdtemp() 会创建一个 唯一的临时文件夹，返回该文件夹的绝对路径字符串。
临时目录会自动放在系统的默认临时目录下（例如 macOS/Linux /tmp/，Windows 通常是 %TEMP%）。
import tempfile
tempfile.mkdtemp(prefix="pypi_numpy_")返回值可能为：'/tmp/pypi_numpy_abc1234'
"abc1234" 是系统自动生成的随机后缀，用来避免重复。
· prefix=f"pypi_{package}_"：用了 f-string 格式化语法，prefix 指定临时文件夹名称前缀。
如果 package 是 "fastapi"，则创建的目录形如：/tmp/pypi_fastapi_h8s7bq9z
· 用 Path() 把它转换成一个 Path 对象，Path 对象有很多内置方法。方便后续的优雅书写
dockerfile_path = work / "Dockerfile" 直接进行拼接
dockerfile_path.write_text(dockerfile_text, encoding="utf-8") .write_text() 写入文本到文件
7、代码附录
#!/usr/bin/env python3
# autodocker_pypi_agent.py
# 功能：给定一个 PyPI 包名，自动生成 Dockerfile，构建并运行；
# 若运行失败，LLM 根据错误日志改写 CMD，重复尝试，直到成功或达到最大次数。
##### 大部分位置都已经写定，LLM 只负责少量关键槽位（CMD/EXPOSE）→ 降低幻觉面
import os, json, subprocess, shlex, sys, tempfile, re, textwrap, time
from pathlib import Path
from typing import Any, Dict, List, Tuple
from openai import OpenAI
from openai import APIStatusError
# ====== 配置 ======
MODEL = os.getenv("MODEL_NAME", "deepseek-chat") # 也可用 deepseek-chat
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1")
API_KEY = os.getenv("DEEPSEEK_API_KEY")
MAX_STEPS = int(os.getenv("MAX_STEPS", "10"))
WORKDIR = Path.cwd().resolve()
IMAGE_PREFIX = "pypi-auto"
if not API_KEY:
print("❌ 请先设置 DEEPSEEK_API_KEY")
sys.exit(1)
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
# ====== Dockerfile 模板（非 root，多阶段+小镜像，安装指定 PyPI 包）======
DOCKERFILE_TPL = """\
# syntax=docker/dockerfile:1.7
FROM python:3.11-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
WORKDIR /app
RUN adduser --disabled-password --gecos "" appuser && chown -R appuser:appuser /app
# 依赖层：仅安装目标包（可选额外系统依赖）
FROM base AS deps
# 常见构建依赖（按需补充）
RUN apt-get update && apt-get install -y --no-install-recommends \\
build-essential gcc \\
&& rm -rf /var/lib/apt/lists/*
RUN --mount=type=cache,target=/root/.cache/pip pip install --no-cache-dir {package_spec}
# 运行层
FROM base AS runtime
USER appuser
COPY --from=deps /usr/local /usr/local
WORKDIR /app
{extra_copy}
# 暴露端口（如果是服务型包可以调整）
{expose_line}
# 由 Agent 决定最终 CMD（JSON 形式）
CMD {cmd_json}
"""
# ====== Tools（供 LLM 调用）的 schema ======
TOOLS_SPEC = [
{
"type": "function",
"function": {
"name": "propose_cmd",
"description": "Propose a CMD (as JSON array) to run the given PyPI package inside container.",
"parameters": {
"type": "object",
"properties": {
"package": {"type": "string", "description": "PyPI package name"},
"previous_error": {"type": "string", "description": "stderr/stdout logs from previous run, if any"},
},
"required": ["package"]
}
}
}
]
# ====== 本地实现的“工具”分发（这里只需要 LLM 规划 CMD，本地不执行外部命令）======
def dispatch_tool_call(name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
if name == "propose_cmd":
return propose_cmd_local(arguments.get("package", ""), arguments.get("previous_error", ""))
return {"cmd": ["python", "-m", arguments.get("package", "")], "expose": None, "extra_copy": ""}
# 工具函数
def propose_cmd_local(package: str, previous_error: str) -> Dict[str, Any]:
"""
本地启发式：
- 继续用 candidates 猜 CMD
- 额外：为常见服务型包自动设置 expose 端口
- 额外：基于错误日志，必要时提供 extra_copy（拷贝一个最小示例脚本）
"""
# 1) 先选一个最可能成功的 CMD
candidates: List[List[str]] = []
if not previous_error:
candidates = [
["python", "-m", package],
[package, "--help"],
["python", "-c", f"import {package}; print({package}.__doc__ or 'ok')"]
]
else:
if ("No module named" in previous_error) or ("ModuleNotFoundError" in previous_error):
candidates = [[package, "--help"], ["python", "-m", package]]
elif ("command not found" in previous_error) or ("executable file not found" in previous_error):
candidates = [["python", "-m", package], [package]]
else:
candidates = [["python", "-m", package], [package, "--help"]]
cmd = candidates[0]
# 2) 针对常见“服务型包”设置默认端口（可按需扩充）
default_ports = {
"uvicorn": 8000,
"fastapi": 8000,
"starlette": 8000,
"flask": 5000,
"streamlit": 8501,
"gradio": 7860,
"mlflow": 5000,
"tensorboard": 6006,
"jupyter": 8888,
"jupyterlab": 8888,
"ray": 8265,
}
expose = default_ports.get(package, None)
# 3) 基于错误日志，给出一个最小可跑 demo 的 extra_copy（可选）
extra_copy = ""
# 常见“找不到脚本/文件”的情况，给一个最小示例
if previous_error:
err = previous_error.lower()
# 如果提示找不到 app.py / main.py / hello.py，就提供一个轻量 demo
if any(x in err for x in ["no such file or directory", "file not found", "can't open file"]):
if package in ("uvicorn", "fastapi", "starlette"):
# 给一个极简 FastAPI 应用，并让 CMD 使用它（通常 LLM 会改 CMD，这里仅兜底）
extra_copy = (
'COPY <<EOF /app/app.py\n'
'from fastapi import FastAPI\n'
'app = FastAPI()\n'
'@app.get("/")\n'
'def root():\n'
' return {"msg": "ok"}\n'
'EOF\n'
)
# 如果当前 cmd 看起来没有指向 app:app，可以保守不改 CMD，交给下一轮 LLM 调整
# 也可以在这里直接强改为 uvicorn 启动（按需开启）：
# cmd = ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", str(expose or 8000)]
expose = expose or 8000
elif package == "streamlit":
extra_copy = (
'COPY <<EOF /app/hello.py\n'
'import streamlit as st\n'
'st.title("Hello Streamlit")\n'
'st.write("It works!")\n'
'EOF\n'
)
expose = expose or 8501
# 同理，这里也可考虑改 CMD 为 ["streamlit","run","hello.py","--server.address","0.0.0.0"]
elif package == "gradio":
extra_copy = (
'COPY <<EOF /app/app.py\n'
'import gradio as gr\n'
'def echo(x):\n'
' return x\n'
'demo = gr.Interface(fn=echo, inputs="text", outputs="text")\n'
'if __name__ == "__main__":\n'
' demo.launch(server_name="0.0.0.0", server_port=7860)\n'
'EOF\n'
)
expose = expose or 7860
return {"cmd": cmd, "expose": expose, "extra_copy": extra_copy}
# ====== 本地执行器 ======
def run_cmd(cmd: List[str], cwd: Path = WORKDIR, timeout: int = 600) -> Tuple[int, str, str]:
# subprocess.Popen(...)启动子进程；stdout/stderr=PIPE 捕获输出；text=True 表示返回字符串（非字节）。
proc = subprocess.Popen(cmd, cwd=str(cwd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
try:
# proc.communicate(timeout=timeout)阻塞等待子进程结束（或直到超时），返回 (out, err)。
out, err = proc.communicate(timeout=timeout)
# 正常返回退出码、标准输出、标准错误。
return proc.returncode, out, err
except subprocess.TimeoutExpired:
# 超时则杀掉子进程，再次 communicate() 把缓冲区读干净。
proc.kill()
out, err = proc.communicate()
# 约定 124 作为超时码（常见约定），若 err 为空则给个 "Timeout"。
return 124, out, err or "Timeout"
def docker_build(image_tag: str, dockerfile_path: Path) -> Tuple[bool, str]:
# cwd=dockerfile_path.parent,工作目录为 Dockerfile 所在目录
code, out, err = run_cmd(["docker", "build", "-t", image_tag, "-f", str(dockerfile_path), "."], cwd=dockerfile_path.parent, timeout=1800)
ok = (code == 0)
# code=0 时则成功
return ok, out + "\n" + err
def docker_run(image_tag: str, cmd: List[str] = None, timeout: int = 60) -> Tuple[int, str]:
base = ["docker", "run", "--rm"]
if cmd:
base += [image_tag] + cmd
else:
base += [image_tag]
code, out, err = run_cmd(base, timeout=timeout)
return code, out + "\n" + err
# ====== 通过 LLM 规划或修正 CMD（带 function calling）======
# If previous run failed, infer a better command from the error/logs. 如果之前的运行失败，则从错误/日志中推断出更好的命令。
SYSTEM_PROMPT = """You are a DevOps agent that generates runnable Dockerfiles for PyPI packages.
Rules:
- Prefer EXEC JSON CMD array, no shell forms.
- Avoid root at runtime (already handled by base image).
- If the package exposes a console_script, use it with --help (non-blocking).
- Otherwise try: python -m <package>.
- If previous run failed, infer a better command from the error/logs.
Return only by calling the tool 'propose_cmd' with {"package": "...", "previous_error": "..."}.
"""
def plan_cmd_with_llm(package: str, previous_error: str) -> Dict[str, Any]:
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Package: {package}\nPrevious error/logs:\n{previous_error or '(none)'}"}
]
try:
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS_SPEC,
tool_choice="auto",
temperature=0.2,
)
except APIStatusError as e:
if e.status_code == 402:
print("❌ API 402 Insufficient Balance：用本地启发式 fallback。")
return propose_cmd_local(package, previous_error)
raise
msg = resp.choices[0].message
if msg.tool_calls:
# 只取第一条 tool_call
tc = msg.tool_calls[0]
name = tc.function.name
args = {}
try:
args = json.loads(tc.function.arguments or "{}")
except Exception:
args = {}
result = dispatch_tool_call(name, args)
return result
else:
# 模型没按要求走 tool，fallback
return propose_cmd_local(package, previous_error)
# ====== 生成 Dockerfile 文本 ======
def make_dockerfile(package: str, cmd: List[str], expose: int = None, extra_copy: str = "") -> str:
# 如果传了端口号（如 8000），生成一行 EXPOSE 8000；否则为空字符串。
expose_line = f"EXPOSE {expose}" if expose else ""
extra_copy = extra_copy or ""
return DOCKERFILE_TPL.format(
package_spec=shlex.quote(package),
expose_line=expose_line,
cmd_json=json.dumps(cmd),
extra_copy=extra_copy
)
# ====== 主流程 ======
def main(package: str):
work = Path(tempfile.mkdtemp(prefix=f"pypi_{package}_"))
dockerfile_path = work / "Dockerfile"
print(f"📦 Target PyPI package: {package}")
print(f"🗂 Workdir: {work}")
previous_error = ""
for step in range(1, MAX_STEPS + 1):
print(f"\n===== Attempt {step}/{MAX_STEPS} =====")
# 1) 让 LLM（或本地启发式）给一版 CMD（和可选 expose）
plan = plan_cmd_with_llm(package, previous_error)
# 兜底：如果计划不含 cmd，用模块执行。
cmd = plan.get("cmd", ["python", "-m", package])
expose = plan.get("expose", None)
print(f"🔧 Proposed CMD: {cmd} (expose={expose})")
# 2) 渲染 Dockerfile
dockerfile_text = make_dockerfile(package, cmd, expose=expose)
dockerfile_path.write_text(dockerfile_text, encoding="utf-8")
print("📝 Dockerfile generated.")
# 3) 构建镜像
image_tag = f"{IMAGE_PREFIX}:{package}-{step}"
print(f"🏗 docker build -t {image_tag}")
ok, build_log = docker_build(image_tag, dockerfile_path)
if not ok:
# 构建失败，多数是系统依赖或包名错误；把日志反馈回去迭代
print("❌ Build failed. Will refine.")
previous_error = f"[BUILD FAIL]\n{tail(build_log, 1000)}"
continue
# 4) 运行容器（若 CMD 是阻塞型长跑服务，建议先尝试 --help 或短命令）
print(f"🚀 docker run {image_tag}")
code, run_log = docker_run(image_tag, timeout=45)
if code == 0:
print("✅ Run succeeded!")
print("====== OUTPUT ======")
print(tail(run_log, 1200))
print("====================")
print(f"🎉 Success with image: {image_tag}")
return
else:
print("⚠️ Run failed, try to refine CMD.")
previous_error = f"[RUN FAIL code={code}]\n{tail(run_log, 1500)}"
print("\n⛔ Reached max attempts. Last error/logs:")
print(previous_error)
def tail(s: str, n: int) -> str:
# 取字符串后 n 字符，防止日志过长
return s[-n:] if len(s) > n else s
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python autodocker_pypi_agent.py <pypi_package_name>")
sys.exit(1)
pkg = sys.argv[1].strip()
main(pkg)
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
13
0- 0
已为社区贡献2条内容
所有评论(0)