如何用 DockerCommandLineCodeExecutor 让 AutoGen 智能体在 Docker 容器中执行代码【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen在 AutoGenautogen里开发智能体应用时一个常见需求是让 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/activateWindows 下激活命令是.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_okTrue) async with DockerCommandLineCodeExecutor(work_dirwork_dir) as executor: # type: ignore print( await executor.execute_code_blocks( code_blocks[ CodeBlock(languagepython, codeprint(Hello, World!)), ], cancellation_tokenCancellationToken(), ) )文档给出的示例输出为CommandLineCodeResult(exit_code0, outputHello, World!\n, code_filecoding/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设为Falsestop_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( contentWrite 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(contentmessage.content, sourceuser)) result await self._model_client.create(self._chat_history) print(f\n{-*80}\nAssistant:\n{result.content}) self._chat_history.append(AssistantMessage(contentresult.content, sourceassistant)) # type: ignore await self.publish_message(Message(contentresult.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(codecode_content, languagelanguage)) 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_tokenctx.cancellation_token ) print(f\n{-*80}\nExecutor:\n{result.output}) await self.publish_message(Message(contentresult.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_dirwork_dir) as executor: # type: ignore[syntax] model_client OpenAIChatCompletionClient( modelgpt-4o, # 文档示例中该参数被注释为 api_keyYOUR_API_KEY # 实际运行需要填入你自己的 OpenAI API key。 ) await Assistant.register( runtime, assistant, lambda: Assistant(model_clientmodel_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、yfinanceExecutor在容器内执行因为python:3-slim镜像里没有这些库输出ModuleNotFoundError: No module named pandasAssistant回复了一个 bash 代码块pip install pandas matplotlib yfinanceExecutor在同一个容器里执行该命令安装完成后重新执行绘图脚本Assistant确认曲线图已保存。成功时绘图文件落在work_dir下示例中为nvidia_vs_tesla_ytd_returns.png文档用下面的方式查看产物from IPython.display import Image Image(filenamef{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），仅供参考 如何用 DockerCommandLineCodeExecutor 让 AutoGen 智能体在 Docker 容器中执行代码？
如何用 DockerCommandLineCodeExecutor 让 AutoGen 智能体在 Docker 容器中执行代码【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen在 AutoGenautogen里开发智能体应用时一个常见需求是让 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/activateWindows 下激活命令是.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_okTrue) async with DockerCommandLineCodeExecutor(work_dirwork_dir) as executor: # type: ignore print( await executor.execute_code_blocks( code_blocks[ CodeBlock(languagepython, codeprint(Hello, World!)), ], cancellation_tokenCancellationToken(), ) )文档给出的示例输出为CommandLineCodeResult(exit_code0, outputHello, World!\n, code_filecoding/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设为Falsestop_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( contentWrite 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(contentmessage.content, sourceuser)) result await self._model_client.create(self._chat_history) print(f\n{-*80}\nAssistant:\n{result.content}) self._chat_history.append(AssistantMessage(contentresult.content, sourceassistant)) # type: ignore await self.publish_message(Message(contentresult.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(codecode_content, languagelanguage)) 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_tokenctx.cancellation_token ) print(f\n{-*80}\nExecutor:\n{result.output}) await self.publish_message(Message(contentresult.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_dirwork_dir) as executor: # type: ignore[syntax] model_client OpenAIChatCompletionClient( modelgpt-4o, # 文档示例中该参数被注释为 api_keyYOUR_API_KEY # 实际运行需要填入你自己的 OpenAI API key。 ) await Assistant.register( runtime, assistant, lambda: Assistant(model_clientmodel_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、yfinanceExecutor在容器内执行因为python:3-slim镜像里没有这些库输出ModuleNotFoundError: No module named pandasAssistant回复了一个 bash 代码块pip install pandas matplotlib yfinanceExecutor在同一个容器里执行该命令安装完成后重新执行绘图脚本Assistant确认曲线图已保存。成功时绘图文件落在work_dir下示例中为nvidia_vs_tesla_ytd_returns.png文档用下面的方式查看产物from IPython.display import Image Image(filenamef{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），仅供参考