在 Google Colab 用 Mock Endpoint 跑通 Microsoft Fara 浏览器 Agent

你会完成什么

这篇教程教你在 Google Colab 里跑通一个最小 Fara 浏览器 Agent 闭环：

  1. 克隆并安装 Microsoft Fara。
  2. 安装 Playwright Firefox 浏览器依赖。
  3. 创建一个本地 OpenAI-compatible Mock Endpoint。
  4. 让 fara-cli 调用这个 Mock Endpoint。
  5. Mock 模型返回浏览器动作：打开 example.com，然后结束任务。
  6. 确认整个链路可用后，再切换到真实 Fara-7B endpoint。

适用场景

适合你在以下情况下使用：

不适合直接用于生产环境。Mock endpoint 只能证明执行链路通了，不能证明真实模型的网页理解、规划和容错能力。

整体架构

用户任务
  ↓
fara-cli
  ↓
endpoint_config.json
  ↓
OpenAI-compatible endpoint
  ↓
返回 Fara 风格的 browser action
  ↓
Fara / Playwright 执行浏览器动作
  ↓
输出任务结果

本教程默认 endpoint 是本地 Mock 服务：

http://127.0.0.1:8001/v1/chat/completions

Mock 服务第一次返回 visit_url，让浏览器打开 https://example.com；第二次返回 terminate，结束任务。

第 0 步：准备 Colab Notebook

在 Google Colab 新建 Notebook。建议把运行时保持为普通 CPU 即可，不需要 GPU。

先建立全局配置：

import json
import os
import socket
import subprocess
import sys
import time
from pathlib import Path

USE_REAL_FARA_ENDPOINT = False

REAL_FARA_BASE_URL = "http://localhost:5000/v1"
REAL_FARA_API_KEY = "not-needed"
REAL_FARA_MODEL = "microsoft/Fara-7B"

TASK = "Open example.com and tell me what the page is."

WORKDIR = Path("/content/fara_tutorial")
REPO_DIR = Path("/content/fara")
REPO_SRC = REPO_DIR / "src"
OUTPUT_DIR = WORKDIR / "outputs"
ENDPOINT_CONFIG_PATH = WORKDIR / "endpoint_config.json"
MOCK_SERVER_FILE = WORKDIR / "mock_fara_endpoint.py"

WORKDIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

print("Python:", sys.version)
print("Working directory:", WORKDIR)

预期输出形态：

Python: 3.x.x (...)
Working directory: /content/fara_tutorial

第 1 步：定义命令执行与端口检测工具

原文使用了通用 shell runner。为了降低注入风险，这里用 subprocess.run(list) 写法；确实需要 shell 字符串时，再明确封装。

