从“观察—决策—行动”到可验证的浏览器 Agent

任务 + 当前页面状态
        ↓
      观察
        ↓
  解释状态并决定动作
        ↓
      执行动作
        ↓
   产生新的页面状态
        ↓
  验证完成 / 继续循环 / 转人工

browser-agent-tutorial/
├── cases/
│   ├── case.json
│   └── case-out-of-stock.json
├── toy_app/
│   └── index.html
├── mock_loop.py
├── real_agent.py
└── pyproject.toml

mkdir -p browser-agent-tutorial/{cases,toy_app}
cd browser-agent-tutorial

{
  "case_id": "CASE-4107",
  "order_id": "ORD-1042",
  "issue": "wrong_item",
  "replacement_in_stock": true,
  "policy": "wrong_item_and_in_stock_means_replacement"
}

from __future__ import annotations

import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


class NeedsHumanReview(RuntimeError):
    pass


@dataclass
class MockBrowser:
    case: dict[str, Any]
    selected: bool = False
    resolution: str | None = None
    note: str = ""
    status: str = "Open"
    audit_log: list[str] = field(default_factory=list)

    def observe(self) -> dict[str, Any]:
        return {
            "selected": self.selected,
            "case": self.case,
            "resolution": self.resolution,
            "note": self.note,
            "status": self.status,
            "audit_log": list(self.audit_log),
        }

    def act(self, name: str, value: str | None = None) -> None:
        if name == "select_case":
            self.selected = True
        elif name == "choose_resolution":
            self.resolution = value
        elif name == "add_note":
            self.note = value or ""
        elif name == "submit":
            if not self.resolution or not self.note:
                raise ValueError("resolution and note are required")
            self.status = "Resolved"
            self.audit_log.append(f"resolution={self.resolution}; note={self.note}")
        else:
            raise ValueError(f"unknown action: {name}")


def decide(state: dict[str, Any]) -> tuple[str, str | None]:
    case = state["case"]
    if not state["selected"]:
        return "select_case", None
    if case["issue"] != "wrong_item" or not case["replacement_in_stock"]:
        raise NeedsHumanReview("policy or inventory does not allow automatic replacement")
    if state["resolution"] is None:
        return "choose_resolution", "Replacement"
    if not state["note"]:
        return "add_note", "Wrong item confirmed; replacement available under policy."
    if state["status"] != "Resolved":
        return "submit", None
    return "done", None


def validate(browser: MockBrowser) -> None:
    state = browser.observe()
    assert state["status"] == "Resolved"
    assert state["resolution"] == "Replacement"
    assert state["audit_log"], "audit log must not be empty"


def main() -> int:
    case_path = Path(sys.argv[1] if len(sys.argv) > 1 else "cases/case.json")
    browser = MockBrowser(json.loads(case_path.read_text(encoding="utf-8")))
    try:
        for turn in range(1, 7):
            action, value = decide(browser.observe())
            print(f"turn={turn} action={action} value={value}")
            if action == "done":
                validate(browser)
                print("STATUS=passed case=CASE-4107 resolution=Replacement audit=present")
                return 0
            browser.act(action, value)
    except NeedsHumanReview as exc:
        print(f"STATUS=needs_human_review reason={exc}")
        return 2
    print("STATUS=failed reason=max_turns_exceeded")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())

python3 mock_loop.py cases/case.json

turn=1 action=select_case value=None
turn=2 action=choose_resolution value=Replacement
turn=3 action=add_note value=Wrong item confirmed; replacement available under policy.
turn=4 action=submit value=None
turn=5 action=done value=None
STATUS=passed case=CASE-4107 resolution=Replacement audit=present

{
  "case_id": "CASE-4108",
  "order_id": "ORD-1043",
  "issue": "wrong_item",
  "replacement_in_stock": false,
  "policy": "wrong_item_and_in_stock_means_replacement"
}

python3 mock_loop.py cases/case-out-of-stock.json

turn=1 action=select_case value=None
STATUS=needs_human_review reason=policy or inventory does not allow automatic replacement

python3 mock_loop.py cases/case-out-of-stock.json
status=$?

if [ "$status" -eq 2 ]; then
  printf '%s\n' 'Route this case to a human reviewer.'
