使用 Gemini 3.5 Flash Computer Use 控制 Android 手机

Gemini 3.5 Flash 内置了 Computer Use 功能。模型会查看截图，决定下一步操作，并返回类似 click(y=300, x=500) 的函数调用。你通过 ADB 在设备上执行该操作，截取新截图，再发送回去。重复此过程直到任务完成。本指南将带你使用 mobile 环境和 Python SDK 控制 Android 模拟器。
视频加速了 8 倍
Github 仓库：https://github.com/google-gemini/gemini-android-computer-use-quickstart

什么是 Computer Use？

Computer Use 是 Gemini 3.5 Flash 的原生工具。你向模型提供截图和一个目标（如“打开设置并开启深色模式”），模型会返回结构化操作：点击、文字输入、滑动、打开应用等。你的代码在目标设备上执行这些操作。其工作方式类似于函数调用：模型提出操作，你执行它们，然后将结果发回。模型通过截图保持循环。

Gemini 支持三种 Computer Use 环境：browser（桌面网页自动化）、mobile（移动设备/模拟器）和 desktop（操作系统级控制）。本指南使用 mobile。

伪代理循环

from google import genai
client = genai.Client()
bridge = ADBBridge()

# 获取初始截图并发送第一个请求
screenshot = bridge.screenshot()
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input=[
        {"type": "text", "text": "打开设置并启用深色模式"},
        {"type": "image", "data": b64(screenshot), "mime_type": "image/png"},
    ],
    tools=[{"type": "computer_use", "environment": "mobile"}],
)

# 代理循环：执行操作，发回结果
while interaction has function_calls:
    for call in interaction.function_calls:
        bridge.execute(call.name, call.args)  # click, type, open_app...
        screenshot = bridge.screenshot()
        interaction = client.interactions.create(
            model="gemini-3.5-flash",
            previous_interaction_id=interaction.id,
            input=[function_results + screenshot],
            tools=[{"type": "computer_use", "environment": "mobile"}],
        )
print(interaction.output_text)

环境设置

无需 Android Studio GUI。在你的 Mac 上运行设置脚本，即可通过终端安装 Android SDK、模拟器并创建虚拟设备。

1. 运行设置脚本
安装脚本链接：https://github.com/google-gemini/gemini-android-computer-use-quickstart/blob/main/setup_emulator.sh

chmod +x setup_emulator.sh
./setup_emulator.sh

2. 安装 Python 依赖

pip install google-genai

Python 代理脚本会自动处理 SDK 路径的定位以及后台启动模拟器，因此无需手动导出环境变量或进行额外的终端设置。

代理循环

完整的工作脚本 agent.py。它会自动设置环境变量、在后台启动模拟器（如果尚未运行）、等待其启动完毕，然后开始 Computer Use 循环。

import base64
import json
import os
import re
import subprocess
import sys
import time
from google import genai

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

def setup_android_env():
    paths_to_check = [
        os.environ.get("ANDROID_HOME"),
        "/opt/homebrew/share/android-commandlinetools",
        "/usr/local/share/android-commandlinetools",
    ]
    android_home = None
    for p in paths_to_check:
        if p and os.path.exists(p):
            android_home = p
            break
    if not android_home:
        print("Error: ANDROID_HOME not found. Run setup_emulator.sh first.", file=sys.stderr)
        sys.exit(1)
    os.environ["ANDROID_HOME"] = android_home
    sdk_paths = [
        os.path.join(android_home, "cmdline-tools", "latest", "bin"),
        os.path.join(android_home, "emulator"),
        os.path.join(android_home, "platform-tools"),
    ]
    current_path = os.environ.get("PATH", "")
    for p in sdk_paths:
        if p not in current_path:
            current_path = p + os.pathsep + current_path
    os.environ["PATH"] = current_path
    return android_home

