Gemini 2.5 Computer Use 实战指南与演示项目：构建求职搜索 Agent

Gemini 团队近期发布了 Gemini 2.5 Computer Use，这是一款专门的模型，能够“看到”实时屏幕并像人工操作员一样执行点击、输入、滚动和网页导航。

在本指南中，我们不做抽象评测，而是动手构建一个实用项目：一个使用 Computer Use 控制真实浏览器的 Streamlit 应用，在不依赖任何第三方搜索 API 的情况下，搜索 Google 职位发布、应用筛选器，并将结果导出为 CSV。

在本教程中，您将学到：

• 配置 Gemini 2.5 Computer Use 循环
• 使用 Playwright（Chromium）运行 agent，配合允许名单与可选的人机确认处理高风险步骤。
• 构建带有实时动作日志和每回合截图的 Streamlit 界面。
• 抓取自然结果并下载 CSV。

完成后，您将获得一个为您筛选准确职位发布的求职搜索 agent。

如果您想进一步了解 Gemini 2.5，建议阅读我们的Gemini 2.5 Pro 教程，涵盖特性、测试、访问方式、基准等内容。您也应了解 Gemini 3 Pro，谷歌最新的模型。

什么是 Gemini 2.5 Computer Use？

Gemini 2.5 Computer Use 是 Gemini API 中的一个专用（预览）模型与工具，可用于构建浏览器控制 agent。与其调用特定网站的 API，该模型基于截图工作。它可以看到页面上的内容，然后通过发出 UI 动作（如 navigate、click_at、scroll_document 等）来执行操作。

客户端代码接收这些候选动作，执行后再回传包含新截图与 URL 的组合，以便模型决定下一步。

Computer Use 循环如何工作

来源：Gemini 文档

上述 agent 循环端到端的工作方式如下：

1. 发送请求： 首先发送一个请求，其中包含 Computer Use 工具、目标提示，以及用户界面的初始截图。
2. 获取响应： 模型返回解说文本以及一个或多个函数调用（如 click_at 或 type_text_at）与安全决策。
3. 执行： 然后执行候选动作，即：若动作为常规/允许，则直接执行；若需确认，则提示用户，仅在明确批准后继续。
4. 返回状态： 动作执行后，通过发送包含新截图与当前 URL 的 FunctionResponse 将状态返回给模型。
5. 重复： 重复该 看–决策–执行–观察 循环，直到任务完成、发生错误，或用户/模型选择停止。

使用 Gemini 2.5 Computer Use 与 Streamlit 构建求职搜索 Agent

本节我们将构建一个基于Streamlit 的求职搜索 agent，使用 Gemini 2.5 Computer Use 与 Playwright 驱动真实浏览器，并将结果导出为 CSV。

其工作流程如下：

• 用户在 Streamlit 侧边栏输入一组职位关键词。
• agent 指示 Gemini 打开 Google，对每个查询执行搜索并应用时间筛选。
• agent 在自然结果的第一页停下，并在每回合显示实时截图与动作日志。
• 随后从第一页抓取标题、链接与摘要，并在所有关键词间汇总。
• 最后渲染结果表格，并提供可下载的 CSV 文件。

在底层，应用通过允许名单限制导航域名、对高风险步骤支持可选人工确认，并使用 Computer Use 循环便于调试。

开始之前：

一些负责任使用与站点策略包括：

• 自动化范围： 此演示仅自动化 Google 搜索（在允许名单中）并抓取第一页的自然结果。
• 遵守站点条款： 切勿自动化具有严格服务条款的网站，亦不要尝试绕过 CAPTCHA。若出现 CAPTCHA，请手动完成，再让模型继续。
• 人机协同安全： 若 Gemini 提出高风险动作（如与 CAPTCHA 的交互、购买），应用默认会暂停。仅当您显式启用自动确认时才会继续。

步骤 1：前置条件

• Python 3.10+
• 来自 Google AI Studio 的 Gemini API 密钥
• 已安装 Playwright Chromium 的 macOS / Windows / Linux

步骤 1.1：安装依赖

首先，确保安装以下依赖：

python -m venv .venv && source .venv/bin/activate
pip install streamlit google-genai playwright python-dotenv
playwright install chromium技术支持

上述命令创建虚拟环境并安装构建应用所需的核心依赖，包括 Streamlit（构建 UI）、google-genai（调用 Gemini API）、Playwright（浏览器自动化），以及 python-dotenv（加载环境变量）与 Chromium（供 Playwright 使用）。

