Hướng Dẫn Gemini 2.5 Computer Use Kèm Dự Án Demo: Xây Dựng Tác Nhân Tìm Việc

Gần đây, đội ngũ Gemini đã giới thiệu Gemini 2.5 Computer Use, một mô hình chuyên biệt có thể quan sát màn hình trực tiếp và thao tác trên đó bằng cách nhấp, nhập, cuộn và điều hướng web như một người vận hành.

Trong hướng dẫn này, chúng ta sẽ bỏ qua các benchmark trừu tượng và xây dựng một thứ thiết thực: một ứng dụng Streamlit sử dụng Computer Use để điều khiển trình duyệt thật, tìm kiếm Google cho các bài đăng tuyển dụng, áp dụng bộ lọc và xuất kết quả ra CSV mà không cần bất kỳ API tìm kiếm bên thứ ba nào.

Trong bài hướng dẫn này, bạn sẽ học cách:

• Cấu hình vòng lặp Gemini 2.5 Computer Use
• Chạy tác nhân với Playwright (Chromium), dùng danh sách cho phép và tùy chọn xác nhận của con người cho các bước rủi ro.
• Xây dựng UI Streamlit với log hành động trực tiếp và ảnh chụp màn hình sau mỗi lượt.
• Thu thập kết quả organic và tải xuống CSV.

Kết thúc, bạn sẽ có một tác nhân tìm việc có thể tuyển chọn các bài đăng việc làm chính xác cho bạn.

Nếu bạn muốn tìm hiểu thêm về Gemini 2.5, tôi khuyên bạn nên xem hướng dẫn Gemini 2.5 Pro của chúng tôi, bao gồm tính năng, kiểm thử, quyền truy cập, benchmark và hơn thế nữa. Bạn cũng nên tìm hiểu về Gemini 3 Pro, mẫu mới nhất từ Google.

Gemini 2.5 Computer Use là gì?

Gemini 2.5 Computer Use là một mô hình (xem trước) chuyên biệt và công cụ trong Gemini API cho phép bạn xây dựng các tác nhân điều khiển trình duyệt. Thay vì gọi các API theo từng trang, mô hình hoạt động từ ảnh chụp màn hình. Nó có thể thấy nội dung trên trang rồi hành động bằng cách phát ra các hành động UI như navigate, click_at, scroll_document, v.v.  

Mã phía client nhận các hành động được đề xuất đó, thực thi, và gửi lại một tổ hợp mới gồm ảnh chụp màn hình và URL để mô hình quyết định bước tiếp theo. 

Vòng lặp Computer Use hoạt động thế nào

Nguồn: Tài liệu Gemini

Cách vòng lặp tác nhân trên hoạt động từ đầu đến cuối:

1. Gửi yêu cầu: Bắt đầu bằng cách gửi yêu cầu bao gồm công cụ Computer Use, một prompt mục tiêu và ảnh chụp màn hình ban đầu của giao diện người dùng.
2. Nhận phản hồi: Mô hình phản hồi bằng phần tường thuật và một hoặc nhiều lời gọi hàm (như click_at hoặc type_text_at) và một quyết định an toàn.
3. Thực thi: Sau đó thực thi các hành động đề xuất, tức là nếu hành động được đánh dấu thông thường/cho phép, chúng ta có thể chạy; nếu cần xác nhận, chúng ta nhắc người dùng và chỉ tiếp tục khi có chấp thuận rõ ràng.
4. Trả về trạng thái: Sau khi thực thi hành động, chúng ta trả trạng thái về cho mô hình bằng cách gửi FunctionResponse chứa ảnh chụp màn hình mới và URL hiện tại.
5. Lặp lại: Chúng ta lặp chu trình quan sát–quyết định–hành động–ghi nhận cho đến khi hoàn thành tác vụ, xảy ra lỗi, hoặc người dùng hay mô hình chọn dừng.

Xây dựng Tác Nhân Tìm Việc với Gemini 2.5 Computer Use và Streamlit

Trong phần này, chúng ta sẽ xây dựng một tác nhân tìm việc chạy bằng Streamlit, điều khiển trình duyệt thật bằng Gemini 2.5 Computer Use và Playwright, rồi xuất kết quả ra CSV.

Cách hoạt động như sau:

• Người dùng nhập danh sách từ khóa công việc ở thanh bên Streamlit.
• Tác nhân hướng dẫn Gemini mở Google, chạy từng truy vấn và áp dụng bộ lọc thời gian.
• Tác nhân dừng ở trang đầu của kết quả organic và hiển thị ảnh chụp màn hình trực tiếp kèm log hành động ở mỗi lượt.
• Sau đó trích xuất tiêu đề, liên kết và đoạn trích từ trang 1 và tổng hợp qua tất cả từ khóa.
• Cuối cùng, hiển thị bảng kết quả cùng tệp CSV có thể tải xuống.