def start_emulator(avd_name="AI_Agent_Phone"):
    setup_android_env()
    try:
        res = subprocess.run(["adb", "devices"], capture_output=True, text=True)
        if "emulator" in res.stdout:
            return
    except FileNotFoundError:
        pass
    print(f"Starting emulator '{avd_name}'...")
    log_file = open(os.path.join(BASE_DIR, "emulator.log"), "w")
    subprocess.Popen(
        ["emulator", "-avd", avd_name, "-delay-adb"],
        stdout=log_file,
        stderr=log_file,
        start_new_session=True,
    )
    print("Waiting for emulator to boot...")
    for _ in range(60):
        try:
            res = subprocess.run(["adb", "devices"], capture_output=True, text=True)
            if "emulator" in res.stdout:
                boot_res = subprocess.run(
                    ["adb", "shell", "getprop", "sys.boot_completed"],
                    capture_output=True,
                    text=True,
                )
                if boot_res.stdout.strip() == "1":
                    print("Emulator ready.")
                    return
        except Exception:
            pass
        time.sleep(2)
    print("Error: Emulator failed to boot.", file=sys.stderr)
    sys.exit(1)

class ADBBridge:
    def __init__(self, device_id=None):
        self.prefix = ["adb"] + (["-s", device_id] if device_id else [])
        self.width, self.height = self._screen_size()

    def _run(self, args, check=True):
        result = subprocess.run(self.prefix + args, capture_output=True, text=True)
        if check and result.returncode != 0:
            raise RuntimeError(f"ADB error: {result.stderr.strip()}")
        return result.stdout

    def _screen_size(self):
        output = self._run(["shell", "wm", "size"])
        match = re.search(r"Physical size: (\d+)x(\d+)", output)
        return (int(match.group(1)), int(match.group(2))) if match else (1080, 1920)

    def _px(self, x, y):
        return int(x / 1000 * self.width), int(y / 1000 * self.height)

    def click(self, y, x, **_):
        px, py = self._px(x, y)
        self._run(["shell", "input", "tap", str(px), str(py)])

    def type(self, text, press_enter=False, **_):
        self._run(["shell", "input", "text", text.replace(" ", "%s")])
        if press_enter:
            self._run(["shell", "input", "keyevent", "66"])

    def open_app(self, app_name=None, package_name=None, **_):
        pkg = app_name or package_name
        if not pkg:
            raise ValueError("open_app requires app_name or package_name")
        stdout = self._run(["shell", "monkey", "--pct-syskeys", "0", "-p", pkg, "-c", "android.intent.category.LAUNCHER", "1"], check=False)
        if "No activities found" in stdout or "monkey aborted" in stdout:
            raise RuntimeError(f"App {pkg} is not installed or has no launcher activity.")

    def scroll(self, y, x, direction, magnitude=800, **_):
        px, py = self._px(x, y)
        dist = int(magnitude / 1000 * self.height)
        dx, dy = {"up": (0, -dist), "down": (0, dist), "left": (-dist, 0), "right": (dist, 0)}.get(direction, (0, 0))
        self._run(["shell", "input", "swipe", str(px), str(py), str(px + dx), str(py + dy), "300"])

    def long_press(self, y, x, seconds=2, **_):
        px, py = self._px(x, y)
        self._run(["shell", "input", "swipe", str(px), str(py), str(px), str(py), str(seconds * 1000)])

    def drag_and_drop(self, start_y, start_x, end_y, end_x, **_):
        sx, sy = self._px(start_x, start_y)
        ex, ey = self._px(end_x, end_y)
        self._run(["shell", "input", "swipe", str(sx), str(sy), str(ex), str(ey), "300"])

    def press_key(self, key, **_):
        keymap = {"home": "3", "back": "4", "enter": "66", "app_switch": "187", "menu": "82"}
        self._run(["shell", "input", "keyevent", keymap.get(key.lower(), key)])

    def go_back(self, **_):
        self._run(["shell", "input", "keyevent", "4"])

    def wait(self, seconds=1, **_):
        time.sleep(seconds)

    def list_apps(self, **_):
        output = self._run(["shell", "pm", "list", "packages", "-3"])
        apps = [l.split(":")[1] for l in output.splitlines() if l.startswith("package:")]
        if not apps:
            return {"apps": "No third-party apps installed on this device."}
        return {"apps": apps}

    def take_screenshot(self, **_):
        return None

    def screenshot(self) -> bytes:
        result = subprocess.run(
            self.prefix + ["exec-out", "screencap", "-p"], capture_output=True
        )
        return result.stdout

SYSTEM_PROMPT = """你正在操作一部 Android 手机。
* 使用提供的工具完成任务。
* 在假设某个元素缺失之前，请向下滚动检查整个屏幕。
* 你可以从任何位置通过包名打开应用。
* 仅使用 `type` 工具输入文本。不要使用虚拟键盘。
* 如果任务已完成，请直接说明。"""

