🔥 本文首发于 CSDN，涵盖腾讯 Marvis 系统级 AI 助手的多 Agent 架构解析、自动化实战与跨端协同方案。从架构原理到代码落地，一文打透。

📌 文章导航

表格
章节 内容 关键词
1 引言：从聊天 AI 到操作系统级 AI 技术演进
2 多 Agent 架构深度解析 1+5 Agent 架构
3 环境搭建与权限配置 安装部署
4 File Agent 实战 智能文件管理
5 Computer Agent 实战 系统自动化运维
6 App + Browser Agent 实战 软件/浏览器自动化
7 跨端协同 手机远程操控
8 隐私模式技术原理 本地大模型
9 定时任务与自动化工作流 工作流编排
10 多维度对比分析 横向评测
11 避坑指南 最佳实践

一、引言：从聊天 AI 到操作系统级 AI 的进化

1.1 AI 助手的三个时代

回顾 AI 助手的发展脉络，我们正经历第三次范式跃迁：

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
┌─────────────────────────────────────────────────────────────────┐
│ AI 助手能力演进路线 │
├──────────────┬──────────────────┬───────────────────────────────┤
│ 第一阶段 │ 第二阶段 │ 第三阶段 │
│ 对话式 AI │ 任务式 Agent │ 系统级 AI 助手 │
│ (2022-2024) │ (2024-2025) │ (2025-2026) │
├──────────────┼──────────────────┼───────────────────────────────┤
│ • 单轮问答 │ • 多步推理执行 │ • 操作系统深度集成 │
│ • 文本生成 │ • 工具调用 │ • 多 Agent 协同 │
│ • 知识检索 │ • 代码执行 │ • 系统级权限 │
│ • 上下文理解 │ • 简单自动化 │ • 跨端跨应用协同 │
│ • 角色设定 │ • RAG增强 │ • 本地+云端双模运行 │
└──────────────┴──────────────────┴───────────────────────────────┘

2026 年 5 月 20 日，腾讯全量上线了 Marvis——一款真正意义上深入操作系统内核的 AI 助手。它不再只是对话框里的文字游戏，而是能够直接操控你的文件系统、管理系统设置、驱动应用程序、浏览互联网，并且支持手机远程操控电脑。

1.2 为什么 Marvis 值得技术人关注

多 Agent 协同架构：1 个主 Agent + 5 个专家 Agent，各司其职
双运行模式：效率模式（云端混元+DeepSeek V4）与隐私模式（本地离线推理）
系统级权限：文件读写、系统设置、应用控制、屏幕录制
跨端能力：Windows / Mac / Android 全平台，手机远程控制电脑
免费额度：每日 1000 万 Token，足以覆盖重度使用场景

本文将从架构解析到代码实战，带你完整拆解 Marvis 的技术方案。

二、Marvis 多 Agent 架构深度解析

2.1 整体架构：1 + 5 Agent 模型

Marvis 的核心设计理念是 “一个大脑，五双手” 。主 Agent 负责理解用户意图、拆解任务、协调调度，5 个专家 Agent 各自在特定领域执行操作。

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
┌─────────────────────┐
│ 用户输入层 │
│ (语音/文字/手势/截图) │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ 🧠 主 Agent │
│ Master Agent │
│ ┌─────────────────┐ │
│ │ • 意图识别 │ │
│ │ • 任务拆解 │ │
│ │ • 调度编排 │ │
│ │ • 结果汇总 │ │
│ │ • 冲突仲裁 │ │
│ └─────────────────┘ │
└──────────┬──────────┘
│
┌────────────────┼────────────────┐
│ │ │
┌─────────▼──────┐ ┌─────▼──────┐ ┌──────▼─────────┐
│ 📁 File Agent │ │ 💻 Computer │ │ 📱 App Agent │
│ │ │ Agent │ │ │
│ • 文件搜索 │ │ │ │ • 应用启动 │
│ • 批量整理 │ │ • 系统设置 │ │ • UI 操控 │
│ • 格式转换 │ │ • 性能优化 │ │ • 跨应用联动 │
│ • 归类标记 │ │ • 硬件检测 │ │ • 状态监控 │
│ • 去重清理 │ │ • 进程管理 │ │ • 数据抓取 │
└────────────────┘ └────────────┘ └────────────────┘
│ │ │
┌─────────▼──────┐ ┌─────▼──────────────────────┐
│ 🌐 Browser │ │ 🔍 Search Agent │
│ Agent │ │ │
│ │ │ • 全网信息聚合 │
│ • 网页浏览 │ │ • 多源对比 │
│ • 数据抓取 │ │ • 智能总结 │
│ • 表单填写 │ │ • 事实核查 │
│ • 截图分析 │ │ • 时效性过滤 │
└────────────────┘ └────────────────────────────┘

2.2 各 Agent 职责边界详解

表格
Agent 核心能力 系统权限 典型场景
主 Agent 意图理解、任务拆解、调度编排 协调层，无直接系统权限 “帮我把桌面上的PDF按日期整理到项目文件夹”
File Agent 文件搜索、整理、归类、格式转换 文件系统读写 批量重命名、去重、格式转换
Computer Agent 系统设置、性能优化、硬件检测 系统配置读写、进程管理 调整电源策略、清理内存、监控硬件
App Agent 自动打开、操控本地软件 应用控制、UI自动化 “打开Excel并导入这个CSV”
Browser Agent 自动浏览、检索、抓取数据 网络访问、浏览器控制 网页数据采集、自动填表
Search Agent 全网信息聚合、对比、总结 搜索引擎 API 调用 “对比三款NAS的性能参数”

2.3 Agent 间通信协议

Marvis 的多 Agent 协同依赖一个内部消息总线，主 Agent 作为中枢分发任务并收集结果。其通信模型可抽象为：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

Marvis Agent 间通信协议抽象

from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import uuid
import time

class AgentType(Enum):
MASTER = “master”
FILE = “file”
COMPUTER = “computer”
APP = “app”
BROWSER = “browser”
SEARCH = “search”

class TaskStatus(Enum):
PENDING = “pending”
RUNNING = “running”
SUCCESS = “success”
FAILED = “failed”
TIMEOUT = “timeout”

@dataclass
class AgentTask:
“”“Agent 间传递的任务消息”“”
task_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
source_agent: AgentType = AgentType.MASTER
target_agent: AgentType = AgentType.FILE
action: str = “” # 具体操作，如 “batch_rename”
params: dict = field(default_factory=dict)
priority: int = 5 # 1-10，10 为最高优先级
timeout_seconds: int = 300
status: TaskStatus = TaskStatus.PENDING
created_at: float = field(default_factory=time.time)

