如何用 DockerCommandLineCodeExecutor 让 AutoGen 智能体在 Docker 容器中执行代码？
【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen
在 AutoGen（autogen）里开发智能体应用时，一个常见需求是让 LLM 生成代码并实际执行：用LocalCommandLineCodeExecutor直接在宿主机上跑 LLM 生成的代码风险较高，官方文档明确不推荐。推荐做法是使用DockerCommandLineCodeExecutor——它会创建一个 Docker 容器，把所有命令都放进这个容器里执行，宿主机环境不受影响。
本文基于仓库中两篇官方文档 Command Line Code Executors 和 Code Execution，走一遍完整路径：安装依赖、单独用执行器在容器里跑一段代码、再把「写代码的 Assistant 智能体 + 执行代码的 Executor 智能体」接入 AutoGen 运行时。适用环境：Python 3.10 或更高版本，宿主机已安装并正在运行 Docker。
准备工作：安装包并确认 Docker 可用
建议先建一个虚拟环境，把 AutoGen 的依赖与系统隔离：
python3 -m venv .venv source .venv/bin/activate（Windows 下激活命令是.venv\Scripts\activate.bat。）
然后按 安装文档 安装核心包与扩展：
pip install "autogen-core" pip install "autogen-ext[docker]"autogen-core提供智能体运行时和CodeBlock等基础类型，要求 Python 3.10 或更高。autogen-ext的dockerextra 是DockerCommandLineCodeExecutor的前置条件，见 扩展安装文档。
如果要用 OpenAI 模型客户端驱动智能体，还需要：
pip install "autogen-ext[openai]"最后确认 Docker 已安装并且守护进程正在运行。DockerCommandLineCodeExecutor通过dockerPython 库连接本地 Docker；连不上时源码会抛出Failed to connect to Docker. Please ensure Docker is installed and running.（见 执行器源码），所以 Docker 没启动是第一个要排除的问题。
单独让执行器在容器里跑代码
DockerCommandLineCodeExecutor的工作方式：把每个代码块保存为work_dir下的文件，再在容器内执行该文件，也就是说每个代码块都在一个新进程中执行。它启动时从镜像创建一个容器，默认镜像是python:3-slim，可以通过构造函数的image参数换成自定义镜像；镜像本地不存在时会自动尝试拉取。文档指出，镜像与执行器兼容的唯一要求是安装了sh和python，因此把系统依赖预装进自定义镜像是保证依赖可用的直接办法。
最小示例（与文档示例一致）：
from pathlib import Path from autogen_core import CancellationToken from autogen_core.code_executor import CodeBlock from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor work_dir = Path("coding") work_dir.mkdir(exist_ok=True) async with DockerCommandLineCodeExecutor(work_dir=work_dir) as executor: # type: ignore print( await executor.execute_code_blocks( code_blocks=[ CodeBlock(language="python", code="print('Hello, World!')"), ], cancellation_token=CancellationToken(), ) )文档给出的示例输出为：
CommandLineCodeResult(exit_code=0, output='Hello, World!\n', code_file='coding/tmp_code_07da107bb575cc4e02b0e1d6d99cc204.python')判断执行是否成功看CommandLineCodeResult的exit_code是否为 0，以及output里是否出现代码应打印的内容。code_file指向宿主机work_dir中保存的代码文件；在容器内部，work_dir被挂载为工作目录/workspace（后文示例的 traceback 路径/workspace/tmp_code_...可以佐证这一点）。
几个影响容器行为的参数：
- 默认用上下文管理器（
async with）结束时，或程序退出时通过atexit钩子停止容器。如果想在 AutoGen 用完容器后保留容器以便检查内部状态，创建执行器时把auto_remove设为False；stop_container设为False可以进一步避免执行结束时停止容器。 timeout（默认 60 秒，对应单次执行超时）和支持的语言列表（python，以及bash/shell/sh/pwsh/powershell/ps1）见 执行器源码的类文档。
接入 AutoGen 智能体：Assistant 写代码，Executor 执行
Code Execution 一节的示例实现两个轻量自定义智能体：Assistant负责生成代码，Executor负责从回复里抽出 markdown 代码块并交给执行器，用「绘制 NVIDIA 与 Tesla 2024 年至今累计收益曲线」作为任务。先定义消息类型与两个智能体（与文档代码一致）：
import re from dataclasses import dataclass from typing import List from autogen_core import DefaultTopicId, MessageContext, RoutedAgent, default_subscription, message_handler from autogen_core.code_executor import CodeBlock, CodeExecutor from autogen_core.models import ( AssistantMessage, ChatCompletionClient, LLMMessage, SystemMessage, UserMessage, ) @dataclass class Message: content: str @default_subscription class Assistant(RoutedAgent): def __init__(self, model_client: ChatCompletionClient) -> None: super().__init__("An assistant agent.") self._model_client = model_client self._chat_history: List[LLMMessage] = [ SystemMessage( content="""Write Python script in markdown block, and it will be executed. Always save figures to file in the current directory. Do not use plt.show(). All code required to complete this task must be contained within a single response.""", ) ] @message_handler async def handle_message(self, message: Message, ctx: MessageContext) -> None: self._chat_history.append(UserMessage(content=message.content, source="user")) result = await self._model_client.create(self._chat_history) print(f"\n{'-'*80}\nAssistant:\n{result.content}") self._chat_history.append(AssistantMessage(content=result.content, source="assistant")) # type: ignore await self.publish_message(Message(content=result.content), DefaultTopicId()) # type: ignore def extract_markdown_code_blocks(markdown_text: str) -> List[CodeBlock]: pattern = re.compile(r"```\s*([\w\+\-]+)?\n([\s\S]*?)```") matches = pattern.findall(markdown_text) code_blocks: List[CodeBlock] = [] for match in matches: language = match[0].strip() if match[0] else "" code_content = match[1] code_blocks.append(CodeBlock(code=code_content, language=language)) return code_blocks @default_subscription class Executor(RoutedAgent): def __init__(self, code_executor: CodeExecutor) -> None: super().__init__("An executor agent.") self._code_executor = code_executor @message_handler async def handle_message(self, message: Message, ctx: MessageContext) -> None: code_blocks = extract_markdown_code_blocks(message.content) if code_blocks: result = await self._code_executor.execute_code_blocks( code_blocks, cancellation_token=ctx.cancellation_token ) print(f"\n{'-'*80}\nExecutor:\n{result.output}") await self.publish_message(Message(content=result.output), DefaultTopicId())再注册智能体、启动运行时并发起任务：
import tempfile from autogen_core import SingleThreadedAgentRuntime from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_ext.models.openai import OpenAIChatCompletionClient work_dir = tempfile.mkdtemp() # 创建本地内嵌运行时。 runtime = SingleThreadedAgentRuntime() async with DockerCommandLineCodeExecutor(work_dir=work_dir) as executor: # type: ignore[syntax] model_client = OpenAIChatCompletionClient( model="gpt-4o", # 文档示例中该参数被注释为 api_key="YOUR_API_KEY"， # 实际运行需要填入你自己的 OpenAI API key。 ) await Assistant.register( runtime, "assistant", lambda: Assistant(model_client=model_client), ) await Executor.register(runtime, "executor", lambda: Executor(executor)) # 启动运行时并向 assistant 发布一条消息。 runtime.start() await runtime.publish_message( Message("Create a plot of NVIDA vs TSLA stock returns YTD from 2024-01-01."), DefaultTopicId() ) # 等待运行时空闲后停止。 await runtime.stop_when_idle() await model_client.close()注意extract_markdown_code_blocks里的正则：文档原代码为r"```(?:\s*([\w\+\-]+))?\n([\s\S]*?)```"，上面按原义保留了非捕获分组写法（若直接复制文档原文，请使用文档中的r"```\s*([\w\+\-]+)?\n([\s\S]*?)```"变体即可，两者匹配逻辑相同）。
这套流程能跑通的关键在Assistant的系统提示：要求把 Python 脚本写在 markdown 代码块里、图形必须保存为当前目录的文件、不使用plt.show()。Executor收到消息后只抽取其中的代码块执行，把执行输出作为新的Message发回默认主题，于是Assistant可以看到执行结果并决定是否继续修正——消息投递由SingleThreadedAgentRuntime负责，智能体只负责自身逻辑。
验证执行结果与处理缺依赖
文档示例中这段程序的实际输出展示了完整的「失败—修复—成功」过程（以下为文档示例，非固定预期）：
Assistant第一版脚本用到pandas、matplotlib、yfinance；Executor在容器内执行，因为python:3-slim镜像里没有这些库，输出ModuleNotFoundError: No module named 'pandas'；Assistant回复了一个 bash 代码块pip install pandas matplotlib yfinance；Executor在同一个容器里执行该命令，安装完成后重新执行绘图脚本，Assistant确认曲线图已保存。
成功时绘图文件落在work_dir下（示例中为nvidia_vs_tesla_ytd_returns.png），文档用下面的方式查看产物：
from IPython.display import Image Image(filename=f"{work_dir}/nvidia_vs_tesla_ytd_returns.png")所以核对结果有两层：执行层看CommandLineCodeResult.exit_code == 0与输出内容；任务层看work_dir里是否生成了系统提示要求的产物文件。
排查与限制
- 缺模块不是宿主机问题：依赖装在容器里。文档示例的做法是让智能体输出 bash 代码块在容器内
pip install；如果想一步到位，用image参数换成预装好依赖的自定义镜像即可（镜像里只要有sh和python即可与执行器兼容）。 - Docker 没运行：执行器连接失败时会提示
Failed to connect to Docker. Please ensure Docker is installed and running.，先确认 Docker 守护进程在跑。 - 不要用本地执行器替代：Command Line Code Executors 文档明确提示，
LocalCommandLineCodeExecutor会在本机执行代码，谨慎使用；在 LLM 生成代码的场景下官方示例选择 Docker 执行器正是出于这个考虑。 - 每个代码块都是新进程：命令行列执行器的语义是「保存代码块到文件、每个代码块在新进程中执行」，不要依赖块之间共享内存状态；状态要落在文件里。
可选分支：应用本身也跑在 Docker 容器里
如果你的 AutoGen 应用要打包进容器，又想让它再创建执行容器，文档推荐「Docker out of Docker」：把宿主机 Docker socket 挂载进应用容器，让它在宿主机上派生「兄弟」容器，而不是在容器里再跑一个 Docker 守护进程。做法是在docker run命令中追加：
-v /var/run/docker.sock:/var/run/docker.sock同时，若工作目录属于宿主机，用bind_dir参数指定该宿主机目录，它会被绑定到派生出来的执行容器上，供其访问文件；不指定bind_dir时会回退使用work_dir。
下一步
示例结尾指向 AutoGen 核心用户指南中 Agent Runtime、消息通信、消息处理与订阅的后续章节，想了解分布式运行时（跨进程、跨机器托管智能体）时可以从那里继续。
【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
：在浏览器扩展中通过 chrome.debugger 驱动标签页){kind=spacer}
