1097 lines
51 KiB
Python
1097 lines
51 KiB
Python
"""Browser Use 执行器：独立 Chrome 窗口，会话保活，支持暂停后续跑。
|
||
|
||
一个 task 对应一个 LiveSession（Agent + Browser）。
|
||
- ask_user 工具以 is_done 结束本轮 agent.run()，浏览器不关闭；
|
||
- 用户回答后 orchestrator 再次调用 run_browser_step(continuation=...)，
|
||
这里用 agent.add_new_task 把回答喂回同一个 Agent，上下文和页面状态都保留；
|
||
- 每一步通过 register_new_step_callback 落一张截图和一行进度，便于事后复盘。
|
||
"""
|
||
|
||
import base64
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import math
|
||
import os
|
||
import re
|
||
import time
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from contextlib import contextmanager
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import urlsplit
|
||
|
||
import httpx
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
from backend.browser.llm_factory import create_llm, get_llm_config, use_vision_for_model
|
||
from backend.browser import diagnostics
|
||
from backend.config import settings
|
||
from backend.browser.progress_guard import ProgressGuard
|
||
from backend.browser.preflight_gate import PreflightGate, PreflightScopeError
|
||
from backend.browser.tools import (
|
||
_write_json_atomic,
|
||
build_prospecting_tools,
|
||
build_tools,
|
||
clear_task_context,
|
||
get_task_artifacts,
|
||
pop_pending_blocker,
|
||
pop_pending_question,
|
||
set_task_context,
|
||
set_prospecting_context,
|
||
refresh_prospecting_request,
|
||
set_prospecting_run_context,
|
||
record_navigation_observation,
|
||
)
|
||
from backend.orchestrator.progress_doc import append_progress, task_doc_dir
|
||
|
||
|
||
@dataclass
|
||
class LiveSession:
|
||
agent: Any
|
||
browser: Any
|
||
started_at: float = field(default_factory=time.time)
|
||
embedded: bool = False
|
||
target_id: str = ""
|
||
tool_profile: str = "legacy"
|
||
allowed_domains: tuple[str, ...] = ()
|
||
allow_initial_blank: bool = False
|
||
last_verified_url: str = ""
|
||
preflight_policy: Any = None
|
||
|
||
|
||
_live: dict[str, LiveSession] = {}
|
||
_revision_views: dict[str, LiveSession] = {}
|
||
MAX_TASK_BROWSER_STEPS = 60
|
||
_stalled_tasks: set[str] = set()
|
||
_evidence_failed_tasks: set[str] = set()
|
||
|
||
|
||
class ManagedTargetError(RuntimeError):
|
||
def __init__(self, code: str):
|
||
self.code = code
|
||
super().__init__(f"受管浏览器页面未通过核对（{code}），已停止本段执行并保留页面与登录状态。请在该任务浏览器打开实际来源页，或明确提供实际入口后继续。")
|
||
|
||
|
||
def _managed_business_url(url: str, session: LiveSession) -> bool:
|
||
if url == "about:blank":
|
||
return session.allow_initial_blank
|
||
try:
|
||
parts = urlsplit(url)
|
||
if parts.scheme not in {"http", "https"} or not parts.hostname or parts.hostname == "ui" or parts.username or parts.password:
|
||
return False
|
||
host = parts.hostname.lower()
|
||
domains = [pattern.lower().removeprefix("https://").removeprefix("http://") for pattern in session.allowed_domains]
|
||
return not domains or any(host == pattern or pattern.startswith("*.") and
|
||
(host == pattern[2:] or host.endswith(pattern[1:])) for pattern in domains)
|
||
except (ValueError, TypeError):
|
||
return False
|
||
|
||
|
||
async def _focus_managed_target(task_id: str, session: LiveSession, *, verify_host: bool = False):
|
||
"""Bind to the task's existing view, never whichever CDP page is frontmost."""
|
||
if not session.embedded:
|
||
return None
|
||
if not session.target_id:
|
||
raise ManagedTargetError("managed_target_missing")
|
||
try:
|
||
if verify_host:
|
||
health = await _ctrl("GET", "/health")
|
||
views = health.get("views") if isinstance(health, dict) else None
|
||
matches = [row for row in views or [] if isinstance(row, dict) and row.get("taskId") == task_id]
|
||
if len(matches) != 1 or matches[0].get("targetId") != session.target_id:
|
||
raise ManagedTargetError("managed_target_binding_changed")
|
||
info = await asyncio.wait_for(session.browser.cdp_client.send.Target.getTargetInfo(
|
||
params={"targetId": session.target_id}), timeout=10)
|
||
target = info.get("targetInfo") if isinstance(info, dict) else None
|
||
if not isinstance(target, dict) or target.get("targetId") != session.target_id or target.get("type") != "page":
|
||
raise ManagedTargetError("managed_target_not_page")
|
||
if not _managed_business_url(target.get("url", ""), session):
|
||
raise ManagedTargetError("managed_target_not_business_page")
|
||
session.last_verified_url = target["url"]
|
||
if target["url"] != "about:blank":
|
||
session.allow_initial_blank = False
|
||
if getattr(session.browser, "agent_focus_target_id", None) != session.target_id:
|
||
# The pinned public focus event also invalidates DOM/selector caches.
|
||
# Its optional diagnostic URL is intentionally empty, not a signed URL.
|
||
from browser_use.browser.events import AgentFocusChangedEvent
|
||
await session.browser.event_bus.dispatch(AgentFocusChangedEvent(target_id=session.target_id, url=""))
|
||
cdp_session = await session.browser.get_or_create_cdp_session(session.target_id, focus=True)
|
||
if getattr(session.browser, "agent_focus_target_id", None) != session.target_id:
|
||
raise ManagedTargetError("managed_target_focus_failed")
|
||
return cdp_session
|
||
except ManagedTargetError:
|
||
raise
|
||
except Exception as error:
|
||
# A detached target cannot fall back to the application UI or another
|
||
# task. No provider request, guessed navigation or view disposal follows.
|
||
raise ManagedTargetError("managed_target_unavailable") from error
|
||
|
||
|
||
def _managed_target_result(task_id: str, error: ManagedTargetError, *, history: dict | None = None) -> dict:
|
||
append_progress(task_id, "受管页面核对失败", "blocked", str(error), level="warning")
|
||
return {"verdict": "failed", "summary": str(error), "artifacts": get_task_artifacts(task_id),
|
||
"history": history or {"steps_this_run": 0, "is_done": False, "is_successful": False,
|
||
"managed_target_error": error.code}, "stop_reason": "managed_target_unavailable"}
|
||
|
||
|
||
def _assert_managed_context(agent, url: str) -> None:
|
||
session = getattr(agent, "_managed_session", None)
|
||
if session and (getattr(session.browser, "agent_focus_target_id", None) != session.target_id
|
||
or not _managed_business_url(url, session) or url != session.last_verified_url):
|
||
error = ManagedTargetError("managed_context_changed")
|
||
agent._managed_target_failure = error
|
||
agent.state.stopped = True
|
||
raise error
|
||
|
||
|
||
class _ManagedActionBoundary:
|
||
"""Delegate all Tools behavior; guard its actual per-action entry point."""
|
||
def __init__(self, delegate, check, after=None):
|
||
self._delegate, self._check, self._after = delegate, check, after
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self._delegate, name)
|
||
|
||
async def act(self, *args, **kwargs):
|
||
action = kwargs.get("action", args[0] if args else None)
|
||
payload = action.model_dump(exclude_none=True) if action is not None else {}
|
||
name = next(iter(payload)) if len(payload) == 1 else "unknown_action"
|
||
rejected = await self._check(action)
|
||
if rejected is not None:
|
||
diagnostics.record_browser_action(name, "blocked")
|
||
return rejected
|
||
action_id = diagnostics.record_browser_action(name, "invoked")
|
||
try:
|
||
result = await self._delegate.act(*args, **kwargs)
|
||
except BaseException:
|
||
diagnostics.record_browser_action(name, "error", action_id=action_id)
|
||
raise
|
||
diagnostics.record_browser_action(name, "returned", action_id=action_id)
|
||
if self._after:
|
||
rejected = await self._after()
|
||
if rejected is not None:
|
||
return rejected
|
||
return result
|
||
|
||
|
||
def _preflight_scope_result(task_id, error, *, history=None):
|
||
append_progress(task_id, "只读预检范围核对", "blocked", str(error), level="warning")
|
||
return {"verdict": "failed", "summary": str(error), "artifacts": get_task_artifacts(task_id),
|
||
"history": history or {"steps_this_run": 0, "is_done": False, "is_successful": False},
|
||
"stop_reason": "preflight_scope_violation"}
|
||
|
||
|
||
@contextmanager
|
||
def preserve_managed_view_for_revision(task_id: str, *, enabled: bool = False):
|
||
"""Protect only this existing view while its old job fully cancels."""
|
||
session = _live.get(task_id)
|
||
preserve = bool(enabled and session and session.embedded)
|
||
if preserve:
|
||
_revision_views[task_id] = session
|
||
try:
|
||
yield preserve
|
||
finally:
|
||
if preserve and _revision_views.get(task_id) is session:
|
||
_revision_views.pop(task_id, None)
|
||
|
||
|
||
def _prospecting_agent_class(base):
|
||
"""Adapter for the pinned 0.13.10 run loop, which accepts an infinite horizon.
|
||
|
||
AgentStepInfo is a dataclass, not a validated finite integer. Keep the
|
||
upstream lifecycle, cancellation and failure handling, but omit budget
|
||
instructions from model context for goal-based tasks. No dependency patch.
|
||
"""
|
||
class GoalBasedAgent(base):
|
||
async def get_model_output(self, input_messages):
|
||
# The pinned framework collapses failed-schema details into a
|
||
# generic history line. Preserve safe field/type feedback on the
|
||
# normal next step; never start a separate repair request.
|
||
hint = diagnostics.pending_correction_hint() if getattr(self, "_goal_based", False) else ""
|
||
if hint:
|
||
from browser_use.llm.messages import UserMessage
|
||
if not input_messages or getattr(input_messages[-1], "text", None) != hint:
|
||
input_messages = [*input_messages, UserMessage(content=hint)]
|
||
return await super().get_model_output(input_messages)
|
||
|
||
async def _prepare_context(self, step_info=None):
|
||
self._goal_step_info = step_info
|
||
guard = getattr(self, "_managed_target_guard", None)
|
||
if guard:
|
||
await guard()
|
||
scope = getattr(self, "_preflight_scope_gate", None)
|
||
if scope:
|
||
try:
|
||
await scope.observe()
|
||
except PreflightScopeError:
|
||
self.state.stopped = True
|
||
raise
|
||
summary = await super()._prepare_context(None if getattr(self, "_goal_based", False) else step_info)
|
||
session = getattr(self, "_managed_session", None)
|
||
_assert_managed_context(self, getattr(summary, "url", ""))
|
||
if session:
|
||
self._managed_observed_url = summary.url
|
||
return summary
|
||
|
||
async def _update_action_models_for_page(self, url):
|
||
# In the pinned prepare path this hook runs after state capture but
|
||
# before prepare_step_state, compression or state message creation.
|
||
_assert_managed_context(self, url)
|
||
return await super()._update_action_models_for_page(url)
|
||
|
||
async def multi_act(self, actions, *args, **kwargs):
|
||
guard = getattr(self, "_managed_target_guard", None)
|
||
scope = getattr(self, "_preflight_scope_gate", None)
|
||
if guard or scope:
|
||
session = getattr(self, "_managed_session", None)
|
||
async def check_boundary(action=None):
|
||
from browser_use import ActionResult
|
||
if guard:
|
||
changed_focus = getattr(session.browser, "agent_focus_target_id", None) != session.target_id
|
||
await guard()
|
||
if changed_focus or (getattr(self, "_managed_observed_url", None)
|
||
and self._managed_observed_url != session.last_verified_url):
|
||
return ActionResult(error="页面焦点或地址已变化，本步旧动作未执行；请重新观察受管页面后定位。")
|
||
if scope:
|
||
try:
|
||
message = await scope.before_action(action)
|
||
except PreflightScopeError:
|
||
self.state.stopped = True
|
||
raise
|
||
if message:
|
||
return ActionResult(error=message)
|
||
return None
|
||
rejected = await check_boundary()
|
||
if rejected is not None:
|
||
return [rejected]
|
||
for action in actions:
|
||
payload = action.model_dump(exclude_none=True)
|
||
navigation = payload.get("navigate")
|
||
if session and isinstance(navigation, dict) and (navigation.get("new_tab") is True
|
||
or urlsplit(navigation.get("url", "")).scheme == "app"
|
||
or urlsplit(navigation.get("url", "")).hostname == UI_HOST):
|
||
error = ManagedTargetError("managed_navigation_not_allowed")
|
||
self._managed_target_failure = error
|
||
self.state.stopped = True
|
||
raise error
|
||
original_tools = self.tools
|
||
async def after_boundary():
|
||
if scope:
|
||
try:
|
||
await scope.observe()
|
||
except PreflightScopeError as error:
|
||
from browser_use import ActionResult
|
||
self.state.stopped = True
|
||
return ActionResult(error=str(error))
|
||
return None
|
||
boundary = _ManagedActionBoundary(original_tools, check_boundary, after_boundary if scope else None)
|
||
self.tools = boundary
|
||
try:
|
||
# Keep the pinned batch loop intact, including inter-action
|
||
# waits and sequence/DOM guards. Tools.act is checked again
|
||
# after those awaits, immediately before each action.
|
||
return await super().multi_act(actions, *args, **kwargs)
|
||
finally:
|
||
if self.tools is boundary:
|
||
self.tools = original_tools
|
||
return await super().multi_act(actions, *args, **kwargs)
|
||
|
||
async def _maybe_compact_messages(self, step_info=None):
|
||
from backend.prospecting.usage import usage_scope
|
||
with usage_scope(phase="compression"):
|
||
return await super()._maybe_compact_messages(
|
||
getattr(self, "_goal_step_info", step_info) if getattr(self, "_goal_based", False) else step_info
|
||
)
|
||
|
||
async def _force_done_after_failure(self):
|
||
maximum = getattr(self.settings, "max_failures", None)
|
||
consecutive = getattr(self.state, "consecutive_failures", 0)
|
||
if (type(maximum) is int and maximum > 0 and consecutive >= maximum
|
||
and getattr(self.settings, "final_response_after_failure", False)):
|
||
self._program_failure_limit = {
|
||
"reason": "model_schema_error" if diagnostics.consecutive_schema_errors() >= maximum else "model_failure_limit",
|
||
"consecutive_failures": consecutive, "max_failures": maximum,
|
||
}
|
||
cause = _provider_failure_reason(" ".join(str(getattr(row, "error", "") or "")
|
||
for row in getattr(self.state, "last_result", None) or []))
|
||
if cause:
|
||
self._program_failure_limit["cause"] = cause
|
||
return await super()._force_done_after_failure()
|
||
|
||
def _log_agent_event(self, max_steps, agent_run_error=None):
|
||
# Telemetry schema requires an integer; zero means no total ceiling.
|
||
return super()._log_agent_event(0 if math.isinf(max_steps) else max_steps, agent_run_error)
|
||
|
||
return GoalBasedAgent
|
||
|
||
# ---------- 内嵌模式（Electron 提供浏览器） ----------
|
||
# Electron 拉起后端时传入这两个变量；后端单独运行时为空，走独立 Chrome。
|
||
|
||
CDP_URL = os.getenv("CDP_URL", "").strip()
|
||
CTRL_URL = os.getenv("ELECTRON_CTRL_URL", "").strip().rstrip("/")
|
||
# 内嵌模式下不让 Agent 切换 / 关闭标签页（界面页也是一个标签页），也不需要文件和搜索动作
|
||
EMBEDDED_EXCLUDED_ACTIONS = ["switch", "close", "search", "upload_file", "write_file", "replace_file", "read_file"]
|
||
UI_HOST = "ui" # 界面页 app://ui/…，必须进 browser-use 的域名白名单，否则会被当成非法标签页关掉
|
||
|
||
|
||
def embedded_mode() -> bool:
|
||
return bool(CDP_URL and CTRL_URL)
|
||
|
||
|
||
async def _ctrl(method: str, path: str, body: dict | None = None) -> dict:
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
resp = await client.request(method, CTRL_URL + path, json=body)
|
||
resp.raise_for_status()
|
||
return resp.json() if resp.content else {}
|
||
|
||
|
||
def _partition_for(profile_path: str) -> str:
|
||
"""把「人 × 系统 × 角色」的 profile 路径映射成 Electron 分区名（ASCII 安全）。"""
|
||
parts = Path(profile_path).parts[-2:]
|
||
raw = "-".join(parts)
|
||
safe = re.sub(r"[^A-Za-z0-9_-]+", "", raw)[:40]
|
||
return f"sa-{safe}-{hashlib.md5(raw.encode('utf-8')).hexdigest()[:8]}"
|
||
|
||
# Electron 上报的浏览器停靠区域（屏幕坐标）：{"left","top","width","height"}
|
||
_dock_bounds: dict | None = None
|
||
|
||
|
||
# ---------- 停靠 / 展开收起 ----------
|
||
|
||
|
||
def set_dock_bounds(bounds: dict | None) -> dict | None:
|
||
global _dock_bounds
|
||
if bounds:
|
||
_dock_bounds = {k: int(bounds[k]) for k in ("left", "top", "width", "height")}
|
||
else:
|
||
_dock_bounds = None
|
||
return _dock_bounds
|
||
|
||
|
||
def get_dock_bounds() -> dict | None:
|
||
return _dock_bounds
|
||
|
||
|
||
async def _window_id(browser: Any) -> int:
|
||
target_id = getattr(browser, "agent_focus_target_id", None)
|
||
params = {"targetId": target_id} if target_id else None
|
||
res = await browser.cdp_client.send.Browser.getWindowForTarget(params=params)
|
||
return int(res["windowId"])
|
||
|
||
|
||
async def set_browser_window(task_id: str, action: str) -> dict:
|
||
"""show：恢复窗口、按停靠区域摆放、置顶；hide：最小化。只对有活动会话的任务有效。
|
||
内嵌模式下改为让 Electron 展开 / 收起浏览器面板。"""
|
||
session = _live.get(task_id)
|
||
if not session:
|
||
raise RuntimeError("该任务当前没有活动的浏览器会话")
|
||
if session.embedded:
|
||
if action not in ("show", "hide"):
|
||
raise ValueError(f"未知动作: {action}")
|
||
await _ctrl("POST", f"/views/{task_id}/{'reveal' if action == 'show' else 'hide'}")
|
||
return {"task_id": task_id, "action": action, "embedded": True}
|
||
browser = session.browser
|
||
cdp = browser.cdp_client.send
|
||
window_id = await _window_id(browser)
|
||
if action == "hide":
|
||
await cdp.Browser.setWindowBounds(params={"windowId": window_id, "bounds": {"windowState": "minimized"}})
|
||
elif action == "show":
|
||
await cdp.Browser.setWindowBounds(params={"windowId": window_id, "bounds": {"windowState": "normal"}})
|
||
if _dock_bounds:
|
||
await cdp.Browser.setWindowBounds(
|
||
params={"windowId": window_id, "bounds": {**_dock_bounds, "windowState": "normal"}}
|
||
)
|
||
target_id = getattr(browser, "agent_focus_target_id", None)
|
||
if target_id:
|
||
try:
|
||
await cdp.Target.activateTarget(params={"targetId": target_id})
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
else:
|
||
raise ValueError(f"未知动作: {action}")
|
||
res = await cdp.Browser.getWindowBounds(params={"windowId": window_id})
|
||
return {"task_id": task_id, "action": action, "bounds": res.get("bounds", res)}
|
||
|
||
|
||
# ---------- 用户接管 ----------
|
||
|
||
_taken_over: set[str] = set()
|
||
|
||
|
||
def is_taken_over(task_id: str) -> bool:
|
||
return task_id in _taken_over
|
||
|
||
|
||
async def takeover(task_id: str) -> None:
|
||
"""用户接管：让 Agent 在下一步前停住（browser-use pause），浏览器交给用户操作。"""
|
||
session = _live.get(task_id)
|
||
if not session:
|
||
raise RuntimeError("该任务当前没有活动的浏览器会话")
|
||
try:
|
||
session.agent.pause()
|
||
except Exception: # noqa: BLE001 Agent 不在运行循环里时 pause 只是置标志
|
||
pass
|
||
_taken_over.add(task_id)
|
||
if session.embedded:
|
||
try:
|
||
await _ctrl("POST", f"/views/{task_id}/mode", {"mode": "user"})
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
|
||
async def takeover_done(task_id: str) -> None:
|
||
session = _live.get(task_id)
|
||
_taken_over.discard(task_id)
|
||
if not session:
|
||
return
|
||
try:
|
||
session.agent.resume()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
if session.embedded:
|
||
try:
|
||
await _ctrl("POST", f"/views/{task_id}/mode", {"mode": "agent"})
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
|
||
# ---------- 会话管理 ----------
|
||
|
||
|
||
def has_live_session(task_id: str) -> bool:
|
||
return task_id in _live
|
||
|
||
|
||
def live_session_ids() -> list[str]:
|
||
return list(_live)
|
||
|
||
|
||
async def close_live_session(task_id: str) -> None:
|
||
if _live.get(task_id) is not None and _revision_views.get(task_id) is _live.get(task_id):
|
||
return
|
||
_taken_over.discard(task_id)
|
||
_stalled_tasks.discard(task_id)
|
||
_evidence_failed_tasks.discard(task_id)
|
||
session = _live.pop(task_id, None)
|
||
if not session:
|
||
return
|
||
if session.embedded:
|
||
# 只断开 CDP 连接，绝不能 kill（那会关掉 Electron 自己的浏览器）；视图由 Electron 释放
|
||
try:
|
||
await session.browser.stop()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
await _ctrl("DELETE", f"/views/{task_id}")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return
|
||
try:
|
||
await session.browser.kill()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
|
||
async def close_all_sessions() -> None:
|
||
for task_id in list(_live):
|
||
await close_live_session(task_id)
|
||
|
||
|
||
# ---------- 可观测：每步截图 + 进度行 ----------
|
||
|
||
|
||
def _action_names(agent_output: Any) -> list[str]:
|
||
names: list[str] = []
|
||
for action in getattr(agent_output, "action", None) or []:
|
||
try:
|
||
names.extend(action.model_dump(exclude_none=True).keys())
|
||
except Exception: # noqa: BLE001
|
||
names.append(str(action)[:40])
|
||
return names
|
||
|
||
|
||
def _progress_url(url: str) -> str:
|
||
"""Display navigation without userinfo, signed query values or fragments."""
|
||
try:
|
||
parsed = urlsplit(url)
|
||
if parsed.scheme in {"http", "https", "app"} and parsed.hostname:
|
||
return f"{parsed.scheme}://{parsed.hostname}{parsed.path}"[:160]
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return "页面地址未记录"
|
||
|
||
|
||
def _make_step_callback(task_id: str, steps_dir: Path):
|
||
# A resumed Agent keeps n_steps, but a new Agent for the same task starts at
|
||
# one again. Allocate task-wide evidence numbers so retries never overwrite
|
||
# a previous screenshot.
|
||
saved_numbers = [
|
||
int(match.group(1))
|
||
for path in steps_dir.glob("step_*.png")
|
||
if (match := re.fullmatch(r"step_(\d+)\.png", path.name))
|
||
]
|
||
evidence_step = max(saved_numbers, default=0)
|
||
guard = ProgressGuard()
|
||
|
||
async def on_step(browser_state_summary: Any, agent_output: Any, step_number: int) -> None:
|
||
nonlocal evidence_step
|
||
evidence_step = max(evidence_step + 1, int(step_number))
|
||
shot_note = ""
|
||
shot = getattr(browser_state_summary, "screenshot", None)
|
||
if shot:
|
||
try:
|
||
data = base64.b64decode(shot)
|
||
while True:
|
||
path = steps_dir / f"step_{evidence_step:03d}.png"
|
||
try:
|
||
with path.open("xb") as handle:
|
||
handle.write(data)
|
||
break
|
||
except FileExistsError:
|
||
evidence_step += 1
|
||
shot_note = f" | 截图 steps/{path.name}"
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
goal = (getattr(agent_output, "next_goal", "") or "").strip()
|
||
evaluation = (getattr(agent_output, "evaluation_previous_goal", "") or "").strip()
|
||
actions = "、".join(_action_names(agent_output)) or "-"
|
||
url = getattr(browser_state_summary, "url", "") or ""
|
||
session = _live.get(task_id)
|
||
if session and session.tool_profile == "prospecting":
|
||
record_navigation_observation(task_id, url)
|
||
try:
|
||
if guard.observe(browser_state_summary, agent_output, get_task_artifacts(task_id)):
|
||
_stalled_tasks.add(task_id)
|
||
session = _live.get(task_id)
|
||
if session and getattr(session.agent, "_goal_based", False):
|
||
session.agent.stop()
|
||
except (AttributeError, TypeError, ValueError):
|
||
pass
|
||
detail = f"Agent 累计步骤：{step_number} | 目标：{goal[:160]} | 动作：{actions} | 页面：{_progress_url(url)}{shot_note}"
|
||
if evaluation:
|
||
detail = f"上步评估：{evaluation[:120]} | {detail}"
|
||
append_progress(task_id, f"步骤 {evidence_step}", "running", detail)
|
||
|
||
return on_step
|
||
|
||
|
||
def _compact_history(history: Any) -> dict:
|
||
def safe(fn, default=None):
|
||
try:
|
||
return fn()
|
||
except Exception: # noqa: BLE001
|
||
return default
|
||
|
||
urls = [u for u in (safe(history.urls, []) or []) if u]
|
||
dedup: list[str] = []
|
||
for u in urls:
|
||
if not dedup or dedup[-1] != u:
|
||
dedup.append(u)
|
||
return {
|
||
"steps": safe(history.number_of_steps, 0),
|
||
"duration_s": round(safe(history.total_duration_seconds, 0.0) or 0.0, 1),
|
||
"is_done": safe(history.is_done, False),
|
||
"is_successful": safe(history.is_successful, None),
|
||
"urls": dedup[:50],
|
||
"actions": (safe(history.action_names, []) or [])[:200],
|
||
"errors": [e for e in (safe(history.errors, []) or []) if e][-20:],
|
||
"final_result": safe(history.final_result, None),
|
||
}
|
||
|
||
|
||
def window_kwargs_for(bounds: dict | None) -> dict[str, Any]:
|
||
"""把 Electron 上报的停靠区域翻译成 browser-use 的窗口参数（独立 Chrome 模式）。
|
||
|
||
browser-use 用 ViewportSize 表示位置：width=x, height=y，且校验坐标必须 >= 0。
|
||
多屏 / 窗口越界时 Electron 报的坐标可能为负（实测 top=-521 会让启动直接失败），这里夹到 0，
|
||
展开时再用 CDP 摆到真实位置。
|
||
"""
|
||
if not bounds:
|
||
return {}
|
||
return {
|
||
"window_size": {"width": max(320, int(bounds["width"])), "height": max(240, int(bounds["height"]))},
|
||
"window_position": {"width": max(0, int(bounds["left"])), "height": max(0, int(bounds["top"]))},
|
||
}
|
||
|
||
|
||
def verdict_from_history(compact: dict, max_steps: int | None) -> tuple[str, str]:
|
||
"""从精简执行历史得出 (verdict, summary)。
|
||
|
||
Agent 主动 done 且没自评失败 → success；done 但自评失败 → failed；
|
||
没 done（跑满步数或中途报错）→ failed，并把模型侧的 402 / 401 翻译成用户能处理的话。
|
||
"""
|
||
final = compact.get("final_result")
|
||
done = bool(compact.get("is_done"))
|
||
ok = compact.get("is_successful")
|
||
if done and ok is not False:
|
||
return "success", str(final or "执行结束")
|
||
if done:
|
||
return "failed", str(final or "Agent 判定任务未成功")
|
||
|
||
errors = compact.get("errors") or []
|
||
last_error = str(errors[-1])[:200] if errors else ""
|
||
if "Insufficient Balance" in last_error or "Error code: 402" in last_error:
|
||
if get_llm_config().get("fallback_api_key"):
|
||
return "failed", "模型服务余额不足：主 Key 与兜底 Key 都返回 402，Agent 无法继续。充值后点「继续执行」。"
|
||
return "failed", "模型服务余额不足（DeepSeek 402），Agent 无法继续。充值、在设置里换 Key 或在 .env 配 DEEPSEEK_API_KEY_FALLBACK 后点「继续执行」。"
|
||
if "Error code: 401" in last_error or "invalid_api_key" in last_error.lower():
|
||
return "failed", "模型服务 Key 无效（401），请在设置里检查 API Key 后点「继续执行」。"
|
||
tail = f"最近错误：{last_error}" if last_error else ""
|
||
return "failed", (f"未在 {max_steps} 步内完成。{tail}" if max_steps is not None else f"执行已中断，已保存部分进度。{tail}")
|
||
|
||
|
||
def _provider_failure_reason(errors: str) -> str | None:
|
||
def http_error(code: int) -> bool:
|
||
return bool(re.search(rf"(?:HTTP|Error code:|status(?:_code)?[=:]?)\s*[:=]?\s*{code}\b", errors, re.I))
|
||
if http_error(402) or "Insufficient Balance" in errors:
|
||
return "model_balance"
|
||
if http_error(401) or "invalid_api_key" in errors.lower():
|
||
return "model_auth"
|
||
if http_error(429):
|
||
return "model_rate_limit"
|
||
return None
|
||
|
||
|
||
def stop_reason_from_history(compact: dict, max_steps: int | None) -> str:
|
||
failure_limit = compact.get("program_failure_limit") or {}
|
||
if failure_limit.get("reason") == "model_schema_error":
|
||
return "model_schema_error"
|
||
current_errors = compact.get("errors_this_run", compact.get("errors")) or []
|
||
if failure_limit.get("reason") == "model_failure_limit":
|
||
# The captured last-result cause belongs to the exact forced handoff.
|
||
# A recovered earlier HTTP error must not override the final failure.
|
||
cause = failure_limit.get("cause")
|
||
if cause in {"model_balance", "model_auth", "model_rate_limit"}:
|
||
return cause
|
||
return _provider_failure_reason(str(current_errors[-1]) if current_errors else "") or "model_failure_limit"
|
||
if compact.get("is_done") and compact.get("is_successful") is not False:
|
||
return "agent_done"
|
||
provider_reason = _provider_failure_reason(" ".join(str(error) for error in current_errors))
|
||
if provider_reason:
|
||
return provider_reason
|
||
if compact.get("is_done"):
|
||
return "agent_reported_failure"
|
||
if max_steps is not None and compact.get("steps_this_run", 0) >= max_steps:
|
||
return "step_limit"
|
||
return "execution_error"
|
||
|
||
|
||
# ---------- 执行 ----------
|
||
|
||
|
||
async def run_browser_step(
|
||
*,
|
||
task_id: str,
|
||
agent_task: str,
|
||
profile_path: str,
|
||
allowed_domains: list[str],
|
||
sensitive_data: dict | None = None,
|
||
continuation: str | None = None,
|
||
max_steps: int | None = 40,
|
||
max_actions_per_step: int = 3,
|
||
keep_session: bool = False,
|
||
tool_profile: str = "legacy",
|
||
prospecting_limits: dict[str, int | None] | None = None,
|
||
prospecting_artifacts: dict | None = None,
|
||
budget_steps_used: int = 0,
|
||
on_evidence_checkpoint=None,
|
||
require_existing_page: bool = False,
|
||
reuse_managed_view: bool = False,
|
||
preflight_policy=None,
|
||
) -> dict:
|
||
"""跑一轮 Agent。返回 verdict: success | failed | paused。
|
||
|
||
paused 时浏览器会话保留在 _live 中，等待下一次带 continuation 的调用。
|
||
keep_session=True（多步业务还有下一步）时，成功结束也保留会话，由编排层在最后关闭。
|
||
"""
|
||
from browser_use import Agent, Browser
|
||
|
||
if tool_profile not in {"legacy", "prospecting", "outreach"}:
|
||
raise ValueError("未知的浏览器工具配置")
|
||
if not settings.reply_relation_preflight_enabled:
|
||
preflight_policy = None
|
||
agent_class = _prospecting_agent_class(Agent) if tool_profile in {"prospecting", "outreach"} or embedded_mode() else Agent
|
||
extra_exclude = ["click", "input", "send_keys"] if tool_profile == "outreach" else []
|
||
|
||
def action_tools(exclude=None):
|
||
if tool_profile == "legacy":
|
||
return build_tools(task_id, exclude=exclude)
|
||
tools = build_prospecting_tools(task_id, exclude=[*(exclude or []), *extra_exclude])
|
||
if tool_profile == "outreach":
|
||
from backend.prospecting.outreach import register_tools
|
||
register_tools(tools, task_id)
|
||
return tools
|
||
cfg = get_llm_config()
|
||
session = _live.get(task_id)
|
||
|
||
if session and (continuation or reuse_managed_view and session.embedded) and session.tool_profile == tool_profile:
|
||
session.allowed_domains = tuple(allowed_domains)
|
||
session.allow_initial_blank = not require_existing_page
|
||
session.preflight_policy = preflight_policy
|
||
# A preserved Agent can still hold the opt-in registry. Apply the off
|
||
# switch before its next model call without replacing the live view.
|
||
existing_tools = getattr(session.agent, "tools", None)
|
||
if tool_profile != "legacy" and not settings.reply_relation_preflight_enabled and existing_tools is not None:
|
||
actions = existing_tools.registry.registry.actions
|
||
if any(name in actions for name in ("inspect_reply_relations", "resolve_reply_target")):
|
||
for name in ("inspect_reply_relations", "resolve_reply_target"):
|
||
actions.pop(name, None)
|
||
session.agent._setup_action_models()
|
||
if tool_profile in {"prospecting", "outreach"}:
|
||
limits = prospecting_limits or {}
|
||
set_prospecting_context(task_id, max_notes=limits.get("max_notes"),
|
||
max_comments_per_note=limits.get("max_comments_per_note"))
|
||
refresh_prospecting_request(task_id, prospecting_artifacts)
|
||
session.agent.add_new_task(continuation or agent_task)
|
||
agent = session.agent
|
||
else:
|
||
if session:
|
||
await close_live_session(task_id)
|
||
set_task_context(task_id)
|
||
if tool_profile in {"prospecting", "outreach"}:
|
||
limits = prospecting_limits or {}
|
||
set_prospecting_context(
|
||
task_id,
|
||
max_notes=limits.get("max_notes"),
|
||
max_comments_per_note=limits.get("max_comments_per_note"),
|
||
artifacts=prospecting_artifacts,
|
||
)
|
||
steps_dir = task_doc_dir(task_id) / "steps"
|
||
steps_dir.mkdir(parents=True, exist_ok=True)
|
||
append_progress(
|
||
task_id,
|
||
"启动浏览器",
|
||
"running",
|
||
f"独立 Chrome | 模型 {cfg['provider']}/{cfg['model']} | 账号自动填写：{'是' if sensitive_data else '否'}",
|
||
)
|
||
llm = create_llm()
|
||
if embedded_mode():
|
||
# 向 Electron 申请一个 WebContentsView，经 CDP 聚焦到它；界面页主机名进白名单
|
||
view = await _ctrl(
|
||
"POST", "/views",
|
||
{"task_id": task_id, "partition": _partition_for(profile_path), "allowed_domains": allowed_domains or []},
|
||
)
|
||
target_id = str(view.get("target_id") or "")
|
||
browser = Browser(
|
||
cdp_url=CDP_URL,
|
||
# 必须显式 headless=False + no_viewport=True：后端是 Electron 拉起的子进程，browser-use 探测不到屏幕
|
||
# 就按 headless 处理，给视图套 Emulation.setDeviceMetricsOverride，Electron 的 WebContentsView 会因此
|
||
# 变成 visibilityState=hidden、不再出帧，CDP 截图每步挂 15s。window_size=None 则避免它去改 Electron 主窗口尺寸。
|
||
headless=False,
|
||
no_viewport=True,
|
||
viewport=None,
|
||
window_size=None,
|
||
screen=None,
|
||
keep_alive=True,
|
||
allowed_domains=([*allowed_domains, UI_HOST] if allowed_domains else None),
|
||
)
|
||
await browser.start()
|
||
agent = agent_class(
|
||
task=agent_task,
|
||
llm=llm,
|
||
browser=browser,
|
||
tools=action_tools(EMBEDDED_EXCLUDED_ACTIONS),
|
||
use_vision=use_vision_for_model(cfg["model"], cfg["provider"]),
|
||
max_actions_per_step=max(1, int(max_actions_per_step)),
|
||
step_timeout=120,
|
||
use_judge=False,
|
||
sensitive_data=sensitive_data or None,
|
||
register_new_step_callback=_make_step_callback(task_id, steps_dir),
|
||
)
|
||
session = LiveSession(agent=agent, browser=browser, embedded=True, target_id=target_id, tool_profile=tool_profile,
|
||
allowed_domains=tuple(allowed_domains), allow_initial_blank=not require_existing_page,
|
||
preflight_policy=preflight_policy)
|
||
_live[task_id] = session
|
||
append_progress(task_id, "浏览器就绪", "running", "内嵌浏览器（Electron WebContentsView）")
|
||
return await _run_agent(task_id, session, agent, max_steps, keep_session, budget_steps_used, on_evidence_checkpoint,
|
||
prospecting_goal_version=(prospecting_artifacts or {}).get("_knowledge_goal_version"))
|
||
window_kwargs = window_kwargs_for(_dock_bounds)
|
||
browser = Browser(
|
||
headless=False,
|
||
user_data_dir=profile_path,
|
||
allowed_domains=allowed_domains or None,
|
||
keep_alive=True,
|
||
**window_kwargs,
|
||
)
|
||
agent = agent_class(
|
||
task=agent_task,
|
||
llm=llm,
|
||
browser=browser,
|
||
tools=action_tools(),
|
||
use_vision=use_vision_for_model(cfg["model"], cfg["provider"]),
|
||
max_actions_per_step=max(1, int(max_actions_per_step)),
|
||
step_timeout=120,
|
||
use_judge=False,
|
||
sensitive_data=sensitive_data or None,
|
||
register_new_step_callback=_make_step_callback(task_id, steps_dir),
|
||
)
|
||
session = LiveSession(agent=agent, browser=browser, tool_profile=tool_profile, preflight_policy=preflight_policy)
|
||
_live[task_id] = session
|
||
|
||
return await _run_agent(task_id, session, agent, max_steps, keep_session, budget_steps_used, on_evidence_checkpoint,
|
||
prospecting_goal_version=(prospecting_artifacts or {}).get("_knowledge_goal_version"))
|
||
|
||
|
||
def read_browser_budget(task_id: str) -> dict:
|
||
"""Read the task-local attempt ledger; damaged records never reset to zero."""
|
||
path = task_doc_dir(task_id) / "browser-budget.json"
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
if path.stat().st_size > 8192:
|
||
raise ValueError("oversized ledger")
|
||
record = json.loads(path.read_text(encoding="utf-8"))
|
||
total, last = record["total_steps_used"], record["last_run_steps"]
|
||
if type(total) is not int or type(last) is not int or not 0 <= last <= total:
|
||
raise ValueError("invalid attempt counts")
|
||
return record
|
||
except (OSError, ValueError, TypeError, KeyError) as error:
|
||
raise RuntimeError("浏览器步骤台账无法核对，保留现有记录，请检查任务证据") from error
|
||
|
||
|
||
def _existing_evidence_step_floor(task_id: str) -> int:
|
||
return max((
|
||
int(match.group(1))
|
||
for path in (task_doc_dir(task_id) / "steps").glob("step_*.png")
|
||
if (match := re.fullmatch(r"step_(\d+)\.png", path.name))
|
||
), default=0)
|
||
|
||
|
||
async def _run_agent(
|
||
task_id: str, session: LiveSession, agent: Any, max_steps: int | None,
|
||
keep_session: bool = False, budget_steps_used: int = 0,
|
||
on_evidence_checkpoint=None,
|
||
prospecting_goal_version: int | None = None,
|
||
) -> dict:
|
||
if session.embedded:
|
||
try:
|
||
cdp_session = await _focus_managed_target(task_id, session, verify_host=True)
|
||
# Retain the existing embedded rendering setup after identity checks.
|
||
try:
|
||
await cdp_session.cdp_client.send.Emulation.clearDeviceMetricsOverride(session_id=cdp_session.session_id)
|
||
except Exception:
|
||
log.warning("清除设备指标模拟失败（不影响运行）")
|
||
except ManagedTargetError as error:
|
||
return _managed_target_result(task_id, error)
|
||
async def managed_guard():
|
||
try:
|
||
await _focus_managed_target(task_id, session)
|
||
except ManagedTargetError as error:
|
||
agent._managed_target_failure = error
|
||
agent.state.stopped = True
|
||
raise
|
||
agent._managed_target_guard = managed_guard
|
||
agent._managed_session = session
|
||
agent._managed_target_failure = None
|
||
# browser-use 0.13.10 uses an absolute, inclusive step ceiling. Both state
|
||
# n_steps (one-based) and history are cumulative across add_new_task/run.
|
||
# Translate our per-call allowance, then expose an attempt delta rather than
|
||
# counting the cumulative history again. State also counts timed-out steps,
|
||
# whereas history may omit them or append a non-step failure sentinel.
|
||
unlimited = max_steps is None
|
||
_evidence_failed_tasks.discard(task_id)
|
||
agent._goal_based = unlimited
|
||
allowance = None if unlimited else int(max_steps)
|
||
if allowance is not None and allowance < 1:
|
||
raise ValueError("本轮浏览器步骤额度必须大于零")
|
||
try:
|
||
start_step = max(1, int(agent.state.n_steps))
|
||
except (AttributeError, TypeError, ValueError):
|
||
start_step = None
|
||
previous = read_browser_budget(task_id)
|
||
total_before = max(
|
||
int(previous.get("total_steps_used", 0)),
|
||
max(0, int(budget_steps_used)),
|
||
max(0, (start_step or 1) - 1),
|
||
_existing_evidence_step_floor(task_id),
|
||
)
|
||
if allowance is not None:
|
||
allowance = min(allowance, max(0, MAX_TASK_BROWSER_STEPS - total_before))
|
||
run_id = uuid.uuid4().hex
|
||
if session.tool_profile in {"prospecting", "outreach"}:
|
||
set_prospecting_run_context(task_id, run_id, goal_version=prospecting_goal_version)
|
||
agent._preflight_scope_gate = None
|
||
if settings.reply_relation_preflight_enabled and session.preflight_policy and session.preflight_policy.active:
|
||
try:
|
||
agent._preflight_scope_gate = PreflightGate(task_id, session.browser, session.preflight_policy,
|
||
run_id, prospecting_goal_version or 1)
|
||
except PreflightScopeError as error:
|
||
return _preflight_scope_result(task_id, error)
|
||
attempts_started = 0
|
||
steps_this_run = 0
|
||
history_start = len(getattr(getattr(agent, "history", None), "history", []) or [])
|
||
program_stop = None
|
||
agent._program_failure_limit = None
|
||
evidence_failures: dict[tuple[str, int], int] = {}
|
||
|
||
def checkpoint(attempts: int) -> None:
|
||
# Write an absolute value anchored to this run's original total. The
|
||
# step-start checkpoint and the finally checkpoint are idempotent, not
|
||
# additive, so the same attempted step is never charged twice.
|
||
_write_json_atomic(task_doc_dir(task_id) / "browser-budget.json", {
|
||
"version": 2,
|
||
"execution_policy": "goal-v1" if unlimited else "legacy-bounded",
|
||
"total_steps_used": total_before + attempts,
|
||
"last_run_steps": attempts,
|
||
"last_run_id": run_id,
|
||
"max_total_steps": None if unlimited else MAX_TASK_BROWSER_STEPS,
|
||
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||
})
|
||
|
||
checkpoint(0)
|
||
if allowance is not None and allowance < 1:
|
||
raise RuntimeError(f"本次任务的 {MAX_TASK_BROWSER_STEPS} 步浏览器额度已用完")
|
||
absolute_limit = math.inf if unlimited else (start_step or 1) + allowance - 1
|
||
|
||
async def count_step_start(_agent: Any) -> None:
|
||
nonlocal attempts_started
|
||
if session.embedded:
|
||
await agent._managed_target_guard()
|
||
if agent._preflight_scope_gate:
|
||
try:
|
||
await agent._preflight_scope_gate.observe()
|
||
except PreflightScopeError:
|
||
agent.state.stopped = True
|
||
raise
|
||
if allowance is not None and attempts_started >= allowance:
|
||
raise RuntimeError("本轮浏览器步骤额度已用完")
|
||
attempts_started += 1
|
||
# Persist before model/browser work starts. Cancellation mid-step and
|
||
# even an abrupt process exit cannot make an attempted step free.
|
||
checkpoint(attempts_started)
|
||
|
||
async def evidence_checkpoint(_agent: Any) -> None:
|
||
nonlocal program_stop
|
||
if session.tool_profile == "prospecting":
|
||
from urllib.parse import urlsplit
|
||
items = getattr(getattr(_agent, "history", None), "history", []) or []
|
||
url = getattr(getattr(items[-1], "state", None), "url", "") if items else ""
|
||
page_key = urlsplit(url).path
|
||
errors = [str(getattr(row, "error", "") or "") for row in getattr(_agent.state, "last_result", []) or []]
|
||
source_error = any(("inspect_source_candidates" in error or "save_source" in error)
|
||
and any(word in error for word in ("容器", "原文", "定位")) for error in errors)
|
||
if source_error and re.fullmatch(r"/(?:explore|discovery/item)/[A-Za-z0-9]+", page_key):
|
||
key = (page_key, len(get_task_artifacts(task_id).get("sources") or []))
|
||
evidence_failures[key] = evidence_failures.get(key, 0) + 1
|
||
if evidence_failures[key] >= 3:
|
||
_evidence_failed_tasks.add(task_id)
|
||
_agent.state.stopped = True
|
||
return
|
||
if on_evidence_checkpoint is None:
|
||
return
|
||
decision = await on_evidence_checkpoint(get_task_artifacts(task_id))
|
||
if decision and decision.get("code") == "target_reached":
|
||
program_stop = decision
|
||
# The pinned run loop checks this flag before another model call.
|
||
# No extra attempt is started or charged to the task ledger.
|
||
_agent.state.stopped = True
|
||
|
||
try:
|
||
hooks = {"on_step_start": count_step_start}
|
||
if on_evidence_checkpoint is not None or session.tool_profile == "prospecting":
|
||
hooks["on_step_end"] = evidence_checkpoint
|
||
with diagnostics.run_scope(task_id, run_id) as run_diagnostics:
|
||
history = await agent.run(max_steps=absolute_limit, **hooks)
|
||
except ManagedTargetError as error:
|
||
return _managed_target_result(task_id, error, history={"steps_this_run": attempts_started,
|
||
"total_steps_used": total_before + attempts_started, "is_done": False, "is_successful": False,
|
||
"managed_target_error": error.code, "model_diagnostics": diagnostics.summary(run_diagnostics)})
|
||
except PreflightScopeError as error:
|
||
return _preflight_scope_result(task_id, error, history={"steps_this_run": attempts_started,
|
||
"total_steps_used": total_before + attempts_started, "is_done": False, "is_successful": False,
|
||
"model_diagnostics": diagnostics.summary(run_diagnostics)})
|
||
except Exception as e:
|
||
append_progress(task_id, "Browser 异常", "failed", str(e)[:500], level="error")
|
||
await close_live_session(task_id)
|
||
clear_task_context(task_id)
|
||
raise
|
||
finally:
|
||
try:
|
||
state_delta = max(0, int(agent.state.n_steps) - start_step) if start_step is not None else 0
|
||
except (AttributeError, TypeError, ValueError):
|
||
state_delta = 0
|
||
steps_this_run = max(attempts_started, state_delta)
|
||
checkpoint(steps_this_run)
|
||
|
||
artifacts = get_task_artifacts(task_id)
|
||
compact = _compact_history(history)
|
||
actual_history = getattr(history, "history", None)
|
||
if isinstance(actual_history, list):
|
||
compact["errors_this_run"] = [str(result.error) for step in actual_history[history_start:]
|
||
for result in getattr(step, "result", []) or [] if getattr(result, "error", None)][-20:]
|
||
compact["steps_this_run"] = steps_this_run
|
||
compact["total_steps_used"] = total_before + steps_this_run
|
||
compact["step_limit_this_run"] = allowance
|
||
compact["absolute_step_limit"] = None if unlimited else absolute_limit
|
||
compact["model_diagnostics"] = diagnostics.summary(run_diagnostics)
|
||
if agent._preflight_scope_gate and agent._preflight_scope_gate.failure:
|
||
return _preflight_scope_result(task_id, agent._preflight_scope_gate.failure, history=compact)
|
||
if isinstance(getattr(agent, "_managed_target_failure", None), ManagedTargetError):
|
||
compact["managed_target_error"] = agent._managed_target_failure.code
|
||
return _managed_target_result(task_id, agent._managed_target_failure, history=compact)
|
||
if getattr(agent, "_program_failure_limit", None):
|
||
compact["program_failure_limit"] = dict(agent._program_failure_limit)
|
||
stalled = task_id in _stalled_tasks
|
||
evidence_failed = task_id in _evidence_failed_tasks
|
||
_evidence_failed_tasks.discard(task_id)
|
||
|
||
blocker = pop_pending_blocker(task_id)
|
||
if blocker:
|
||
return {
|
||
"verdict": "paused",
|
||
"question": blocker["message"],
|
||
"blocker": blocker,
|
||
"artifacts": artifacts,
|
||
"history": compact,
|
||
"stop_reason": "platform_limited",
|
||
}
|
||
question = pop_pending_question(task_id)
|
||
if question:
|
||
return {"verdict": "paused", "question": question, "artifacts": artifacts, "history": compact, "stop_reason": "awaiting_user"}
|
||
|
||
verdict, summary = verdict_from_history(compact, max_steps)
|
||
stop_reason = "no_progress" if stalled else stop_reason_from_history(compact, max_steps)
|
||
if stalled:
|
||
verdict, summary = "failed", "相同页面与动作连续重复且没有新增来源，已保留结果；请调整路径后继续。"
|
||
if stop_reason == "model_schema_error":
|
||
verdict, summary = "failed", "模型动作连续未通过结构校验，框架已停止执行并保留阶段结果；失败动作未执行，请查看诊断和模型用量记录。"
|
||
elif stop_reason == "model_failure_limit":
|
||
verdict, summary = "failed", "模型或动作连续失败触发执行保护，已有结果保留；请根据具体诊断修复后继续。"
|
||
elif compact.get("program_failure_limit") and stop_reason in {"model_balance", "model_auth", "model_rate_limit"}:
|
||
detail = {"model_balance": "模型服务返回402，余额不足", "model_auth": "模型服务返回401，认证失败", "model_rate_limit": "模型服务返回429，请求受限"}[stop_reason]
|
||
verdict, summary = "failed", detail + "，触发连续失败保护；已有结果保留，请处理模型服务后继续。"
|
||
if evidence_failed:
|
||
verdict, stop_reason = "failed", "evidence_stalled"
|
||
summary = "同一笔记多次取证失败且没有新增来源，已停止重复尝试；已有结果保留，其他路线未视为完成。"
|
||
if program_stop:
|
||
compact["program_stop"] = program_stop
|
||
verdict, stop_reason = "success", "target_reached"
|
||
summary = program_stop.get("detail") or "已达到用户本次明确的候选人数，停止新增采集。"
|
||
|
||
if keep_session and (verdict == "success" or agent._preflight_scope_gate is not None):
|
||
return {"verdict": verdict, "summary": summary, "artifacts": artifacts, "history": compact, "stop_reason": stop_reason}
|
||
await close_live_session(task_id)
|
||
clear_task_context(task_id)
|
||
return {"verdict": verdict, "summary": summary, "artifacts": artifacts, "history": compact, "stop_reason": stop_reason}
|
||
|
||
|
||
async def debug_browser(url: str, task: str) -> dict:
|
||
from browser_use import Agent, Browser
|
||
|
||
cfg = get_llm_config()
|
||
browser = Browser(headless=False, keep_alive=False)
|
||
agent = Agent(
|
||
task=f"打开 {url}，然后：{task}",
|
||
llm=create_llm(),
|
||
browser=browser,
|
||
use_vision=use_vision_for_model(cfg["model"], cfg["provider"]),
|
||
max_actions_per_step=3,
|
||
use_judge=False,
|
||
)
|
||
history = await agent.run(max_steps=25)
|
||
return {"result": str(history.final_result()) if history else "done", "history": _compact_history(history)}
|