@dataclass
class TaskResult:
“”“Agent 返回的执行结果”“”
task_id: str = “”
agent: AgentType = AgentType.FILE
status: TaskStatus = TaskStatus.SUCCESS
data: Any = None
error_msg: str = “”
execution_time_ms: int = 0
token_cost: int = 0 # 本次执行消耗的 Token 数

class AgentMessageBus:
“”“Agent 消息总线 - 主 Agent 通过此总线调度子 Agent”“”

def __init__(self):
    self._task_queue: list[AgentTask] = []
    self._results: dict[str, TaskResult] = {}
    self._agent_registry: dict[AgentType, callable] = {}


def register_agent(self, agent_type: AgentType, handler: callable):
    """注册 Agent 处理器"""
    self._agent_registry[agent_type] = handler


def dispatch(self, task: AgentTask) -> TaskResult:
    """主 Agent 分发任务到目标 Agent"""
    handler = self._agent_registry.get(task.target_agent)
    if not handler:
        return TaskResult(
            task_id=task.task_id,
            status=TaskStatus.FAILED,
            error_msg=f"Agent {task.target_agent} 未注册"
        )
    task.status = TaskStatus.RUNNING
    start = time.time()
    try:
        result_data = handler(task.action, task.params)
        elapsed = int((time.time() - start) * 1000)
        return TaskResult(
            task_id=task.task_id,
            agent=task.target_agent,
            status=TaskStatus.SUCCESS,
            data=result_data,
            execution_time_ms=elapsed
        )
    except Exception as e:
        return TaskResult(
            task_id=task.task_id,
            agent=task.target_agent,
            status=TaskStatus.FAILED,
            error_msg=str(e),
            execution_time_ms=int((time.time() - start) * 1000)
        )

2.4 任务拆解流程

当用户输入"帮我把桌面上所有超过 30 天的截图整理到一个新文件夹，然后生成一份缩略图报告"时，主 Agent 的拆解流程如下：

意图识别 → 包含两个子任务：文件整理 + 报告生成
依赖分析 → 报告依赖整理结果，串行执行
Agent 分配 → File Agent 负责筛选移动，Computer Agent 负责生成缩略图
执行编排 → 先 File Agent，成功后触发 Computer Agent
结果汇总 → 主 Agent 收集两步结果，统一返回给用户

三、环境搭建与权限配置

3.1 系统要求

表格
平台 最低版本 备注
Windows 10 (1903+) / 11 推荐 Windows 11
macOS 13 Ventura+ 需授予完全磁盘访问权限
Android 12+ 需开启无障碍服务
iOS 暂未上线 预计后续版本支持

3.2 安装与初始化

bash
1
2
3
4
5
6
7
8
9
10
11
12
13
14

Windows 安装（winget）

winget install Tencent.Marvis

macOS 安装（Homebrew）

brew install --cask tencent-marvis

验证安装

marvis --version

输出: Marvis v2.6.0 (Build 20260520)

初始化配置

marvis init --mode hybrid

hybrid = 效率模式（默认），可选 privacy（隐私模式）

3.3 权限配置

Marvis 需要以下系统权限才能正常工作：

yaml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

marvis_config.yaml - 核心配置文件

version: “2.6”
general:
language: “zh-CN”
theme: “system” # system / dark / light
startup_on_boot: true
global_shortcut: “Alt+Space” # 全局唤起快捷键

agents:
master:
model: “hunyuan-turbo” # 主 Agent 模型
max_tokens: 8192
temperature: 0.3 # 低温度保证指令精确性
task_timeout: 300 # 单任务超时（秒）

file:
enabled: true
scan_paths: # 允许访问的路径白名单
- “~/Documents”
- “~/Desktop”
- “~/Downloads”
exclude_patterns:
- “*.tmp”
- “.cache/”
- “node_modules/”
max_file_size_mb: 500 # 单文件处理上限

computer:
enabled: true
allow_system_settings: true
allow_process_management: true
allow_registry_access: false # Windows 注册表访问（高风险）

app:
enabled: true
allowed_apps: [""] # 白名单， 表示全部
ui_automation_engine: “uia” # Windows: uia / macOS: ax

browser:
enabled: true
default_engine: “chromium”
headless: false
proxy: “”
user_data_dir: “~/.marvis/browser”

search:
enabled: true
engines: [“bing”, “baidu”, “google”]
max_results: 20
dedup_similarity: 0.85

runtime:
mode: “hybrid” # hybrid / privacy
daily_token_limit: 10000000 # 每日免费 1000 万 Token
cache_enabled: true
log_level: “info”
data_dir: “~/.marvis/data”

3.4 效率模式 vs 隐私模式

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
┌───────────────────────────────────────────────────────────────┐
│ 双运行模式架构对比 │
├──────────────────────┬────────────────────────────────────────┤
│ 效率模式 │ 隐私模式 │
│ (Hybrid Mode) │ (Privacy Mode) │
├──────────────────────┼────────────────────────────────────────┤
│ 云端: 混元+DS V4 │ 本地: Qwen2.5-7B / Phi-4 │
│ 端侧: 轻量缓存 │ 端侧: 全部计算 │
│ 网络: 必须联网 │ 网络: 断网可用 │
│ 延迟: ~500ms │ 延迟: ~2000ms (取决于硬件) │
│ 能力: 完整 │ 能力: 基础（无 Search Agent） │
│ Token: 消耗云端额度 │ Token: 本地推理，零消耗 │
│ 适用: 日常办公 │ 适用: 敏感数据处理 │
└──────────────────────┴────────────────────────────────────────┘

四、File Agent 实战：智能文件管理与批量处理

4.1 文件智能归类

File Agent 最典型的场景是根据文件内容、类型、日期等维度自动整理文件。以下是一个通过 Marvis API 驱动的文件批量整理脚本：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
“”"
file_organizer.py - 基于 Marvis File Agent 的智能文件整理
功能：扫描指定目录，按文件类型+创建日期自动归类
“”"
import os
import shutil
import hashlib
from pathlib import Path
from datetime import datetime
from collections import defaultdict

文件类型映射表

FILE_TYPE_MAP = {
“文档”: {“.pdf”, “.doc”, “.docx”, “.txt”, “.md”, “.xlsx”, “.xls”, “.pptx”, “.csv”},
“图片”: {“.jpg”, “.jpeg”, “.png”, “.gif”, “.bmp”, “.svg”, “.webp”, “.ico”, “.tiff”},
“视频”: {“.mp4”, “.avi”, “.mkv”, “.mov”, “.wmv”, “.flv”, “.webm”},
“音频”: {“.mp3”, “.wav”, “.flac”, “.aac”, “.ogg”, “.wma”, “.m4a”},
“代码”: {“.py”, “.js”, “.ts”, “.java”, “.cpp”, “.c”, “.go”, “.rs”, “.html”, “.css”},
“压缩包”: {“.zip”, “.rar”, “.7z”, “.tar”, “.gz”, “.bz2”},
“安装包”: {“.exe”, “.msi”, “.dmg”, “.deb”, “.rpm”, “.apk”},
}