fi

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Support Console</title>
  <style>
    body { font: 16px/1.5 system-ui, sans-serif; max-width: 760px; margin: 40px auto; padding: 0 20px; }
    section { border: 1px solid #ccc; border-radius: 10px; padding: 18px; margin: 16px 0; }
    textarea { width: 100%; min-height: 90px; }
    button { margin: 6px 6px 6px 0; padding: 8px 14px; }
    #audit-log { white-space: pre-wrap; background: #f5f5f5; padding: 12px; }
  </style>
</head>
<body>
  <h1>Customer Support Console</h1>
  <section aria-labelledby="case-title">
    <h2 id="case-title">CASE-4107 · Order ORD-1042</h2>
    <p>Issue: Customer received the wrong item.</p>
    <p>Expected item: Blue Mug. Received item: Red Mug.</p>
    <p>Inventory: Blue Mug replacement is available.</p>
    <p>Policy: Wrong item + replacement in stock → Replacement.</p>
    <p>Status: <strong id="status">Open</strong></p>
  </section>

  <section aria-labelledby="resolution-title">
    <h2 id="resolution-title">Resolution</h2>
    <button id="replacement" type="button">Choose Replacement</button>
    <p>Selected action: <strong id="selected-action">None</strong></p>
    <label for="note">Internal note</label>
    <textarea id="note" placeholder="Explain why this resolution follows policy"></textarea>
    <br>
    <button id="submit" type="button">Submit resolution</button>
    <p id="message" role="status"></p>
  </section>

  <section aria-labelledby="audit-title">
    <h2 id="audit-title">Audit log</h2>
    <div id="audit-log">No entries</div>
  </section>

  <script>
    let resolution = null;
    const status = document.querySelector("#status");
    const selectedAction = document.querySelector("#selected-action");
    const note = document.querySelector("#note");
    const message = document.querySelector("#message");
    const auditLog = document.querySelector("#audit-log");

    document.querySelector("#replacement").addEventListener("click", () => {
      resolution = "Replacement";
      selectedAction.textContent = resolution;
      message.textContent = "";
    });

    document.querySelector("#submit").addEventListener("click", () => {
      if (!resolution || !note.value.trim()) {
        message.textContent = "Resolution and internal note are required.";
        return;
      }
      status.textContent = "Resolved";
      auditLog.textContent = `resolution=${resolution}; note=${note.value.trim()}`;
      message.textContent = "Resolution recorded successfully.";
    });
  </script>
</body>
</html>

python3 -m http.server 8000 --bind 127.0.0.1 --directory toy_app

http://127.0.0.1:8000

[project]
name = "browser-agent-tutorial"
version = "0.1.0"
description = "A bounded browser-agent tutorial based on OpenAI Agents SDK and Playwright MCP"
requires-python = ">=3.13"
dependencies = [
  "openai-agents==0.19.0",
]

uv sync

node --version
npx --version

export OPENAI_API_KEY='replace-with-your-key'
export OPENAI_API_VERSION='replace-with-your-api-version'
export OPENAI_API_BASE='https://your-resource.openai.azure.com/'
export OPENAI_MODEL='your-azure-deployment-name'
export APP_URL='http://127.0.0.1:8000'

from __future__ import annotations

import asyncio
import os

from agents import (
    Agent,
    ModelSettings,
    Runner,
    set_default_openai_api,
    set_default_openai_client,
)
from agents.mcp import MCPServerStdio
from openai import AsyncAzureOpenAI
from openai.types.shared import Reasoning


AGENT_INSTRUCTIONS = """
You can interact with a web browser.
Stay inside the supplied APP_URL.
Do not submit a resolution unless the visible policy and inventory support it.
After submitting, verify both the case status and the audit log.
If required evidence is missing or contradictory, stop and request human review.
""".strip()


async def main() -> None:
    app_url = os.environ.get("APP_URL", "http://127.0.0.1:8000")
    model = os.environ["OPENAI_MODEL"]

    azure_client = AsyncAzureOpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        api_version=os.environ["OPENAI_API_VERSION"],
        azure_endpoint=os.environ["OPENAI_API_BASE"],
    )
    set_default_openai_client(azure_client)
    set_default_openai_api("responses")

    playwright_server = MCPServerStdio(
        name="Playwright MCP",
        params={
            "command": "npx",
            "args": [
                "-y",
                "@playwright/[email protected]",
                "--browser",
                "chrome",
            ],
        },
    )

    agent = Agent(
        name="Support Console Browser Agent",
        model=model,
        model_settings=ModelSettings(
            reasoning=Reasoning(effort="medium"),
        ),
        instructions=AGENT_INSTRUCTIONS,
        mcp_servers=[playwright_server],
    )

    task = f"""
Open {app_url} and resolve CASE-4107 for order ORD-1042.

The customer received the wrong item. Use only information visible in the
application to determine whether a replacement is allowed. If allowed, choose
the resolution, add a concise internal note, and submit it.

Before reporting completion, verify that:
1. the case status is Resolved;
2. the selected action is Replacement;
3. the audit log contains the resolution entry.

If any condition is not satisfied, do not claim success.
""".strip()

    async with playwright_server:
        result = await Runner.run(agent, task, max_turns=20)

    print("FINAL_OUTPUT")
    print(result.final_output)
    print("TRACE_ITEMS")
    for item in result.new_items:
        print(type(item).__name__, item)


if __name__ == "__main__":
    asyncio.run(main())

uv run python real_agent.py

FINAL_OUTPUT
Resolved CASE-4107 for order ORD-1042 with Replacement.
Verified status=Resolved and confirmed the audit log entry.

TRACE_ITEMS
...

@playwright/mcp@latest

uv run python -c "import agents; print(agents.__file__)"
npx -y @playwright/[email protected] --help