Bên dưới, ứng dụng áp dụng danh sách miền cho phép để điều hướng an toàn, hỗ trợ tùy chọn xác nhận của con người cho các bước rủi ro và sử dụng vòng lặp Computer Use để dễ gỡ lỗi.

Trước khi bắt đầu: 

Một vài trường hợp sử dụng có trách nhiệm và chính sách trang web gồm:

• Phạm vi tự động hóa: Bản demo này chỉ tự động hóa Tìm kiếm Google (trong danh sách cho phép) và thu thập kết quả organic ở trang 1.
• Tôn trọng điều khoản trang: Không tự động hóa các trang có Điều khoản Dịch vụ hạn chế và không bao giờ cố vượt CAPTCHA. Nếu CAPTCHA xuất hiện, hãy hoàn thành thủ công, rồi để mô hình tiếp tục.
• An toàn có người giám sát: Nếu Gemini đề xuất hành động rủi ro (như tương tác CAPTCHA, mua hàng), ứng dụng mặc định dừng lại. Chỉ tiếp tục khi bạn chủ động bật tự động xác nhận.

Bước 1: Điều kiện tiên quyết

• Python 3.10+
• Khóa API Gemini từ Google AI Studio
• macOS / Windows / Linux có cài Playwright Chromium

Bước 1.1: Cài đặt gói 

Trước hết, hãy đảm bảo bạn đã cài các gói sau:

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

Các lệnh trên tạo môi trường ảo và cài mọi phụ thuộc cốt lõi cần thiết để xây dựng ứng dụng, gồm  Streamlit cho UI, google-genai để gọi Gemini API, Playwright cho tự động hóa trình duyệt, và python-dotenv để tải biến môi trường và Chromium cho Playwright.

Bước 1.2: Thiết lập khóa API Gemini 

Sau khi cài phụ thuộc, tiếp theo thiết lập khóa API Gemini từ AI Studio. 

• Đăng nhập AI Studio và tìm “Get API Key”
• Create API key (Tạo khóa API)
• Chọn một dự án Google Cloud hiện có hoặc tạo mới với thanh toán đã bật.

• Chỉ cần nhấp link a billing account và thêm thông tin thẻ để xác minh.
• Cuối cùng, sao chép khóa API từ AI Studio và lưu lại.

Giờ hãy tạo tệp .env trong thư mục dự án và thêm khóa API của bạn:

GOOGLE_API_KEY=YOUR_REAL_KEY
ALLOWED_HOSTS=google.com

Gói python-dotenv sẽ tải GOOGLE_API_KEY khi chạy để ứng dụng có thể gọi mô hình Gemini 2.5 Computer Use.

Bước 2: Thiết lập cấu hình mô hình

Bây giờ, hãy chuẩn bị cho thời điểm chạy, gồm import, hằng số, ID mô hình và xác thực. Việc này đảm bảo ứng dụng giao tiếp với Gemini và việc điều hướng luôn nằm trong miền cho phép để an toàn.

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)

Chúng ta bắt đầu bằng cách import các thư viện cốt lõi như Streamlit, Playwright, google-genai và các tiện ích nhỏ (dotenv, os, io, csv, time, urllib) cho cấu hình và I/O.

Một khung nhìn cố định (W, H = 1440 × 900) giúp ảnh chụp màn hình và ánh xạ tọa độ nhất quán cho Computer Use, và mô hình xem trước được đặt là gemini-2.5-computer-use-preview-10-2025. 

Khóa API được tải bằng load_dotenv() và os.getenv("GOOGLE_API_KEY"). Điều hướng được thiết lập bằng danh sách cho phép (mặc định google.com) để đảm bảo an toàn. 

Cuối cùng, genai.Client(api_key=API_KEY) khởi tạo SDK Gemini để ứng dụng có thể chạy vòng lặp tác nhân và hiển thị kết quả trong Streamlit.

Bước 3: Tạo các hàm tiện ích

Trước khi chuyển sang phần lõi của ứng dụng, chúng ta cần thiết lập vài hàm trợ giúp. Chúng hỗ trợ kiểm tra an toàn, chuyển đổi tọa độ, thực thi hành động cho các lời gọi hàm của Gemini, thu thập SERP và xuất kết quả thành CSV. 

Bước 3.1: An toàn và thực thi hành động