步骤 1.2：设置 Gemini API 密钥

安装好依赖后，接下来从 AI Studio 配置 Gemini API 密钥。

• 登录 AI Studio，搜索“Get API Key”
• Create API key 并点击
• 选择或创建一个已启用结算的 Google Cloud 项目。

• 点击 link a billing account，添加银行卡信息进行验证。
• 最后，从 AI Studio 复制 API 密钥并保存。

现在，在项目文件夹中创建 .env 文件，并将您的 API 密钥写入：

GOOGLE_API_KEY=YOUR_REAL_KEY
ALLOWED_HOSTS=google.com技术支持

python-dotenv 会在运行时加载 GOOGLE_API_KEY，这样应用即可调用 Gemini 2.5 Computer Use 模型。

步骤 2：设置模型配置

现在准备运行环境，包括导入、常量、模型 ID 与认证。这样可确保应用能与 Gemini 通信，并将导航安全地限制在允许名单域内。

import os, io, time, csv, base64, urllib.parse
from typing import List, Dict, Tuple
import streamlit as st
from dotenv import load_dotenv
from playwright.sync_api import sync_playwright
from google import genai
from google.genai import types
from google.genai.types import Content, Part
W, H = 1440, 900
MODEL = "gemini-2.5-computer-use-preview-10-2025"  
load_dotenv()
API_KEY = os.getenv("GOOGLE_API_KEY")
if not API_KEY:
    st.stop()
ALLOWED_HOSTS = {h.strip().lower() for h in os.getenv("ALLOWED_HOSTS", "google.com").split(",") if h.strip()}
client = genai.Client(api_key=API_KEY)技术支持

我们首先导入核心库（Streamlit、Playwright、google-genai），以及用于配置与 I/O 的小工具（dotenv、os、io、csv、time、urllib）。

固定视口（W, H = 1440 × 900）可让截图与坐标映射在 Computer Use 中保持一致；预览模型设为 gemini-2.5-computer-use-preview-10-2025。

通过 load_dotenv() 与 os.getenv("GOOGLE_API_KEY") 加载 API 密钥。导航使用允许名单（默认 google.com）进行安全限制。

最后，通过 genai.Client(api_key=API_KEY) 初始化 Gemini SDK，使应用能运行 agent 循环并在 Streamlit 中渲染结果。

步骤 3：创建辅助函数

在进入应用核心前，需要先设置一些辅助函数。它们用于安全检查、坐标转换、执行 Gemini 函数调用动作、SERP 抓取，以及导出 CSV。

步骤 3.1：安全与动作执行

先添加一个支撑安全浏览器控制的后端辅助模块。这包括：用于约束导航的域名允许名单检查函数、用于将模型 0–999 网格映射到固定视口的坐标转换函数、以及将 Computer Use 函数调用解释并通过 Playwright 执行的动作分发函数。

def host_allowed(url: str) -> bool:
    try:
        netloc = urllib.parse.urlparse(url).netloc.lower()
        return any(netloc.endswith(allowed) for allowed in ALLOWED_HOSTS)
    except Exception:
        return False
def denorm(v: int, size: int) -> int:
    return int(v/1000*size)
def exec_calls(candidate, page, viewport, *, approve_all=False) -> List[Tuple[str, Dict]]:
    W, H = viewport
    results = []
    for part in candidate.content.parts:
        fc = getattr(part, "function_call", None)
        if not fc:
            continue
        name, args = fc.name, (fc.args or {})
        sd = args.get("safety_decision")
        if sd and sd.get("decision") == "require_confirmation":
            reason = sd.get("explanation", "Model flagged a risky action.")
            log_box.warning(f"[SAFETY requires confirmation] {reason}")
            if not approve_all:
                st.stop()
            results.append((name, {"safety_acknowledgement": "true"}))
        if name == "navigate":
            target = args.get("url", "")
            if target and not host_allowed(target):
                log_box.error(f"[BLOCKED] Non-allowlisted host: {target}")
                results.append((name, {"error": "blocked_by_allowlist"}))
                continue
        try:
            if name == "open_web_browser":
                pass
            elif name == "navigate":
                page.goto(args["url"], timeout=30000)
            elif name == "search":
                page.goto("https://www.google.com", timeout=30000)
            elif name == "click_at":
                page.mouse.click(denorm(args["x"], W), denorm(args["y"], H))
            elif name == "hover_at":
                page.mouse.move(denorm(args["x"], W), denorm(args["y"], H))
            elif name == "type_text_at":
                x, y = denorm(args["x"], W), denorm(args["y"], H)
                page.mouse.click(x, y)
                if args.get("clear_before_typing", True):
                    page.keyboard.press("Meta+A"); page.keyboard.press("Backspace")
                page.keyboard.type(args["text"])
                if args.get("press_enter", True):
                    page.keyboard.press("Enter")
            elif name == "scroll_document":
                page.mouse.wheel(0, 800 if args["direction"] == "down" else -800)
            elif name == "key_combination":
                page.keyboard.press(args["keys"])
            page.wait_for_load_state("networkidle", timeout=10000)
            results.append((name, {}))
            time.sleep(0.6)
        except Exception as e:
            results.append((name, {"error": str(e)}))
    return results技术支持

