Intro#

1
dependencies = [
2
    "playwright>=1.47.0",        # 浏览器引擎
3
    "playwright-stealth>=1.0.6", # 反检测
4
    "pyautogui>=0.9.54",         # 桌面窗口控制、坐标兜底点击
5
    "openai>=1.40.0",            # AI 决策（OpenAI 兼容协议）
6
    "transitions>=0.9.2",        # 有限状态机
7
    "fastapi>=0.115.0",          # HTTP API 层
8
    "uvicorn>=0.30.6",           # ASGI 服务器
9
    "pywebview>=5.1",            # 桌面窗口壳（内嵌 WebView2）
10
    "pydantic>=2.8.2",           # 配置解析 + 数据结构
11
    "sqlalchemy>=2.0.32",        # 任务/步骤持久化
12
    "tenacity>=8.5.0",           # LLM 调用重试
13
]
14

15
[tool.uv]
16
package = false

1
uv sync --all-groups
2
uv run playwright install chromium

1
WEBAGENT_AGENT__MAX_STEPS=25
2
→ 映射到 Settings().agent.max_steps = 25

1
if getattr(sys, "frozen", False):  # PyInstaller 打包模式
2
    PROJECT_ROOT = Path(sys.executable).resolve().parent
3
else:
4
    PROJECT_ROOT = Path(__file__).resolve().parents[2]

1
# ❌ 类定义时就创建了实例，所有 Settings 共享同一个子对象
2
agent: AgentSettings = AgentSettings()
3

4
# ✅ 每次创建新 Settings 时才创建新的 AgentSettings
5
agent: AgentSettings = Field(default_factory=AgentSettings)

1
@lru_cache(maxsize=1)
2
def get_settings() -> Settings:
3
    for d in (DATA_DIR, LOG_DIR, SCREENSHOT_DIR):
4
        d.mkdir(parents=True, exist_ok=True)  # 幂等：目录已存在也不报错
5
    return Settings()

1
ActionType=Literal[
2
    "click", "input", "goto", "press_key", "wait",
3
    "scroll", "back", "refresh", "done", "error"
4
]
5

6
class Action(BaseModel):
7
    action_type: ActionType
8
    dom_selector: str | None = None
9
    screen_x:int |None=None
10
    screen_y:int |None=None
11
    # ...
12

13
    def is_terminal(self)->bool:
14
        return self.action_type in ("done","error")

1
self._browser=self._pw.chromium.launch(
2
    headless=self.settings.headless,
3
    args=[
4
        "--disable-blink-features=AutomationControlled",
5
        "--start-maximized",
6
        # ...
7
    ]
8
)
9
self._context=self._browser.new_context(
10
    viewport={"width":1440,"height":900},
11
    user_agent=_DEFAULT_UA,
12
    locale='zh-CN',
13
    timezone_id='Asia/Shanghai',
14
)
15
Stealth().apply_stealth_sync(self._context)

1
except Exception as e:
2
    msg=str(e)
3
    if "0" in msg and ("操作成功完成" in msg or "successfully" in msg.lower()):
4
        log.debug(f"已激活此窗口（win32误报）：{target.title}")
5
    else:
6
        raise BrowserError(f"激活窗口失败: {e}")

1
@property
2
def page(self)->Page:
3
    if self._page is None:
4
        raise BrowserError("浏览器未启动")
5
    try:
6
        closed=self._page.is_closed()
7
    except Exception:
8
        closed=True
9
    if closed and self._context is not None:
10
        pages=[p for p in self._context.pages if not p.is_closed()]
11
        if pages:
12
            self._page=pages[-1]  # 自动切换到最新的 page
13
        else:
14
            raise BrowserError("context中没有可用的page")
15
    return self._page

1
// 清理上一次注入的 data-wa-id，避免污染
2
document.querySelectorAll('[data-wa-id]').forEach(el => el.removeAttribute('data-wa-id'));
3

4
// 标记 data-wa-id，作为稳定 selector 的兜底
5
el.setAttribute('data-wa-id', String(idx));
6
let selector = `[data-wa-id="${idx}"]`;

1
// 排序：视口内优先 → 面积大优先 → DOM 顺序
2
result.sort((a, b) => {
3
  if (a.inViewport !== b.inViewport) return a.inViewport ? -1 : 1;
4
  if (a.area !== b.area) return b.area - a.area;
5
  return a.index - b.index;
6
});
7
// ...
8
elements: result.slice(0, maxElements),

1
if max(img.size)>max_edge:
2
    ratio=max_edge/max(img.size)
3
    new_size=(int(img.size[0]*ratio),int(img.size[1]*ratio))
4
    img=img.resize(new_size,Image.LANCZOS)

1
# ❌ 不用字典映射：10种动作=10个分支，每加一种动作都要回来改主入口
2
if action.action_type == "click":
3
    self._click(action)
4
elif action.action_type == "input":
5
    self._input(action)
6
# ...
7

