手把手教你实现自己的“Manus”:构建基于容器的多用户Agent应用【上】




点击上方蓝字关注我们




Manus火了,万元的“邀请码”最后也引起口诛笔伐,不过这多少也说明了大家对AGI的向往与期待。所谓外行看热闹,内行看门道,Manus本质上是一个运行在云端支持多用户使用的Agent as a Service应用,与Deepseek不一样的是,其技术复杂性更多体现在工程上而非底层算法。

本文将尝试构建一个基于容器(沙盒)隔离的多用户Agent系统的后台原型,以帮助理解相关的原理。

内容目录:

  1. Manus原理乱猜

  2. 设计一个多用户的Agent后台

  3. 准备:Docker Image

  4. 准备:Docker Container管理接口

  5. 构建Tool:Python代码执行器(Docker版)

  6. 构建Tool:浏览器自动化(Docker版)

  7. 组装Prompts与ReAct Agent

  8. 测试与改进


我们分两篇介绍以上内容(配套代码获取请参考文末说明)。
01

Manus原理乱猜


从为数不多的演示视频看,Manus背后的一些技术的要点有:

  • 一个自主的多Agent/多Tools系统。

    由一个任务调度的Agent与多个负责任务执行的Agent(或嵌套多Agent系统)组成;不同的Agent可以根据需要使用擅长不同任务的LLM。


  • 编程任务Agent与Web浏览Agent是核心Agent之一。

    为什么?因为这两种Agent最具“通用”能力。

  • 借助虚拟环境(比如Docker容器)做用户隔离。

    为什么?一个是减少用户间的影响;一个是系统安全。你敢让用户指示AI写出的代码随便在本机环境运行吗?


  • 目前擅长的任务流程还是以线性任务为主。

    纯推测。因为非线形任务对LLM推理要求极高,或者借助Workflow来实现,但Workflow又缺乏通用型。


简单的说,每个登录用户会有一个自己的“manus”(Agent),每个"manus"有自己的电脑(虚拟机)与工作空间来完成任务。比如运行代码、浏览网络、创建文件等。

大致的工作流程:

当然,这里的细节我们在实际工程中可根据自己的实际情况参考。
02

设计一个多用户Agent后台


参考这些要点,我们实现一个简化版的支持多用户的Agent系统。

【基本能力】

  • 有简单的自我规划并结合上下文的步骤推理能力

  • 至少有编码Agent与Web Agent两种子任务能力

  • 这两种Agent工作在动态的容器中以确保安全

  • 可多用户同时使用,用户的容器与工作空间要隔离


【系统架构】

系统架构与大致流程设计:

这个系统中包含的组件有:

  • 一个预建的Docker镜像用来启动用户虚拟环境

  • 一个容器管理器组件用来管理多个用户虚拟环境

  • 一个ReAct范式Agent模拟任务步骤推理与调度

  • 四个任务执行的Tool(也可以是Agent),用于:

    • CodeGenerator: 生成Python代码

    • WebpageCrawler: 采集指定网址的内容

    • CodeExecutor: 虚拟环境中执行代码脚本

    • Browser: 执行网络自动浏览

  • 一个Web Agent给Browser调用的自动化浏览Agent(在虚拟环境中)

  • 一个共享工作区存放任务过程中输入输出文件


【工作流程】

下面以一个简单的任务,结合实际运行的效果图,说明系统工作流程。

1. 通过Docker Build创建一个用来启动用户虚拟环境的镜像(Docker Image)。你需要能在本地的Docker镜像库中看到它:

2. 用户输入任务(这里用输入用户ID来模拟不同用户)

此时系统会:

  • 创建一个用户ID对应的ReActAgent

  • 启动一个用户ID对应的虚拟环境(Docker容器),注意这里的容器名:


3. ReActAgent进入任务推理循环,根据上图中的输入任务,自动完成如下步骤:

  • 首先调用WebpageCrawler采集网页内容保存到用户工作区下的一个文件。


  • 调用CodeGenerator生成对该文件分析的python代码。


  • 调用CodeExecutor连接到Docker容器,在容器中执行该代码。


  • 获得代码执行结果,并反馈用户。


  • 以上步骤如果失败,由ReActAgent推理进行再次尝试,直到成功或到达最大迭代次数。比如这里执行发现版本问题,会自动使用pip install做更新:


4. 返回任务执行结果。如果没有新任务,则销毁agent并停止容器。

下面我们介绍核心组件的实现过程,你可以使用配套源码进行实验。
03

准备:Docker Image


由于需要借助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包以外,这里有一些特殊的增加:

说明如下:

  • playwright install:playwright是一个自动化Web测试工具,用来实现在Docker中控制浏览器。这一步是在镜像中安装对应的chromium浏览器。


  • start.sh启动脚本:启动脚本是每一个容器在启动后执行的动作(ENTRYPOINT指令设置)。start.sh内容为:

#!/bin/bash
Xvfb :99 -screen 0 1024x768x16 &
sleep 1
exec "$@"

这个脚本作用是启动一个虚拟显示服务器,用来在没有显示器(docker容器)的情况下运行图形应用,也就是Chromium浏览器,用来实现浏览自动化。

3. 最后使用docker build命令生成镜像。你可以使用我们代码中的build.sh来构建,等待出现提示:

现在,你的Docker镜像就已经准备完毕。如果后期有需要,可以修改Dockerfile后重新build即可。
04

Docker Container管理接口


为了方便对容器管理,比如启动、停止、执行代码等,我们来准备简单的Container管理组件与接口。

【Docker API封装】

对Docker的Container API做封装。大致接口如下(详细请参考源码包)

......
class DockerContainer:
    """管理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
    )
:

        ...
        
    def start(self):
        """启动Docker容器"""
        ...
    
    def set_work_dir(self, work_dir: str) -> None:
        """设置当前工作目录
        
        Args:
            work_dir: 新的工作目录
        """
 
        ...

        
    def stop(self):
        """停止Docker容器"""
        ...
            
    def execute(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/"

#用来获取特定用户的容器单体实例
def get_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生成
    if container_name is None:
        container_name = f"llamaindex-executor-{user_id}"
    
    # 为用户创建专属容器
    if user_id not in _docker_containers or _docker_containers[user_id] is None:
        _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容器
def close_docker_container(user_id: str = "default"):
    global _docker_containers
    ...

# 关闭所有Docker容器
def close_all_docker_containers():
    global _docker_containers
   ...

【测试】

使用如下代码来测试能否动态启动一个容器,并在容器中执行一段代码:

def test_docker_container():
    #启动一个容器
    container = get_docker_container(user_id="test_user")

    #执行的代码
    code = """
import sys
print("Testing Docker container...")
print(f"Python 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")

你应该可以看到如下的输出:

现在我们已经完成了基本的准备工作,我们将在下篇中介绍如何构建Tools与Agent部分,欢迎继续关注。

本文源码获取方法:

关注公众号,发送消息"manus"

end


福利时间

为了帮助LLM开发人员更系统性与更深入的学习RAG应用,特别是企业级的RAG应用场景下,当前主流的优化方法与技术实现,我们编写了《基于大模型的RAG应用开发与优化 — 构建企业级LLM应用》这本长达500页的开发指南,与大家一起来深入到LLM应用开发的全新世界。

更多细节,点击如下链接了解

交流请识别以下名片

请使用浏览器的分享功能分享到微信等