下面解释每个函数在流程中的作用：

• host_allowed() 函数： 解析目标 URL，将主机名转为小写，并检查其是否以 .env 中 ALLOWED_HOSTS 的任一域名结尾。
• denorm() 函数： 将指针坐标转换为当前视口的实际像素。对于精确的点击、悬停和输入至关重要，尤其在更换屏幕尺寸或无头运行时。
• exec_calls() 函数： 最后，执行调用函数会扫描模型响应中的函数调用，并将其分发给 Playwright。它会阻止导航至不在允许名单中的主机，执行受支持的动作，等待网络空闲以稳定页面，并记录每个动作的结果（含错误）。随后这些结果会用于构建回传给模型的 FunctionResponses。

这些函数共同用于设定边界、确保坐标计算准确，并细粒度记录结果。

步骤 3.2：动作后的反馈与数据采集

在执行动作之后，agent 需要反馈以决定下一步并返回有用的输出。本步骤提供三个辅助函数如下：

def fr_from(page, results):
    shot = page.screenshot(type="png")
    url = page.url
    frs = []
    for name, result in results:
        frs.append(types.FunctionResponse(
            name=name,
            response={"url": url, **result},
            parts=[types.FunctionResponsePart(
                inline_data=types.FunctionResponseBlob(mime_type="image/png", data=shot)
            )]
        ))
    return frs, shot
def scrape_google_serp(page, max_items=10):
    items = []
    anchors = page.locator('div#search a:has(h3)')
    count = min(anchors.count(), max_items)
    for i in range(count):
        a = anchors.nth(i)
        title = a.locator('h3').inner_text()
        link = a.get_attribute('href')
        snippet = ""
        snips = page.locator('div#search .VwiC3b')
        if snips.count() > i:
            snippet = snips.nth(i).inner_text()
        items.append({"title": title, "link": link, "snippet": snippet})
    return items
def to_csv_download(rows: List[Dict], name="results.csv"):
    if not rows:
        return None
    out = io.StringIO()
    writer = csv.DictWriter(out, fieldnames=["keyword", "title", "link", "snippet"])
    writer.writeheader(); writer.writerows(rows)
    b = out.getvalue().encode("utf-8")
    href = f"data:text/csv;base64,{base64.b64encode(b).decode()}"
    st.download_button("Download CSV", data=b, file_name=name, mime="text/csv")技术支持

来看每个动作后辅助函数的作用：

• fr_from() 函数： 捕获新的 PNG 截图与当前 URL，并为每个已执行的动作构建一个 FunctionResponse。这为下一次推理提供可视化上下文，并保留可审计的过程记录。
• scrape_google_serp() 函数： 使用稳健的选择器（如 a:has(h3) 获取标题/链接，.VwiC3b 获取摘要）从 Google 第一页提取标题、链接与摘要。抓取结果数量受 max_items 限制，并返回干净的结构化行，便于分析或导出。
• to_csv_download() 函数： 在内存中生成带一致表头的 CSV，并通过 Streamlit 的 download_button() 提供下载。

现在辅助函数已就绪，可以围绕它们构建一个 Streamlit 应用。

步骤 4：构建 Streamlit 应用

本步骤将 Streamlit UI 与 Computer Use 的 agent 循环连接起来。它渲染界面、管理用户与 agent 的交互，并确保输出符合用户需求。