def run_cmd(args: list[str], cwd: Path | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
    print("\n$", " ".join(args))
    result = subprocess.run(
        args,
        cwd=str(cwd) if cwd else None,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        check=False,
    )
    print(result.stdout)
    if check and result.returncode != 0:
        raise RuntimeError(f"Command failed with exit code {result.returncode}: {' '.join(args)}")
    return result


def wait_for_port(host: str, port: int, timeout: int = 60) -> bool:
    start = time.time()
    while time.time() - start < timeout:
        try:
            with socket.create_connection((host, port), timeout=2):
                return True
        except OSError:
            time.sleep(1)
    return False

验证点：这一步没有外部副作用，只要代码单元格不报错即可。

第 2 步：克隆并安装 Fara

os.chdir("/content")

if REPO_DIR.exists():
    print("Fara repo already exists. Pulling latest changes...")
    run_cmd(["git", "pull"], cwd=REPO_DIR, check=False)
else:
    run_cmd(["git", "clone", "https://github.com/microsoft/fara.git", str(REPO_DIR)])

print("\nInstalling Fara and tutorial dependencies...")
run_cmd([sys.executable, "-m", "pip", "install", "-q", "setuptools<82", "wheel", "pip"])
run_cmd([sys.executable, "-m", "pip", "install", "-q", "-e", str(REPO_DIR), "fastapi", "uvicorn", "requests", "pillow"])

if str(REPO_SRC) not in sys.path:
    sys.path.insert(0, str(REPO_SRC))

预期输出形态：

$ git clone https://github.com/microsoft/fara.git /content/fara
...
Installing Fara and tutorial dependencies...
...

如果仓库已经存在，会执行 git pull，即使失败也不会中断整个教程。

第 3 步：安装 Playwright Firefox

Fara 需要浏览器运行时来执行 browser action。

print("\nInstalling Playwright Firefox browser and system dependencies...")
run_cmd([sys.executable, "-m", "playwright", "install", "--with-deps", "firefox"])

预期输出形态：

Installing Playwright Firefox browser and system dependencies...
...

常见问题：

第 4 步：检查 Fara 包与动作定义

这一步不是必须，但能帮助你确认 Fara 的安装路径和 action 定义是否存在。

import importlib

print("\nInspecting Fara package files...")

try:
    import fara
    print("Imported fara from:", getattr(fara, "__file__", "unknown"))
except Exception as exc:
    print("Could not import fara:", repr(exc))

print("\nAvailable files inside /content/fara/src/fara:")
if (REPO_SRC / "fara").exists():
    for path in sorted((REPO_SRC / "fara").glob("*.py")):
        print("-", path.name)
else:
    print("Could not find /content/fara/src/fara")

print("\nTrying to inspect Fara action definitions...")
try:
    fara_agent = importlib.import_module("fara.fara_agent")
    action_defs = getattr(fara_agent, "FARA_ACTION_DEFINITIONS", None)
    if action_defs:
        print("\nFara action space:")
        for action_name, arg_names in action_defs.items():
            args = ", ".join(sorted(arg_names)) if arg_names else "no arguments"
            print(f"- {action_name}: {args}")
    else:
        print("FARA_ACTION_DEFINITIONS was not found. Continuing because this step is optional.")
except Exception as exc:
    print("Could not import fara.fara_agent directly:", repr(exc))
    print("Continuing because this inspection step is optional.")

验证点：

第 5 步：创建 Mock OpenAI-Compatible Endpoint

Mock 服务用 FastAPI 实现，暴露 /v1/chat/completions。它返回 OpenAI-compatible 的响应结构，并在 message.content 中放入 Fara 能识别的 browser action JSON。

mock_server_code = r'''
import time
from fastapi import FastAPI, Request

app = FastAPI()
STATE = {"calls": 0}


@app.post("/v1/chat/completions")
async def chat_completions(request: Request) -> dict:
    payload = await request.json()
    STATE["calls"] += 1
    model_name = payload.get("model", "mock-fara-7b")

    if STATE["calls"] == 1:
        content = (
            "I will open a stable public test page so the browser-control loop can be demonstrated.\n"
            "{\"name\":\"computer\",\"arguments\":{\"action\":\"visit_url\",\"url\":\"https://example.com\"}}"
        )
    else:
        content = (
            "The browser has opened Example Domain, a stable demonstration page used for documentation and examples.\n"
            "{\"name\":\"computer\",\"arguments\":{\"action\":\"terminate\",\"status\":\"success\"}}"
        )

    return {
        "id": f"chatcmpl-mock-{STATE['calls']}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model_name,
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": content},
                "finish_reason": "stop",
            }
        ],
        "usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150},
    }
'''

MOCK_SERVER_FILE.write_text(mock_server_code)
print("Mock endpoint written to:", MOCK_SERVER_FILE)

为什么这样写：

第 6 步：写入 endpoint 配置

if USE_REAL_FARA_ENDPOINT:
    endpoint_config = {
        "model": REAL_FARA_MODEL,
        "base_url": REAL_FARA_BASE_URL,
        "api_key": REAL_FARA_API_KEY,
    }
else:
    endpoint_config = {
        "model": "mock-fara-7b",
        "base_url": "http://127.0.0.1:8001/v1",
        "api_key": "not-needed",
    }

ENDPOINT_CONFIG_PATH.write_text(json.dumps(endpoint_config, indent=2))
print("\nEndpoint config:")
print(ENDPOINT_CONFIG_PATH.read_text())

预期输出形态：

{
  "model": "mock-fara-7b",
  "base_url": "http://127.0.0.1:8001/v1",
  "api_key": "not-needed"
}

第 7 步：启动 Mock 服务

mock_process: subprocess.Popen[str] | None = None

if not USE_REAL_FARA_ENDPOINT:
    print("\nStarting mock OpenAI-compatible endpoint...")
    mock_process = subprocess.Popen(
        [
            sys.executable,
            "-m",
            "uvicorn",
            "mock_fara_endpoint:app",
            "--host",
            "127.0.0.1",
            "--port",
            "8001",
        ],
        cwd=str(WORKDIR),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )

    if not wait_for_port("127.0.0.1", 8001, timeout=60):
        if mock_process.stdout:
            print(mock_process.stdout.read())
        raise RuntimeError("Mock endpoint did not start on port 8001.")

    print("Mock endpoint is running at http://127.0.0.1:8001/v1")
else:
    print("\nUsing real Fara endpoint. Make sure it is reachable.")

验证点：

Mock endpoint is running at http://127.0.0.1:8001/v1

如果端口启动失败：

第 8 步：运行 Fara 浏览器 Agent

优先使用 fara-cli。如果 CLI 入口失败，再尝试模块形式。

print("\nRunning Fara browser agent...")

fara_command = [
    "fara-cli",
    "--task",
    TASK,
    "--endpoint_config",
    str(ENDPOINT_CONFIG_PATH),
]

agent_result = run_cmd(fara_command, cwd=REPO_DIR, check=False)

if agent_result.returncode != 0:
    print("\nfara-cli failed, trying module form...")
    module_command = [
        sys.executable,
        "-m",
        "fara.run_fara",
        "--task",
        TASK,
        "--endpoint_config",
        str(ENDPOINT_CONFIG_PATH),
    ]
    agent_result = run_cmd(module_command, cwd=REPO_DIR, check=False)

print("\nFara command exit code:", agent_result.returncode)

预期输出形态：

Running Fara browser agent...
...
Fara command exit code: 0

如果 exit code 不是 0：

第 9 步：检查输出并关闭 Mock 服务

print("\nSaved tutorial outputs:")
if OUTPUT_DIR.exists():
    files = sorted(OUTPUT_DIR.glob("*"))
    if files:
        for path in files:
            print("-", path)
    else:
        print("No files saved in output directory.")
else:
    print("Output directory does not exist.")

if mock_process is not None:
    print("\nStopping mock endpoint...")
    mock_process.terminate()
    try:
        mock_process.wait(timeout=10)
    except subprocess.TimeoutExpired:
        mock_process.kill()
    print("Mock endpoint stopped.")

print("\nTutorial complete.")

验证点：

Tutorial complete.

即使没有输出文件，只要 Fara 命令成功执行并能看到浏览器动作日志，也说明主链路已跑通。

第 10 步：切换到真实 Fara-7B Endpoint

Mock 跑通之后，再切到真实模型。不要跳过 Mock 直接上真实模型，否则出错时很难判断是模型、endpoint、CLI 还是浏览器执行器的问题。

方案 A：Azure Foundry Endpoint

USE_REAL_FARA_ENDPOINT = True
REAL_FARA_BASE_URL = "https://your-endpoint.inference.ml.azure.com/"
REAL_FARA_API_KEY = "YOUR_AZURE_FOUNDRY_KEY"
REAL_FARA_MODEL = "Fara-7B"

注意：真实密钥不要写进 Notebook 分享版本，建议用 Colab Secrets 或环境变量。

方案 B：GPU 机器上用 vLLM 自托管

在 GPU 机器上运行：

vllm serve "microsoft/Fara-7B" --port 5000 --dtype auto

然后 Notebook 中配置：

USE_REAL_FARA_ENDPOINT = True
REAL_FARA_BASE_URL = "http://localhost:5000/v1"
REAL_FARA_API_KEY = "not-needed"
REAL_FARA_MODEL = "microsoft/Fara-7B"

如果 Colab 和 GPU 机器不在同一环境，需要把 localhost 换成 Colab 能访问到的地址，并处理网络安全边界。

方案 C：LM Studio 或 Ollama

如果你在本地加载兼容模型，并启用了 OpenAI-compatible server，可使用类似地址：

LM Studio: http://localhost:1234/v1
Ollama-style OpenAI server: http://localhost:11434/v1

配置方式仍然是修改 endpoint_config.json 的三个字段：

{
  "model": "你的模型名",
  "base_url": "http://localhost:1234/v1",
  "api_key": "not-needed"
}

最小验收清单

跑完教程后，至少确认这些项：

安全边界

浏览器 Agent 会真实控制浏览器，所以测试时必须收窄范围：

常见故障处理

1. Mock endpoint did not start on port 8001

处理：

2. fara-cli 找不到

处理：

python -m pip show fara

3. Playwright 浏览器启动失败

处理：

python -m playwright install --with-deps firefox

4. 真实模型接入后动作不稳定

处理：

教程结论

这篇教程的核心价值不是“证明 Fara-7B 很强”，而是提供一个更稳的 Agent 验证顺序：

先验证执行器链路 → 再接真实模型 → 最后扩大任务范围

Mock endpoint 让你能在没有 GPU、没有真实模型服务的情况下，先确认 Fara、CLI、endpoint 配置、Playwright 和浏览器动作执行都能工作。等这条链路稳定后，再切换到 Azure Foundry、vLLM、LM Studio 或 Ollama 托管的真实 Fara-7B endpoint。