Chúng ta bắt đầu bằng cách thêm một helper backend để điều khiển trình duyệt an toàn. Bao gồm hàm kiểm tra danh sách miền cho phép để giới hạn điều hướng, hàm chuyển đổi tọa độ để ánh xạ lưới 0–999 của mô hình sang khung nhìn cố định, và hàm điều phối hành động để diễn giải các lời gọi hàm của Computer Use và thực thi chúng qua 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

Hãy xem từng hàm khớp vào pipeline thế nào:

• host_allowed(): Hàm này phân tích URL đích, chuẩn hóa hostname về chữ thường và kiểm tra xem nó có kết thúc bằng bất kỳ miền nào trong ALLOWED_HOSTS được liệt kê trong tệp .env hay không. 
• denorm(): Tiện ích này chuyển đổi tọa độ con trỏ thành đơn vị pixel thực cho khung nhìn hiện tại. Rất quan trọng để nhấp, hover và nhập chính xác, đặc biệt khi thay đổi kích thước màn hình hoặc chạy headless.
• exec_calls(): Cuối cùng, hàm thực thi này quét phản hồi của mô hình để tìm các lời gọi hàm và điều phối từng lời gọi sang Playwright. Nó chặn các lệnh điều hướng tới miền không trong danh sách cho phép, thực thi các hành động được hỗ trợ, chờ trạng thái rỗi mạng để trang ổn định, và ghi lại kết quả theo từng hành động (kể cả lỗi). Các kết quả này sau đó được dùng ở bước tiếp theo để tạo FunctionResponses trả về mô hình.

Các hàm này kết hợp để áp đặt ranh giới, giữ phép tính tọa độ chính xác và ghi log chi tiết.

Bước 3.2: Phản hồi sau hành động và thu thập dữ liệu

Sau khi thực thi hành động, tác nhân cần phản hồi để quyết định bước tiếp theo và trả về đầu ra hữu ích. Bước này cung cấp ba hàm trợ giúp như sau:

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")

Cách mỗi hàm hậu hành động hoạt động:

• fr_from(): Hàm này chụp ảnh PNG mới và URL hiện tại, rồi tạo FunctionResponse cho mỗi hành động đã thực thi. Điều này nhúng ngữ cảnh trực quan cho lượt suy luận tiếp theo và lưu vết có thể kiểm tra được những gì đã xảy ra.
• scrape_google_serp(): Tiếp theo, chúng ta trích xuất tiêu đề, liên kết và đoạn trích từ trang kết quả đầu tiên của Google bằng các bộ chọn bền vững như a:has(h3) cho tiêu đề/liên kết và .VwiC3b cho đoạn trích. Trình thu thập giới hạn kết quả ở max_items và trả về các hàng dữ liệu gọn gàng, có cấu trúc sẵn sàng để phân tích hay xuất.
• to_csv_download(): Hàm này tạo CSV trong bộ nhớ với tiêu đề nhất quán và cung cấp nó qua download_button() của Streamlit.

Giờ chúng ta đã có đủ các hàm trợ giúp, tiếp theo có thể xây dựng ứng dụng Streamlit xoay quanh chúng.

Bước 4: Xây dựng ứng dụng Streamlit

Bước này kết nối UI Streamlit với vòng lặp tác nhân Computer Use. Nó hiển thị giao diện, quản lý cách người dùng tương tác với tác nhân và đảm bảo đầu ra phù hợp nhu cầu.

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()

Giải thích pipeline chi tiết:

• Ứng dụng bắt đầu bằng cách cấu hình trang với bố cục rộng và đặt tiêu đề rõ ràng để người dùng hiểu ngay công cụ làm gì. 
• Ở thanh bên, người dùng cung cấp từ khóa công việc, chọn kiểu truy vấn, bộ lọc thời gian, số lượt tối đa của tác nhân cho mỗi từ khóa và bật/tắt tự động xác nhận cho các hành động rủi ro. Sau đó nút Run search khởi chạy quy trình.
• Khu vực chính chia thành hai cột: “log” bên trái để phát luồng cập nhật văn bản và “ảnh chụp” bên phải hiển thị hình ảnh trực tiếp từ mỗi lượt. Bên dưới là vùng riêng cho bảng kết quả cuối cùng.
• Khi người dùng nhấp Run search, giao diện ghi thông điệp tiến trình vào bảng log, cập nhật bảng ảnh chụp khi có hình mới, và sau khi xử lý mọi từ khóa, hiển thị kết quả tổng hợp dưới dạng dataframe. 

Để chạy ứng dụng này, chỉ cần chạy lệnh bash sau trong terminal:

python -m streamlit run app.py

Mẹo: Luôn khởi chạy Streamlit từ venv của bạn để tránh lỗi “ModuleNotFoundError”.