st.set_page_config(page_title="Jobs Search with Gemini Computer Use", layout="wide")
st.title("Jobs Search with Gemini 2.5 Computer Use")
with st.sidebar:
    st.markdown("**Search inputs**")
    default_kw = "data scientist remote\nml engineer \ngenai research india"
    kw_text = st.text_area("Job keywords", value=default_kw, height=120)
    query_mode = st.selectbox("Query style", ["<kw> jobs", 'site:linkedin.com/jobs "<kw>"'], index=0)
    use_past_week = st.checkbox('Ask agent to set "Past week" filter', value=True)
    turns = st.slider("Max agent turns per keyword", 3, 12, 6)
    auto_confirm = st.checkbox("Auto-approve risky actions", value=False, help="If model requests confirmation (e.g., CAPTCHA), auto-approve instead of pausing.")
    run_btn = st.button("Run search")
log_col, shot_col = st.columns([0.45, 0.55])
log_box = log_col.container(height=520)
shot_box = shot_col.container(height=520)
table_box = st.container()
if run_btn:
    keywords = [k.strip() for k in kw_text.splitlines() if k.strip()]
    all_rows = []
    log_box.write("Initializing browser...")
    pw = sync_playwright().start()
    browser = pw.chromium.launch(headless=False)
    ctx = browser.new_context(viewport={"width": W, "height": H})
    page = ctx.new_page()
    try:
        for kw in keywords:
            log_box.subheader(f"▶ {kw}")
            page.goto("https://www.google.com", timeout=30000)
            initial_shot = page.screenshot(type="png")
            base_query = f'{kw} jobs' if query_mode == "<kw> jobs" else f'site:linkedin.com/jobs "{kw}"'
            goal = (
                f'Search Google for "{base_query}". '
                f'{"Open Tools and set time filter to Past week. " if use_past_week else ""}'
                'Stop when first-page results are fully visible; do NOT open the Jobs panel.'
            )
            contents = [Content(role="user", parts=[Part(text=goal), Part.from_bytes(data=initial_shot, mime_type="image/png")])]
            cfg = types.GenerateContentConfig(
                tools=[types.Tool(computer_use=types.ComputerUse(
                    environment=types.Environment.ENVIRONMENT_BROWSER
                ))]
            )
            for turn in range(turns):
                log_box.caption(f"Turn {turn+1}: thinking…")
                resp = client.models.generate_content(model=MODEL, contents=contents, config=cfg)
                cand = resp.candidates[0]
                contents.append(cand.content)
                narr = " ".join([p.text for p in cand.content.parts if getattr(p, "text", None)])
                if narr:
                    log_box.info(narr[:400])
                fcs = [p.function_call for p in cand.content.parts if getattr(p, "function_call", None)]
                if not fcs:
                    log_box.success("Agent stopped proposing actions.")
                    break
                for fc in fcs:
                    log_box.write(f"→ {fc.name} {fc.args or {}}")
                results = exec_calls(cand, page, (W, H), approve_all=auto_confirm)
                frs, shot = fr_from(page, results)
                contents.append(Content(role="user", parts=[Part(function_response=fr) for fr in frs]))
                shot_box.image(shot, caption=page.url, width='stretch')
            rows = scrape_google_serp(page)
            for r in rows:
                r["keyword"] = kw
            all_rows.extend(rows)
            log_box.success(f"{kw}: collected {len(rows)} results")
        if all_rows:
            table_box.dataframe(all_rows, width='stretch')
            to_csv_download(all_rows, name="jobs_google_results.csv")
        else:
            st.warning("No rows collected. Try fewer keywords or fewer turns.")
    finally:
        browser.close()
        pw.stop()技术支持

具体流程说明：

• 应用以宽屏布局配置页面并设置明确标题，便于用户快速了解工具用途。
• 在侧边栏，用户可提供职位关键词、选择查询样式与时间筛选、设置每个关键词的 agent 最大回合数，并启用或禁用对高风险动作的自动确认。随后点击 Run search 按钮触发流程。
• 主区域分为两列：左侧“日志”容器用于滚动输出文本更新，右侧“截图”容器显示每回合的实时图像；下方另有一个容器用于最终结果表格。
• 当用户点击 Run search 后，界面会将进度写入日志面板，在新截图可用时更新截图面板，并在处理完全部关键词后，将汇总结果以数据框形式展示。

要运行该应用，请在终端执行以下 bash 命令：

python -m streamlit run app.py技术支持

提示：请始终在虚拟环境中启动 Streamlit，以避免出现 “ModuleNotFoundError” 问题。