|
Manus火了,万元的“邀请码”最后也引起口诛笔伐,不过这多少也说明了大家对AGI的向往与期待。所谓外行看热闹,内行看门道,Manus本质上是一个运行在云端支持多用户使用的Agent as a Service应用,与Deepseek不一样的是,其技术复杂性更多体现在工程上而非底层算法。 本文将尝试构建一个基于容器(沙盒)隔离的多用户Agent系统的后台原型,以帮助理解相关的原理。 内容目录:
Manus原理乱猜
设计一个多用户的Agent后台 准备:Docker Image 准备:Docker Container管理接口 构建Tool:Python代码执行器(Docker版) 构建Tool:浏览器自动化(Docker版)
组装Prompts与ReAct Agent 测试与改进
我们分两篇介绍以上内容(配套代码获取请参考文末说明)。从为数不多的演示视频看,Manus背后的一些技术的要点有: 为什么?因为这两种Agent最具“通用”能力。
简单的说,每个登录用户会有一个自己的“manus”(Agent),每个"manus"有自己的电脑(虚拟机)与工作空间来完成任务。比如运行代码、浏览网络、创建文件等。  当然,这里的细节我们在实际工程中可根据自己的实际情况参考。参考这些要点,我们实现一个简化版的支持多用户的Agent系统。 【基本能力】 【系统架构】 系统架构与大致流程设计: 
这个系统中包含的组件有:
【工作流程】 下面以一个简单的任务,结合实际运行的效果图,说明系统工作流程。 1. 通过Docker Build创建一个用来启动用户虚拟环境的镜像(Docker Image)。你需要能在本地的Docker镜像库中看到它: 
2. 用户输入任务(这里用输入用户ID来模拟不同用户): 
此时系统会:

3. ReActAgent进入任务推理循环,ingFang SC", system-ui, -apple-system, BlinkMacSystemFont, "Helvetica Neue", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: var(--articleFontsize);letter-spacing: 0.034em;">根据上图中的输入任务,自动完成如下步骤: 




4. 返回任务执行结果。如果没有新任务,则销毁agent并停止容器。 下面我们介绍核心组件的实现过程,你可以使用配套源码进行实验。由于需要借助AI编程来完成不同用户的任务,提供虚拟环境用于代码执行是必要的。需要注意的是,我们为了调试方便,没有把整个Agent系统都放到虚拟环境下,只把两种类型的任务执行放在容器中完成:Python代码执行与Web浏览自动化。 【Image内容】 启动Docker容器首先需要准备好必要的镜像(Build Image)。这个镜像预置如下内容: 基本的Python代码执行环境 常见的Python第三方库。比如数据分析pandas 需要在虚拟环境下使用的工具。比如我们需要Chromium用来web浏览 必要的脚本,包括shell或者预建的python代码。比如我们这里把用于自动网络浏览的Web Agent代码build到这个虚拟镜像中(agent_browser.py),用来给Browser这个Tool直接调用(为什么?因为对于复杂且需要在虚拟环境下的工作任务,你不可能每次现场编程来完成,既不稳定也不高效)
尽量把最常见的第三方库直接build进去,否则会在执行任务时现场安装,影响性能。 【构建过程】 1. 安装并启动好Docker后台。
2. 准备好Dockerfile: Dockerfile是构建Image镜像的配置文件。除了常见的拉取基础镜像,安装必要的操作系统与Python包以外,这里有一些特殊的增加: 
说明如下: #!/bin/bash Xvfb :99 -screen 0 1024x768x16 & sleep 1 exec"$@"
这个脚本作用是启动一个虚拟显示服务器,用来在没有显示器(docker容器)的情况下运行图形应用,也就是Chromium浏览器,用来实现浏览自动化。 3. 最后使用docker build命令生成镜像。你可以使用我们代码中的build.sh来构建,等待出现提示: ingFang SC", system-ui, -apple-system, BlinkMacSystemFont, "Helvetica Neue", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: var(--articleFontsize);letter-spacing: 0.034em;text-align: justify;"> 现在,你的Docker镜像就已经准备完毕。如果后期有需要,可以修改Dockerfile后重新build即可。为了方便对容器管理,比如启动、停止、执行代码等,我们来准备简单的Container管理组件与接口。 【Docker API封装】 对Docker的Container API做封装。大致接口如下(详细请参考源码包): ...... classDockerContainer: """管理Docker容器的简单类""" def__init__( self, image: str = "python-data-analysis:3.11", container_name: str = "llamaindex-executor", base_work_dir: str = "/Users/pingcy/workspace", auto_remove: bool = True ): ...
defstart(self): """启动Docker容器""" ...
defset_work_dir(self, work_dir: str)-> None: """设置当前工作目录
Args: work_dir: 新的工作目录 """ ...
defstop(self): """停止Docker容器""" ...
defexecute(self, code: str, language: str = "python", work_dir: Optional[str] = None)-> Dict[str, str]: """在Docker容器中执行代码
Args: code: 要执行的代码 language: 代码语言,支持 "python", "sh", "bash" work_dir: 执行代码的工作目录,如果不提供则使用当前工作目录
Returns: Dict包含output和error字段 """ ... 【多用户下的Container管理接口】 现在可以在此基础上提供Container管理的方便函数给Agent使用。这里有两个要点: 支持多用户。需保留用户与容器的对应关系。 确保每个用户只有一个容器(单体)。
获取/启动容器的函数: ...
# 全局变量 - Docker容器映射表(按用户ID组织) _docker_containers: Dict[str, DockerContainer] = {} BASE_WORK_DIR = "/Users/"
#用来获取特定用户的容器单体实例 defget_docker_container( user_id: str = "default", image: str = "python_code_executor:3.11", container_name: Optional[str] = None, )-> DockerContainer: """获取或创建特定用户的Docker容器
Args: user_id: 用户ID,用于区分不同用户 image: Docker镜像名称 container_name: 容器名称,如不提供则根据用户ID生成
Returns: DockerContainer: 用户专属的容器实例 """ global_docker_containers
# 如果不提供容器名称则根据用户ID生成 ifcontainer_name isNone: container_name = f"llamaindex-executor-{user_id}"
# 为用户创建专属容器 ifuser_id notin_docker_containers or_docker_containers[user_id] isNone: _docker_containers[user_id] = DockerContainer( image=image, container_name=container_name, base_work_dir=os.path.join(BASE_WORK_DIR, user_id) ) _docker_containers[user_id].start()
# 确保用户基本工作目录存在 user_work_dir = os.path.join(BASE_WORK_DIR, user_id) os.makedirs(user_work_dir, exist_ok=True)
return_docker_containers[user_id]
停止容器的函数: ...
# 关闭特定用户的Docker容器 defclose_docker_container(user_id: str = "default"): global_docker_containers ...
# 关闭所有Docker容器 defclose_all_docker_containers(): global_docker_containers ...
【测试】 使用如下代码来测试能否动态启动一个容器,并在容器中执行一段代码: deftest_docker_container(): #启动一个容器 container = get_docker_container(user_id="test_user")
#执行的代码 code = """ import sys print("Testing Docker container...") print(f" ython version: {sys.version}") """ try: #在容器中执行代码 result = container.execute(code, "python") print("Execution result:") print(f"Output: {result['output']}") print(f"Error: {result['error']}") finally:
#关闭容器 close_docker_container("test_user")
你应该可以看到如下的输出:

|