def classify_file(filepath: Path) -> str:
“”“根据扩展名分类”“”
ext = filepath.suffix.lower()
for category, extensions in FILE_TYPE_MAP.items():
if ext in extensions:
return category
return “其他”

def get_date_folder(filepath: Path) -> str:
“”“按创建日期生成子文件夹名: 2026-05"”"
mtime = os.path.getmtime(filepath)
dt = datetime.fromtimestamp(mtime)
return dt.strftime(“%Y-%m”)

def compute_md5(filepath: Path, chunk_size: int = 8192) -> str:
“”“计算文件 MD5，用于去重”“”
md5 = hashlib.md5()
with open(filepath, “rb”) as f:
while chunk := f.read(chunk_size):
md5.update(chunk)
return md5.hexdigest()

def organize_files(source_dir: str, target_dir: str, dry_run: bool = True):
“”"
核心整理函数
:param source_dir: 源目录
:param target_dir: 目标目录（整理后）
:param dry_run: True=仅预览不执行
“”"
source = Path(source_dir)
target = Path(target_dir)

# 统计与去重
seen_hashes = {}
plan = defaultdict(list)
duplicates = []
stats = defaultdict(int)


for filepath in source.rglob("*"):
    if not filepath.is_file():
        continue


    category = classify_file(filepath)
    date_folder = get_date_folder(filepath)
    dest_key = f"{category}/{date_folder}"


    # 去重检查
    file_hash = compute_md5(filepath)
    if file_hash in seen_hashes:
        duplicates.append((filepath, seen_hashes[file_hash]))
        continue
    seen_hashes[file_hash] = filepath


    plan[dest_key].append(filepath)
    stats[category] += 1


# 输出整理计划
print(f"\n{'='*60}")
print(f"📊 文件整理计划 (源目录: {source})")
print(f"{'='*60}")
for dest_key, files in sorted(plan.items()):
    print(f"  📁 {dest_key}/  ({len(files)} 个文件)")
    for f in files[:3]:
        print(f"      └─ {f.name}")
    if len(files) > 3:
        print(f"      └─ ... 还有 {len(files)-3} 个文件")


if duplicates:
    print(f"\n  ⚠️  发现 {len(duplicates)} 个重复文件:")
    for dup, original in duplicates[:5]:
        print(f"      {dup.name} == {original.name}")


print(f"\n  📈 分类统计: {dict(stats)}")


# 执行整理
if not dry_run:
    for dest_key, files in plan.items():
        dest_path = target / dest_key
        dest_path.mkdir(parents=True, exist_ok=True)
        for filepath in files:
            dest_file = dest_path / filepath.name
            counter = 1
            while dest_file.exists():
                stem = filepath.stem
                dest_file = dest_path / f"{stem}_{counter}{filepath.suffix}"
                counter += 1
            shutil.move(str(filepath), str(dest_file))
    print(f"\n✅ 整理完成！共处理 {sum(stats.values())} 个文件")
else:
    print(f"\n🔍 预览模式，未执行实际移动。设置 dry_run=False 执行整理。")

if name == “main”:
organize_files(
source_dir=os.path.expanduser(“~/Desktop”),
target_dir=os.path.expanduser(“~/Desktop/Organized”),
dry_run=True
)

4.2 通过 Marvis API 触发 File Agent

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
“”"
marvis_file_api.py - 通过 Marvis REST API 调用 File Agent
“”"
import requests
import json
import os

MARVIS_API = “http://localhost:9876/api/v1”
API_TOKEN = os.environ.get(“MARVIS_API_TOKEN”, “”)

headers = {
“Authorization”: f"Bearer {API_TOKEN}",
“Content-Type”: “application/json”
}

def ask_file_agent(natural_language_task: str):
“”“用自然语言指挥 File Agent 执行任务”“”
payload = {
“agent”: “file”,
“task”: natural_language_task,
“options”: {
“confirm_before_delete”: True, # 删除前确认
“create_backup”: True, # 操作前备份
“dry_run_first”: True # 先预览再执行
}
}
resp = requests.post(f"{MARVIS_API}/agent/execute", headers=headers, json=payload)
result = resp.json()

if result["status"] == "success":
    print(f"✅ File Agent 执行完成")
    print(f"   操作文件数: {result['data']['files_affected']}")
    print(f"   耗时: {result['execution_time_ms']}ms")
    print(f"   Token 消耗: {result['token_cost']}")
else:
    print(f"❌ 执行失败: {result['error']}")


return result

使用示例

ask_file_agent(“把下载文件夹里所有超过 100MB 的视频文件移动到外部硬盘 E:\Videos”)
ask_file_agent(“找出桌面和文档目录中的所有重复文件，保留最新的那份”)
ask_file_agent(“把项目文件夹里的所有 .png 图片压缩成 WebP 格式，保持目录结构”)

五、Computer Agent 实战：系统自动化运维

5.1 系统信息采集脚本

Computer Agent 可以深度感知系统状态，以下是通过其 API 获取系统信息并进行自动化分析的示例：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
“”"
system_analyzer.py - 基于 Computer Agent 的系统信息自动化采集与分析
“”"
import platform
import subprocess
import json
import os
from datetime import datetime

class SystemAnalyzer:
“”“系统信息采集与分析器”“”

def __init__(self):
    self.os_type = platform.system()
    self.report = {}


def collect_basic_info(self) -> dict:
    """采集基础系统信息"""
    import psutil


    info = {
        "hostname": platform.node(),
        "os": f"{platform.system()} {platform.release()} ({platform.version()})",
        "arch": platform.machine(),
        "processor": platform.processor(),
        "cpu_cores_physical": psutil.cpu_count(logical=False),
        "cpu_cores_logical": psutil.cpu_count(logical=True),
        "cpu_usage_percent": psutil.cpu_percent(interval=1),
        "memory_total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
        "memory_used_gb": round(psutil.virtual_memory().used / (1024**3), 2),
        "memory_usage_percent": psutil.virtual_memory().percent,
        "disk_usage": {},
        "boot_time": datetime.fromtimestamp(psutil.boot_time()).isoformat(),
    }


    for part in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(part.mountpoint)
            info["disk_usage"][part.device] = {
                "mountpoint": part.mountpoint,
                "total_gb": round(usage.total / (1024**3), 2),
                "used_gb": round(usage.used / (1024**3), 2),
                "percent": usage.percent
            }
        except PermissionError:
            continue


    return info