def run_agent(task: str, device_id: str = None, max_turns: int = 100):
    start_emulator()
    client = genai.Client()
    bridge = ADBBridge(device_id)
    print(f"\nTask: {task}")
    print("-" * 40)
    screenshot_bytes = bridge.screenshot()
    user_input = [
        {"type": "text", "text": task},
        {
            "type": "image",
            "data": base64.b64encode(screenshot_bytes).decode(),
            "mime_type": "image/png",
        },
    ]
    previous_interaction_id = None
    turn = 0
    while turn < max_turns:
        turn += 1
        interaction = client.interactions.create(
            model="gemini-3.5-flash",
            system_instruction=SYSTEM_PROMPT,
            input=user_input,
            tools=[{"type": "computer_use", "environment": "mobile"}],
            previous_interaction_id=previous_interaction_id,
        )
        function_responses = []
        for step in interaction.steps:
            if step.type == "function_call":
                print(f"[function_call] {step.name}({step.arguments})")
                handler = getattr(bridge, step.name, None)
                result_text = {"status": "ok"}
                if handler:
                    try:
                        res = handler(**step.arguments)
                        if isinstance(res, dict):
                            result_text.update(res)
                    except Exception as e:
                        result_text = {"status": "error", "error": str(e)}
                else:
                    result_text = {"status": "error", "error": f"Unknown action: {step.name}"}
                print(f"[function_result] {result_text}")
                if "safety_decision" in step.arguments:
                    # 自动批准安全决策（演示用）
                    result_text["safety_acknowledgement"] = True
                screenshot_bytes = bridge.screenshot()
                fr = {
                    "type": "function_result",
                    "name": step.name,
                    "call_id": step.id,
                    "result": [
                        {"type": "text", "text": json.dumps(result_text)},
                        {
                            "type": "image",
                            "data": base64.b64encode(screenshot_bytes).decode(),
                            "mime_type": "image/png",
                        },
                    ],
                }
                function_responses.append(fr)
            else:
                print(f"\nResult: {interaction.output_text}")
                break
        user_input = function_responses
        previous_interaction_id = interaction.id
        if not function_responses:
            break
    return interaction

if __name__ == "__main__":
    task_desc = "Find the latest blog post from philipp schmid and summarize it."
    if len(sys.argv) > 1:
        task_desc = " ".join(sys.argv[1:])
    run_agent(task_desc)

如何运行

# 设置 API 密钥
export GEMINI_API_KEY="your-key"
# 运行代理（必要时自动启动模拟器）
python agent.py "打开设置并开启深色模式"

连接到远程设备

你也可以将目标设为物理 Android 设备或远程云模拟器，而非本地虚拟设备。

• 启用 USB/无线调试：在目标设备上，打开开发者选项并开启 USB 调试或无线调试。
• 关于无线调试的详细设置，请参考 Android 官方开发者文档中关于通过 Wi-Fi 进行 ADB 的部分。

adb connect <设备IP>:5555

• 将设备 ID 传递给代理：将远程设备的连接字符串作为 device_id 参数传递，以在代理循环中定位它：

# 定位远程或云端模拟器
run_agent("查看天气", device_id="35.200.100.10:5555")

下一步与开发者提示

希望这能帮助你快速上手 Gemini 3.5 Flash 在移动设备上的 Computer Use 功能。下一步可以：

• 支持 iOS / iPhone：Gemini API 的 mobile 环境是平台无关的。无论设备是 Android 还是 iOS，模型都会在标准化的 0-999 网格上输出操作（点击、滑动、输入）。要针对 iPhone 或 iOS Simulator，你只需将 ADBBridge 替换为与 iOS 兼容的工具，例如用于模拟器控制的 Apple simctl CLI、Appium，或用于物理设备的 go-ios。
• 生产环境健壮性：这里提供的 Python 桥接代码是同步且针对演示优化的。在生产环境中，你应该实现针对网络断开的稳健重试逻辑，优雅处理 ADB 断开连接，并异步执行操作。
• 处理安全决策：在实际任务中（尤其是修改状态或进行支付的任务），模型可能会用 safety_decision 标记操作步骤，要求确认。请确保你的生产循环检查 step.arguments 中的安全标志，并在执行操作前提示用户。详情请参阅 Gemini API Computer Use 安全指南。