8
# ✅ 字典映射后：查表取函数 → 判空兜底 → 统一调用
9
handler={
10
    "click":self._click,
11
    "input":self._input,
12
    "goto":self._goto,
13
    # ...
14
    "done":lambda _a:True,
15
    "error":lambda _a:True,
16
}.get(action.action_type)
17
if handler is None:
18
    raise ActionExecutionError(f"未知action_type: {action.action_type}")
19
handler(action)

1
retries=0 if action.is_terminal() else self.cfg.action_retry
2
for attempt in range(retries+1):
3
    try:
4
        handler(action)
5
        return True
6
    except Exception as e:
7
        log.warning(f"执行动作第{attempt+1}次失败: {e}")
8
        time.sleep(0.4)

1
def _click(self,a:Action)->None:
2
    if a.dom_selector:
3
        try:
4
            self._dom_click(a.dom_selector)   # 第一层：DOM 精准定位
5
            return
6
        except Exception as e:
7
            log.warning(f"点击DOM元素失败，尝试坐标兜底: {e}")
8
    if a.screen_x is not None and a.screen_y is not None:
9
        self._coord_click(a.screen_x,a.screen_y)  # 第二、三层：坐标点击
10
        return

1
offset=self.page.evaluate(
2
    "()=>({sx:window.screenX,sy:window.screenY,"
3
    "ox:window.outerWidth-windows.innerWidth,"
4
    "oy:window.outerHeight-windows.innerHeight})"
5
)
6
abs_x=int(offset["sx"]+offset["ox"]+x)
7
abs_y=int(offset["sy"]+offset["oy"]+y)

1
- 必须输出单个 JSON 对象，严禁输出 JSON 以外的任何内容
2
- 优先使用 dom_selector，仅在确实无 selector 可用时给出 screen_x/screen_y
3
- 不要重复执行历史中已经失败/无效的动作
4
- input 元素是输入框，点击输入框只会聚焦，不会提交表单；
5
  输入完成后优先用 press_key（Enter）提交

1
text = raw.strip()
2
if text.startswith("```"):          # 1. 剥掉 ```json 代码围栏
3
    text = text.strip("`")
4
    if text.lower().startswith("json"):
5
        text = text[4:]
6
    text = text.strip()
7
if not text.startswith("{"):        # 2. 前后夹带自然语言 → 首尾大括号截取
8
    start=text.find("{")
9
    end=text.find("}")
10
    if start>=0 and end>start:
11
        text=text[start:end+1]
12
# 3. json.loads + Action.model_validate 双重校验

1
try:
2
    resp=self._client.chat.completions.create(..., response_format={"type":"json_object"})
3
except TypeError:
4
    resp=self._client.chat.completions.create(...)  # 回落

1
if last_error:
2
    reflexion_parts.append(
3
        "【上一步失败】\n"
4
        f"错误：{last_error}\n"
5
        "请换思路：换selector/滚动/等待/改用screen_x/y坐标兜底"
6
    )
7
else:
8
    if history:
9
        page_still=(prev_url==snapshot.url) and (prev_title==snapshot.title)
10
        if page_still:
11
            reflexion_parts.append(
12
                "【警告】上一步动作已执行但页面URL和标题均未变化，该动作很可能无效！"
13
                "严禁重复相同动作，请改选其他元素或换操作方式"
14
            )
15
        else:
16
            reflexion_parts.append("【上一步成功】...请检查任务是否已经达成")

1
_STATES = ["idle", "perceiving", "reasoning", "executing", "checking", "done", "failed"]
2

3
transitions=[
4
    {"trigger": "start",      "source": "idle",       "dest": "perceiving"},
5
    {"trigger": "perceived",  "source": "perceiving", "dest": "reasoning"},
6
    {"trigger": "decided",    "source": "reasoning",  "dest": "executing"},
7
    {"trigger": "executed",   "source": "executing",  "dest": "checking"},
8
    {"trigger": "loop",       "source": "checking",   "dest": "perceiving"},
9
    {"trigger": "finish_done","source": "*",          "dest": "done"},
10
    {"trigger": "fail",       "source": "*",          "dest": "failed"},
11
]

1
cur_key=(action.action_type,action.dom_selector,action.input_text,action.url)
2
if ok and not page_changed and cur_key == ctx.repeat_key:
3
    ctx.repeat_count+=1
4
else:
5
    ctx.repeat_count=1 if ok else 0
6
    ctx.repeat_key=cur_key if ok else None
7

8
if ctx.repeat_count>=3:
9
    log.error("检测到死循环：同动作连续 {} 次执行且页面未变化，终止任务", ctx.repeat_count)
10
    ctx.task.status=TaskStatus.FAILED
11
    self.fail()
12
    break

1
AgentLoop(
2
    on_step=self._on_step,      # 每执行一步就回调，api 层把它转成 WS 事件推给前端
3
    cancel_event=self._cancel_event,  # 取消信号，REST 接口置位即可
4
)