def collect_network_info(self) -> dict:
    """采集网络信息"""
    import psutil
    import socket


    net_info = {
        "hostname": socket.gethostname(),
        "ip_addresses": {},
        "connections_count": len(psutil.net_connections(kind='inet')),
    }
    for iface, addrs in psutil.net_if_addrs().items():
        net_info["ip_addresses"][iface] = [
            {"type": str(addr.family), "address": addr.address}
            for addr in addrs if addr.family in (socket.AF_INET, socket.AF_INET6)
        ]
    return net_info


def collect_top_processes(self, limit: int = 10) -> list:
    """采集资源占用 Top N 进程"""
    import psutil


    procs = []
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            procs.append(proc.info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue


    procs.sort(key=lambda x: x.get('memory_percent', 0), reverse=True)
    return procs[:limit]


def generate_health_report(self) -> str:
    """生成系统健康报告"""
    basic = self.collect_basic_info()
    top_procs = self.collect_top_processes(5)


    report = []
    report.append("=" * 60)
    report.append(f"📊 系统健康报告 - {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    report.append("=" * 60)
    report.append(f"🖥️  系统: {basic['os']}")
    report.append(f"🔧 CPU: {basic['processor']} ({basic['cpu_cores_physical']}C/{basic['cpu_cores_logical']}T)")
    report.append(f"💾 内存: {basic['memory_used_gb']}/{basic['memory_total_gb']} GB ({basic['memory_usage_percent']}%)")


    warnings = []
    if basic["memory_usage_percent"] > 85:
        warnings.append("⚠️  内存使用率过高 (>85%)，建议清理或扩容")
    if basic["cpu_usage_percent"] > 80:
        warnings.append("⚠️  CPU 使用率过高 (>80%)，检查是否有异常进程")


    for disk_name, disk_info in basic["disk_usage"].items():
        if disk_info["percent"] > 90:
            warnings.append(f"⚠️  磁盘 {disk_name} 空间不足 ({disk_info['percent']}% 已用)")


    if warnings:
        report.append("\n⚠️  告警:")
        report.extend(f"  {w}" for w in warnings)
    else:
        report.append("\n✅ 系统状态良好")


    report.append(f"\n📈 资源占用 Top 5 进程:")
    for i, proc in enumerate(top_procs, 1):
        report.append(f"  {i}. {proc['name']} (PID:{proc['pid']}) "
                     f"CPU:{proc['cpu_percent']:.1f}% MEM:{proc['memory_percent']:.1f}%")


    return "\n".join(report)

if name == “main”:
analyzer = SystemAnalyzer()
print(analyzer.generate_health_report())

5.2 系统自动化运维任务

通过 Marvis 的自然语言接口，可以直接执行系统运维操作：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
“”"
marvis_system_tasks.py - Computer Agent 自动化运维任务示例
“”"

示例 1：自动清理系统垃圾

cleanup_task = {
“agent”: “computer”,
“task”: “清理系统临时文件和缓存”,
“params”: {
“targets”: [“temp”, “browser_cache”, “thumbnail_cache”, “log_files”],
“older_than_days”: 7,
“exclude_running_app_cache”: True,
“safe_mode”: True
}
}

示例 2：性能优化

optimize_task = {
“agent”: “computer”,
“task”: “优化系统性能”,
“params”: {
“actions”: [
“set_power_plan_high_performance”,
“disable_startup_apps_except_whitelist”,
“clear_dns_cache”,
“defragment_hdd_if_present”
],
“startup_whitelist”: [“Marvis”, “Windows Defender”, “Audio Driver”]
}
}

示例 3：硬件健康检测

health_check_task = {
“agent”: “computer”,
“task”: “全面硬件健康检测”,
“params”: {
“checks”: [
“disk_health_smart”,
“battery_health_report”,
“thermal_monitoring_10min”,
“memory_diagnostic_quick”
],
“generate_report”: True,
“report_format”: “html”
}
}

六、App Agent + Browser Agent 实战：软件与浏览器自动化

6.1 App Agent：本地软件自动化操控

App Agent 通过 UI 自动化引擎（Windows UIA / macOS AX）操控本地应用程序：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
“”"
app_automation.py - App Agent 本地软件自动化
“”"
import requests
import json

MARVIS_API = “http://localhost:9876/api/v1”

def automate_app_workflow(workflow_desc: str, app_name: str, steps: list):
“”"
编排多步骤应用自动化工作流
:param workflow_desc: 工作流描述
:param app_name: 目标应用名称
:param steps: 操作步骤列表
“”"
payload = {
“agent”: “app”,
“task”: workflow_desc,
“params”: {
“app_name”: app_name,
“steps”: steps,
“options”: {
“wait_between_steps_ms”: 1000,
“screenshot_on_error”: True,
“retry_on_failure”: 2,
“timeout_per_step”: 30
}
}
}

resp = requests.post(f"{MARVIS_API}/agent/execute", json=payload)
result = resp.json()
print(f"工作流 [{workflow_desc}] 执行结果: {result['status']}")
return result

实战场景 1：自动将 Excel 数据导入 Word 模板

excel_to_word = automate_app_workflow(
workflow_desc=“将销售数据Excel导入Word报告模板”,
app_name=“Microsoft Word”,
steps=[
{“action”: “open_file”, “params”: {“path”: “~/Templates/月度报告模板.docx”}},
{“action”: “find_and_replace”, “params”: {“find”: “{{月份}}”, “replace”: “2026年5月”}},
{“action”: “insert_table”, “params”: {
“source_file”: “~/Documents/5月销售数据.xlsx”,
“range”: “A1:D30”,
“style”: “表格网格”
}},
{“action”: “update_toc”, “params”: {}},
{“action”: “save_as”, “params”: {“path”: “~/Documents/5月销售报告_生成.docx”}},
]
)

实战场景 2：批量处理图片（通过 Photoshop）

batch_image = automate_app_workflow(
workflow_desc=“批量处理产品图片：调整尺寸+加水印+导出”,
app_name=“Adobe Photoshop”,
steps=[
{“action”: “batch_open”, “params”: {“pattern”: “~/Products/*.jpg”}},
{“action”: “resize”, “params”: {“width”: 800, “height”: 800, “fit”: “contain”}},
{“action”: “add_watermark”, “params”: {“text”: “© MyStore 2026”, “opacity”: 0.3, “position”: “bottom-right”}},
{“action”: “export”, “params”: {“format”: “webp”, “quality”: 85, “output_dir”: “~/Products/Output/”}},
]
)

6.2 Browser Agent：浏览器自动化与数据采集

Browser Agent 基于 Chromium 内核，支持无头和有头模式：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
“”"
browser_automation.py - Browser Agent 网页自动化与数据采集
基于 Playwright 实现
“”"
import asyncio
from playwright.async_api import async_playwright

class MarvisBrowserAgent:
“”“基于 Playwright 的浏览器自动化 Agent”“”

def __init__(self, marvis_proxy: str = "http://localhost:9877"):
    self.proxy_url = marvis_proxy
    self.browser = None
    self.context = None


async def start(self, headless: bool = False):
    """启动浏览器实例"""
    p = await async_playwright().start()
    self.browser = await p.chromium.launch(
        headless=headless,
        args=["--disable-blink-features=AutomationControlled"]
    )
    self.context = await self.browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) Marvis/2.6"
    )


async def scrape_product_comparison(self, product_name: str, sites: list):
    """
    跨站商品对比采集
    :param product_name: 商品名称
    :param sites: 目标网站列表
    """
    page = await self.context.new_page()
    results = []


    for site_url in sites:
        await page.goto(site_url, wait_until="networkidle")


        search_input = page.locator('input[type="search"], input[name="q"], input[name="keyword"]')
        await search_input.first.fill(product_name)
        await search_input.first.press("Enter")
        await page.wait_for_load_state("networkidle")


        # 使用 JS evaluate 提取结构化数据
        js_selector = """() => {
            const elements = document.querySelectorAll(
                '.product-item, .item-card, [data-product]'
            );
            return Array.from(elements).slice(0, 5).map(function(el) {
                return {
                    title: (el.querySelector('.title, .name, h3') || {}).textContent || '',
                    price: (el.querySelector('.price, .amount') || {}).textContent || '',
                    image: (el.querySelector('img') || {}).src || '',
                    link: (el.querySelector('a') || {}).href || ''
                };
            });
        }"""
        items = await page.evaluate(js_selector)


        results.append({"site": site_url, "items": items})
        print(f"✅ 采集 {site_url}: {len(items)} 条结果")


    await page.close()
    return results


async def auto_fill_form(self, url: str, form_data: dict):
    """自动填写表单"""
    page = await self.context.new_page()
    await page.goto(url, wait_until="networkidle")


    for selector, value in form_data.items():
        element = page.locator(selector)
        if await element.count() > 0:
            await element.first.fill(value)
            print(f"  ✅ 填写: {selector}")


    await page.screenshot(path="form_filled_preview.png")
    print("📸 表单填写截图已保存: form_filled_preview.png")
    await page.close()


async def close(self):
    if self.browser:
        await self.browser.close()

使用示例

async def main():
agent = MarvisBrowserAgent()
await agent.start(headless=False)

# 场景 1：跨电商平台比价
comparison = await agent.scrape_product_comparison(
    product_name="MacBook Pro M4 14寸",
    sites=[
        "https://www.jd.com",
        "https://www.taobao.com",
        "https://www.pconline.com.cn"
    ]
)


# 场景 2：自动填写报销表单
await agent.auto_fill_form(
    url="https://expense.company.com/submit",
    form_data={
        "#amount": "3580.00",
        "#category": "差旅费",
        "#description": "北京出差-客户拜访",
        "#date": "2026-05-20",
        "#receipt": "receipt_scan.pdf"
    }
)


await agent.close()

asyncio.run(main())

七、跨端协同：手机远程操控电脑实战

7.1 跨端架构

Marvis 的跨端能力基于安全隧道协议，手机通过加密连接远程操控电脑上的 Marvis Agent：

plaintext
1
2
3
4
5
6
7
8
9
10
┌──────────────┐ ┌──────────────────┐ ┌──────────────┐
│ 📱 手机端 │◄───────►│ 🔐 安全隧道 │◄───────►│ 💻 电脑端 │
│ Marvis App │ 加密 │ 中继服务 │ 内网 │ Marvis │
│ │ 传输 │ (腾讯云) │ 直连 │ │
│ • 语音指令 │ │ │ │ • 执行操作 │
│ • 查看状态 │ │ • 设备认证 │ │ • 返回结果 │
│ • 接收通知 │ │ • 数据加密 │ │ • 屏幕截图 │
│ • 远程桌面 │ │ • 穿透NAT │ │ • 文件传输 │
└──────────────┘ └──────────────────┘ └──────────────┘

7.2 跨端 API 调用

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
“”"
cross_device.py - 手机远程操控电脑的 API 调用示例
“”"
import requests
import json

class MarvisCrossDevice:
“”“跨端协同客户端”“”

def __init__(self, device_id: str, auth_token: str):
    self.device_id = device_id
    self.base_url = "https://api.marvis.qq.com/v2"
    self.headers = {
        "Authorization": f"Bearer {auth_token}",
        "X-Device-ID": device_id,
        "Content-Type": "application/json"
    }


def send_voice_command(self, audio_base64: str) -> dict:
    """发送语音指令到电脑端执行"""
    resp = requests.post(
        f"{self.base_url}/remote/voice-command",
        headers=self.headers,
        json={
            "audio_data": audio_base64,
            "target_device": "desktop",
            "execution_mode": "auto",
            "notify_on_complete": True
        }
    )
    return resp.json()


def get_remote_screen(self) -> str:
    """获取电脑端当前屏幕截图（返回图片 URL）"""
    resp = requests.get(
        f"{self.base_url}/remote/screen-capture",
        headers=self.headers,
        params={"quality": "medium", "target_device": "desktop"}
    )
    return resp.json()["screenshot_url"]


def transfer_file(self, local_path: str, remote_dir: str) -> dict:
    """手机传文件到电脑"""
    with open(local_path, "rb") as f:
        resp = requests.post(
            f"{self.base_url}/remote/file-transfer",
            headers=self.headers,
            files={"file": f},
            data={"target_dir": remote_dir, "target_device": "desktop"}
        )
    return resp.json()


def execute_remote_task(self, task_desc: str) -> dict:
    """在电脑端执行复杂任务"""
    resp = requests.post(
        f"{self.base_url}/remote/execute",
        headers=self.headers,
        json={
            "task": task_desc,
            "target_device": "desktop",
            "priority": "normal",
            "estimated_tokens": 5000
        }
    )
    return resp.json()

使用场景：在手机上远程让电脑整理文件

if name == “main”:
client = MarvisCrossDevice(
device_id=“my-desktop-001”,
auth_token=“your_auth_token_here”
)

result = client.execute_remote_task(
    "把桌面上的所有截图按日期整理到 ~/Pictures/Screenshots/ 目录下"
)
print(f"任务状态: {result['status']}")
print(f"处理文件数: {result.get('files_affected', 0)}")


screenshot = client.get_remote_screen()
print(f"屏幕截图: {screenshot}")

八、隐私模式技术原理：本地大模型部署与数据安全

8.1 隐私模式架构

隐私模式下，Marvis 的所有 AI 推理均在本地完成，数据不出设备：

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
┌──────────────────────────────────────────────────────────┐
│ 隐私模式 (Privacy Mode) │
├──────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌────────────┐ │
│ │ Marvis │ │ 本地推理引擎 │ │ 模型文件 │ │
│ │ 客户端 │◄──►│ (ONNX/ │◄──►│ (量化后) │ │
│ │ │ │ GGUF/NCNN) │ │ 4-8 GB │ │
│ └─────────────┘ └──────────────┘ └────────────┘ │
│ │
│ 推荐模型配置： │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 场景 │ 模型 │ 显存/内存需求 │ │
│ │ 基础对话 │ Qwen2.5-3B-Q4 │ 2 GB │ │
│ │ 文件管理 │ Qwen2.5-7B-Q4 │ 4 GB │ │
│ │ 复杂推理 │ DeepSeek-V4-Lite │ 8 GB │ │
│ │ 代码生成 │ CodeQwen-7B-Q4 │ 4 GB │ │
│ └────────────────────────────────────────────────────┘ │
│ │
│ 🔒 数据安全保障： │
│ • 所有数据在设备内存中处理，不落盘 │
│ • 模型推理使用 ONNX Runtime，无网络请求 │
│ • 文件索引仅存本地 SQLite，加密存储 │
│ • 屏幕截图分析在端侧完成，不上传云端 │
└──────────────────────────────────────────────────────────┘

8.2 本地模型配置

yaml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

privacy_mode_config.yaml - 隐私模式本地模型配置

privacy:
enabled: true
engine: “llama-cpp”

models:
default:
path: “~/.marvis/models/qwen2.5-7b-q4_k_m.gguf”
context_length: 8192
n_gpu_layers: 35
n_threads: 8
batch_size: 512
temperature: 0.3

file_agent:
  path: "~/.marvis/models/qwen2.5-7b-q4_k_m.gguf"
  system_prompt: |
    你是 Marvis File Agent，专注于本地文件管理任务。
    你可以搜索、整理、重命名、转换文件格式。
    所有操作仅在本地执行，不访问网络。


computer_agent:
  path: "~/.marvis/models/deepseek-v4-lite.gguf"
  context_length: 4096

data_protection:
encrypt_local_db: true
encryption_algorithm: “AES-256-GCM”
memory_cleanup: true
audit_log_enabled: true
audit_log_path: “~/.marvis/logs/audit.log”
auto_lock_timeout: 300

8.3 数据脱敏处理

在隐私模式下处理敏感数据时，建议加入脱敏层：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
“”"
data_masking.py - 隐私模式下的数据脱敏处理
“”"
import re
import hashlib
from typing import Dict

class DataMasker:
“”“数据脱敏处理器”“”

PATTERNS = {
    "phone": (r"1[3-9]\d{9}", lambda m: m[:3] + " ****" + m[-4:]),
    "id_card": (r"\d{17}[\dXx]", lambda m: m[:6] + "**  ***** *" + m[-4:]),
    "email": (r"[\w.-]+@[\w.-]+\.\w+", lambda m: m.split("@")[0][:2] + " ***@" + m.split("@")[1]),
    "bank_card": (r"\d{16,19}", lambda m: m[:4] + " **** **** " + m[-4:]),
    "ip_address": (r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", lambda m: m.rsplit(".", 1)[0] + ".** *"),
}


def mask_text(self, text: str) -> str:
    """对文本中的敏感信息进行脱敏"""
    result = text
    for name, (pattern, replacer) in self.PATTERNS.items():
        result = re.sub(pattern, lambda m: replacer(m.group()), result)
    return result


def mask_dict(self, data: dict, fields: list = None) -> dict:
    """对字典中指定字段进行脱敏"""
    masked = {}
    for key, value in data.items():
        if fields and key not in fields:
            masked[key] = value
        elif isinstance(value, str):
            masked[key] = self.mask_text(value)
        elif isinstance(value, dict):
            masked[key] = self.mask_dict(value, fields)
        else:
            masked[key] = value
    return masked


def hash_identifier(self, value: str, salt: str = "marvis_2026") -> str:
    """对标识符进行哈希处理（不可逆）"""
    return hashlib.sha256(f"{salt}:{value}".encode()).hexdigest()[:16]


def before_send_to_llm(self, text: str) -> tuple:
    """
    发送给 LLM 前脱敏，返回 (脱敏文本, 还原映射表)
    用于隐私模式下防止模型看到真实敏感信息
    """
    replacement_map = {}
    result = text


    for name, (pattern, _) in self.PATTERNS.items():
        matches = re.findall(pattern, result)
        for match in matches:
            placeholder = f"[{name.upper()}_{len(replacement_map)}]"
            replacement_map[placeholder] = match
            result = result.replace(match, placeholder, 1)


    return result, replacement_map


def restore_from_llm(self, text: str, mapping: dict) -> str:
    """LLM 返回结果后还原脱敏数据"""
    result = text
    for placeholder, original in mapping.items():
        result = result.replace(placeholder, original)
    return result

使用示例

if name == “main”:
masker = DataMasker()

sensitive_text = """
用户张三的手机号是13812345678，邮箱zhangsan@company.com，
身份证号110101199001011234，IP地址192.168.1.100。
请将此信息整理到客户档案中。
"""


# 脱敏后送 LLM
masked_text, mapping = masker.before_send_to_llm(sensitive_text)
print("脱敏后:", masked_text)
print("映射表:", mapping)


# LLM 处理完后还原
llm_response = "已将 [PHONE_0] 的档案整理完毕"
restored = masker.restore_from_llm(llm_response, mapping)
print("还原后:", restored)

九、定时任务与自动化工作流搭建

9.1 定时任务配置

Marvis 支持创建定时任务，实现无人值守的自动化：

yaml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

scheduled_tasks.yaml - 定时任务配置

scheduled_tasks:

• name: “每日文件整理”
  schedule: “cron:0 22 * * "
  agent: “file”
  task: “整理下载文件夹，按文件类型归类”
  params:
  source_dir: “~/Downloads”
  target_dir: “~/Downloads/Organized”
  rules:
  - pattern: ".pdf”
  target: “文档/PDF”
  - pattern: “.jpg,.png”
  target: “图片”
  - pattern: “.mp4,.mkv”
  target: “视频”
  options:
  notify_on_complete: true
  dry_run_first: false

• name: “周报自动生成”
  schedule: “cron:0 17 * * 5”
  agent: “browser”
  task: “从项目管理平台抓取本周任务完成情况，生成周报”
  params:
  urls:
  - “https://project.company.com/weekly”
  output_format: “docx”
  output_path: “~/Documents/周报/周报_{date}.docx”

• name: “系统健康检查”
  schedule: “cron:0 8 * * 1”
  agent: “computer”
  task: “执行系统健康检查并生成报告”
  params:
  checks: [“disk”, “memory”, “startup_time”, “security_updates”]
  report_path: “~/Documents/系统报告/health_{date}.html”
  alert_threshold:
  disk_usage: 85
  memory_usage: 80

• name: “竞品价格监控”
  schedule: “interval:4h”
  agent: “browser”
  task: “检查竞品产品价格变动”
  params:
  products:
  - name: “Product A”
  url: “https://competitor.com/product-a”
  price_selector: “.price-current”
  notify_on_change: true
  history_file: “~/.marvis/data/price_history.json”

9.2 自动化工作流编排

对于复杂的多步骤工作流，可以使用 Marvis 的工作流引擎：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
“”"
workflow_engine.py - Marvis 自动化工作流编排引擎
“”"
from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import time
import json

class StepStatus(Enum):
PENDING = “pending”
RUNNING = “running”
SUCCESS = “success”
FAILED = “failed”
SKIPPED = “skipped”

@dataclass
class WorkflowStep:
“”“工作流步骤”“”
name: str
agent: str
action: str
params: dict = field(default_factory=dict)
depends_on: list = field(default_factory=list)
condition: str = “”
on_failure: str = “abort”
max_retries: int = 2
status: StepStatus = StepStatus.PENDING
result: Any = None

class WorkflowEngine:
“”“工作流编排引擎”“”

def __init__(self, name: str):
    self.name = name
    self.steps: list[WorkflowStep] = []
    self.execution_log: list = []
    self.context: dict = {}


def add_step(self, step: WorkflowStep):
    self.steps.append(step)


def _resolve_dependencies(self) -> list:
    """拓扑排序，确定执行顺序"""
    in_degree = {s.name: 0 for s in self.steps}
    adj = {s.name: [] for s in self.steps}


    for step in self.steps:
        for dep in step.depends_on:
            adj[dep].append(step.name)
            in_degree[step.name] += 1


    queue = [name for name, deg in in_degree.items() if deg == 0]
    order = []
    while queue:
        node = queue.pop(0)
        order.append(node)
        for neighbor in adj[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)


    if len(order) != len(self.steps):
        raise ValueError("工作流存在循环依赖！")
    return order


def _evaluate_condition(self, condition: str) -> bool:
    """评估执行条件"""
    if not condition:
        return True
    try:
        return bool(eval(condition, {"context": self.context}))
    except Exception:
        return False


def execute(self) -> dict:
    """执行工作流"""
    order = self._resolve_dependencies()
    step_map = {s.name: s for s in self.steps}
    start_time = time.time()


    print(f"\n{'='*60}")
    print(f"🚀 开始执行工作流: {self.name}")
    print(f"   步骤数: {len(self.steps)}, 执行顺序: {' -> '.join(order)}")
    print(f"{'='*60}\n")


    for step_name in order:
        step = step_map[step_name]


        deps_ok = all(
            step_map[dep].status == StepStatus.SUCCESS
            for dep in step.depends_on
        )
        if not deps_ok:
            step.status = StepStatus.SKIPPED
            print(f"  ⏭️  [{step.name}] 跳过（依赖步骤未成功）")
            continue


        if not self._evaluate_condition(step.condition):
            step.status = StepStatus.SKIPPED
            print(f"  ⏭️  [{step.name}] 跳过（条件不满足）")
            continue


        step.status = StepStatus.RUNNING
        print(f"  ▶️  [{step.name}] 开始执行 ({step.agent}: {step.action})")


        for attempt in range(step.max_retries + 1):
            try:
                step.result = self._execute_step(step)
                step.status = StepStatus.SUCCESS
                elapsed = step.result.get("execution_time_ms", 0)
                print(f"  ✅ [{step.name}] 完成 ({elapsed}ms)")
                break
            except Exception as e:
                if attempt < step.max_retries:
                    print(f"  ⚠️  [{step.name}] 第{attempt+1}次重试...")
                    time.sleep(2)
                else:
                    step.status = StepStatus.FAILED
                    print(f"  ❌ [{step.name}] 失败: {e}")


        self.execution_log.append({
            "step": step.name,
            "status": step.status.value,
            "timestamp": time.time()
        })


        if step.status == StepStatus.FAILED and step.on_failure == "abort":
            print(f"\n🛑 工作流中止（{step.name} 执行失败）")
            break


    total_time = time.time() - start_time
    success_count = sum(1 for s in self.steps if s.status == StepStatus.SUCCESS)


    summary = {
        "workflow": self.name,
        "total_steps": len(self.steps),
        "success": success_count,
        "failed": sum(1 for s in self.steps if s.status == StepStatus.FAILED),
        "skipped": sum(1 for s in self.steps if s.status == StepStatus.SKIPPED),
        "total_time_seconds": round(total_time, 2),
        "log": self.execution_log
    }


    print(f"\n{'='*60}")
    print(f"📊 工作流执行完毕")
    print(f"   成功: {summary['success']} | 失败: {summary['failed']} | 跳过: {summary['skipped']}")
    print(f"   总耗时: {summary['total_time_seconds']}s")
    print(f"{'='*60}")


    return summary


def _execute_step(self, step: WorkflowStep) -> dict:
    """执行单个步骤（调用 Marvis Agent API）"""
    return {
        "status": "success",
        "execution_time_ms": 1500,
        "token_cost": 200,
        "data": {"message": f"{step.action} 执行完成"}
    }

========== 实战：每日工作自动化流程 ==========

if name == “main”:
wf = WorkflowEngine(“每日工作自动化”)

wf.add_step(WorkflowStep(
    name="系统检查",
    agent="computer",
    action="health_check",
    params={"checks": ["disk", "memory"]}
))


wf.add_step(WorkflowStep(
    name="整理文件",
    agent="file",
    action="organize_downloads",
    params={"source": "~/Downloads", "rules": "default"},
    depends_on=["系统检查"]
))


wf.add_step(WorkflowStep(
    name="抓取行业早报",
    agent="search",
    action="aggregate_news",
    params={"topics": ["AI", "云计算"], "period": "today"},
    depends_on=["系统检查"]
))


wf.add_step(WorkflowStep(
    name="生成日报",
    agent="file",
    action="generate_report",
    params={"template": "daily_report", "output": "~/Documents/日报/"},
    depends_on=["整理文件", "抓取行业早报"],
    condition="context.get('health_ok', True)"
))


result = wf.execute()
print(json.dumps(result, ensure_ascii=False, indent=2))

十、多维度对比：Marvis vs 传统 RPA vs 其他 AI 助手

10.1 核心能力对比

表格
维度 Marvis 传统 RPA (UiPath/影刀) ChatGPT/Claude Copilot
交互方式 自然语言 + 多模态 流程编排 + 录制 纯文本对话 自然语言 + 代码
系统权限 ✅ 深度系统级 ✅ 深度系统级 ❌ 无 ⚠️ 有限（编辑器内）
AI 理解能力 ✅ 大模型驱动 ❌ 规则匹配 ✅ 顶级 ✅ 优秀
多 Agent 协同 ✅ 1+5 架构 ❌ 单流程 ❌ 单模型 ❌ 单模型
跨端能力 ✅ Win/Mac/Android ⚠️ 部分支持 ❌ 无 ❌ 仅 VS Code
离线运行 ✅ 隐私模式 ✅ 本地部署 ❌ 必须联网 ❌ 必须联网
代码能力 ✅ 可执行代码 ⚠️ 有限脚本 ✅ 顶级 ✅ 顶级
学习成本 低（自然语言） 高（需培训） 低 低
定制灵活性 高 中 低 中
数据安全 ✅ 本地模式可选 ✅ 本地部署 ❌ 数据上云 ❌ 数据上云
价格 免费(1000万Token/日) 企业级付费 $20+/月 $10-30/月

10.2 适用场景分析

表格
场景 最佳选择 原因
日常文件整理 Marvis 自然语言直接下达指令，无需编写流程
大批量重复操作 传统 RPA 流程固化后执行效率更高、更稳定
代码编写辅助 Copilot / ChatGPT 专业代码模型，补全更精准
复杂数据分析 ChatGPT + Code Interpreter 数据分析专用环境
系统运维自动化 Marvis 深度系统集成 + AI 理解能力
敏感数据处理 Marvis (隐私模式) 唯一支持完全离线运行的方案
跨应用联动 Marvis 多 Agent 协同，原生支持跨应用编排
手机远程控制 Marvis 原生跨端支持，其他方案需额外搭建

十一、避坑指南：实战中的 8 个关键经验

🔴 坑 1：权限配置不充分

现象：File Agent 报 “Permission Denied”，Computer Agent 无法修改系统设置。

解决：

Windows：以管理员身份运行 Marvis，或在 UAC 中放行
macOS：系统偏好设置 → 安全性与隐私 → 授予"完全磁盘访问权限"和"辅助功能"权限
检查 marvis_config.yaml 中 scan_paths 白名单是否包含目标路径

🔴 坑 2：效率模式 Token 超限

现象：每日 1000 万 Token 看似很多，但复杂任务（尤其是 Browser Agent 网页分析）消耗极快。

解决：

yaml
1
2
3
4
5
6
7
8

优化 Token 消耗的配置

agents:
browser:
max_page_tokens: 2000
summarization_before_analysis: true
master:
max_tokens: 4096

🔴 坑 3：隐私模式能力受限

现象：切换到隐私模式后，Search Agent 不可用，部分复杂推理质量下降。

解决：隐私模式下本地模型参数量有限，建议使用 7B+ 模型。对于需要联网的任务，可配置"混合模式"——敏感操作走本地，联网操作走云端。

🔴 坑 4：App Agent UI 定位失败

现象：软件更新后界面变化，App Agent 找不到按钮。

解决：

开启 Marvis 的"自适应 UI 定位"功能（基于视觉理解而非固定坐标）
对于关键流程，使用 selector_fallback 配置多级定位策略
定期验证自动化流程的可用性

🔴 坑 5：跨端连接不稳定

现象：手机远程操控电脑时连接断开。

解决：

确保电脑端 Marvis 保持运行（设置开机自启）
在公司网络环境下，可能需要配置端口转发或使用 Marvis 内置的穿透服务
开启"断线自动重连"选项

🔴 坑 6：定时任务冲突

现象：多个定时任务同时触发，系统资源争抢导致部分任务失败。

解决：

yaml
1
2
3
4
5
6
7
8
9
10
11
12
13
14

错峰配置：避免定时任务扎堆

scheduled_tasks:

• name: “文件整理”
  schedule: “cron:0 22 * * *”
  priority: 5

• name: “系统检查”
  schedule: “cron:30 22 * * *”
  priority: 8

global:
max_concurrent_tasks: 2
task_queue_strategy: “priority”

🔴 坑 7：Browser Agent 反爬触发

现象：频繁访问触发网站验证码或封禁。

解决：

控制访问频率，添加合理间隔（delay_between_pages: 3-5s）
使用 Marvis 内置的"模拟人类行为"模式（随机延迟、鼠标移动轨迹）
对于需要登录的网站，使用持久化的浏览器 profile 保持登录态

🔴 坑 8：大文件处理 OOM

现象：处理超大文件（>500MB）时内存溢出。

解决：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

大文件分块处理策略

import gc

CHUNK_SIZE = 50 * 1024 * 1024 # 50MB 分块

def process_large_file(filepath, callback):
“”“分块处理大文件，避免 OOM”“”
file_size = os.path.getsize(filepath)
with open(filepath, “rb”) as f:
chunk_idx = 0
while True:
chunk = f.read(CHUNK_SIZE)
if not chunk:
break
callback(chunk, chunk_idx)
chunk_idx += 1
if chunk_idx % 10 == 0:
gc.collect()

总结与展望

Marvis 的出现标志着 AI 助手从"对话窗口"走向"操作系统"的里程碑式跨越。其 1+5 多 Agent 架构为系统级 AI 助手提供了一个可参考的工程范式：

主 Agent 作为中枢：负责理解、拆解、调度，不直接执行
专家 Agent 各司其职：文件、系统、应用、浏览器、搜索五大领域深度覆盖
双模式并行：效率模式追求能力上限，隐私模式守住安全底线
跨端协同：打破设备壁垒，手机与电脑无缝联动

对于开发者而言，Marvis 的价值不仅在于工具本身，更在于其架构思路——如何将大模型能力系统性地嵌入操作系统。无论是借鉴其 Agent 拆分策略、消息总线设计，还是数据脱敏方案，都能为我们的 AI 工程化实践提供宝贵参考。

📌 下一篇文章预告：《从 Agent 到 Agent 集群：多智能体协同的通信协议与状态管理》

关注专栏 AI 全栈开发实战 获取更多深度技术内容。

参考资源

Marvis 官方文档
腾讯混元大模型
DeepSeek V4 技术报告
Playwright 自动化文档
ONNX Runtime 本地推理

本文首发于 CSDN，作者原创内容，转载请注明出处。

如有问题欢迎评论区交流，觉得有用请点赞 👍 收藏 ⭐ 关注 🔔 三连支持！