08.BrowserUse架构分析

作者:

在

项目: browser-use/browser-use

定位: AI 浏览器自动化 Agent 框架,MIT 许可,86k+ Stars

核心理念: 让 AI 像人类一样操控浏览器——理解页面结构、规划操作序列、执行交互动作

架构总览

Browser-Use 采用三层分离架构:Agent 层(LLM 决策)→ Tools/Controller 层(动作注册与执行)→ BrowserSession 层(CDP 浏览器控制)。事件总线(EventBus)贯穿全栈,所有浏览器操作以事件驱动方式执行。以下是 10 个核心维度的深度分析。


1. Agent 核心循环(Step-Loop)

Agent 的执行是一个经典的感知-思考-行动循环。每个 step 由三个阶段组成:准备上下文、获取 LLM 输出、执行动作。

# agent/service.py - Agent.step()
@observe(name='agent.step', ignore_output=True, ignore_input=True)
@time_execution_async('--step')
async def step(self, step_info: AgentStepInfo | None = None) -> None:
    """Execute one step of the task"""
    self.step_start_time = time.time()
    browser_state_summary = None
    try:
        # Phase 0: CAPTCHA 等待
        if self.browser_session:
            captcha_wait = await self.browser_session.wait_if_captcha_solving()
            if captcha_wait and captcha_wait.waited:
                self.step_start_time = time.time()

        # Phase 1: 准备上下文(浏览器状态 + 消息构建)
        browser_state_summary = await self._prepare_context(step_info)
        self.state.last_model_output = None
        self.state.last_result = None

        # Phase 2: LLM 推理 + 动作执行
        await self._get_next_action(browser_state_summary)
        await self._execute_actions()

        # Phase 3: 后处理(下载追踪、循环检测、日志)
        await self._post_process()

    except Exception as e:
        await self._handle_step_error(e)
    finally:
        await self._finalize(browser_state_summary)

关键设计:每个 step 的超时由 step_timeout=180 控制,LLM 调用有独立的 llm_timeout(根据模型自动配置,如 Gemini 75s、Claude 90s)。CAPTCHA 解算时间不计入 step 耗时。


2. LLM 交互与结构化输出

Agent 使用 Pydantic 模型约束 LLM 输出格式,确保每次返回都是可解析的结构化动作。支持三种输出模式:标准模式(含 thinking)、无 thinking 模式、flash 模式。

# agent/service.py - _setup_action_models()
def _setup_action_models(self) -> None:
    """Setup dynamic action models from tools registry"""
    # 根据注册的 action 动态创建 ActionModel
    self.ActionModel = self.tools.registry.create_action_model()
    
    # 根据模式选择输出结构
    if self.settings.flash_mode:
        self.AgentOutput = AgentOutput.type_with_custom_actions_flash_mode(self.ActionModel)
    elif self.settings.use_thinking:
        self.AgentOutput = AgentOutput.type_with_custom_actions(self.ActionModel)
    else:
        self.AgentOutput = AgentOutput.type_with_custom_actions_no_thinking(self.ActionModel)

    # 用于 max_steps 强制结束时的精简输出
    self.DoneActionModel = self.tools.registry.create_action_model(include_actions=['done'])

AgentOutput 包含 current_state(thinking、evaluation_previous_goal、memory、next_goal)和 action(动作列表),LLM 每步最多输出 max_actions_per_step=5 个动作。结构化输出通过 output_model_schema 参数支持自定义 Pydantic 模型。


3. 动作注册系统(Registry)

Browser-Use 的动作系统采用装饰器注册 + 动态 Pydantic 模型生成模式。每个 action 通过 @registry.action() 注册,框架自动从函数签名生成参数模型。

# tools/registry/service.py - Registry.action()
def action(self, description: str, param_model: type[BaseModel] | None = None,
           domains: list[str] | None = None, terminates_sequence: bool = False):
    """Decorator for registering actions"""
    def decorator(func: Callable):
        if func.__name__ in self.exclude_actions:
            return func
        # 自动规范化函数签名,分离特殊参数和动作参数
        normalized_func, actual_param_model = self._normalize_action_function_signature(
            func, description, param_model
        )
        action = RegisteredAction(
            name=func.__name_,
            description=description,
            function=normalized_func,
            param_model=actual_param_model,
            domains=final_domains,
            terminates_sequence=terminates_sequence,
        )
        self.registry.actions[func.__name__] = action
        return normalized_func
    return decorator

特殊参数(如 browser_session、page_extraction_llm、file_system)由框架自动注入,action 函数只需声明即可使用。domains 参数支持按 URL 域名过滤可用动作,实现页面级动作隔离。


4. 工具服务层(Tools Service)

Tools 类是动作执行的中枢,注册了所有内置浏览器操作:导航、点击、输入、滚动、截图、提取等。每个动作返回 ActionResult。

# tools/service.py - Tools 类初始化中的内置动作注册
class Tools(Generic[Context]):
    def __init__(self, exclude_actions=None, output_model=None, display_files_in_done_text=True):
        self.registry = Registry[Context](exclude_actions or [])
        self._output_model = output_model
        self._coordinate_clicking_enabled = False
        
        self._register_done_action(output_model)
        
        # 基本导航动作
        @self.registry.action('', param_model=SearchAction, terminates_sequence=True)
        async def search(params: SearchAction, browser_session: BrowserSession):
            encoded_query = urllib.parse.quote_plus(params.query)
            search_engines = {
                'duckduckgo': f'https://duckduckgo.com/?q={encoded_query}',
                'google': f'https://www.google.com/search?q={encoded_query}&udm=14',
                'bing': f'https://www.bing.com/search?q={encoded_query}',
            }
            # ... 导航到搜索引擎

坐标点击功能需要模型支持(Claude Sonnet 4、Gemini 3 Pro 等),Tools 会自动检测模型能力:

# agent/service.py - 模型能力检测
model_name = getattr(llm, 'model', '').lower()
supports_coordinate_clicking = any(
    pattern in model_name
    for pattern in ['claude-sonnet-4', 'claude-opus-4', 'claude-fable-5', 'gemini-3-pro', 'browser-use/']
)
if supports_coordinate_clicking:
    self.tools.set_coordinate_clicking(True)

5. 浏览器会话管理(BrowserSession)

BrowserSession 是浏览器控制的核心抽象,基于 Pydantic BaseModel,支持本地浏览器和云端浏览器两种模式。采用事件驱动架构,所有 CDP 操作通过 EventBus 分发。

# browser/session.py - BrowserSession 核心结构
class BrowserSession(BaseModel):
    """Event-driven browser session with backwards compatibility.
    This class provides a 2-layer architecture:
    - High-level event handling for agents/tools
    - Direct CDP/Playwright calls for browser operations
    """
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        validate_assignment=True,
        extra='forbid',
        revalidate_instances='never',
    )
    
    # 支持云端和本地两种初始化方式
    def __init__(self, *, 
                 cdp_url: str | None = None,
                 browser_profile: BrowserProfile | None = None,
                 cloud_profile_id: UUID | str | None = None,
                 use_cloud: bool | None = None,
                 headless: bool | None = None,
                 executable_path: str | Path | None = None,
                 ...):

CDP 通信通过 CDPSession 管理,每个 target(页面/iframe/worker)有独立的 CDP 会话:

# browser/session.py - Target 与 CDPSession
class Target(BaseModel):
    """Browser target (page, iframe, worker)"""
    target_id: TargetID
    target_type: str  # 'page', 'iframe', 'worker'
    url: str = 'about:blank'
    title: str = 'Unknown title'

class CDPSession(BaseModel):
    """CDP communication channel to a target"""
    cdp_client: CDPClient
    target_id: TargetID
    session_id: SessionID

6. 消息管理与上下文压缩(MessageManager)

MessageManager 负责构建发送给 LLM 的消息序列,包括系统提示、浏览器状态、历史动作和结果。支持消息压缩机制防止上下文窗口溢出。

# agent/views.py - MessageCompactionSettings
class MessageCompactionSettings(BaseModel):
    """Summarizes older history into a compact memory block to reduce prompt size."""
    enabled: bool = True
    compact_every_n_steps: int = 25
    trigger_char_count: int | None = None  # ~10k tokens
    trigger_token_count: int | None = None
    chars_per_token: float = 4.0
    keep_last_items: int = 6
    summary_max_chars: int = 6000
    compaction_llm: BaseChatModel | None = None

上下文准备流程精心设计,注入多种辅助信息:

# agent/service.py - _prepare_context()
async def _prepare_context(self, step_info=None) -> BrowserStateSummary:
    browser_state_summary = await self.browser_session.get_browser_state_summary(
        include_screenshot=True,  # 始终截图,即使 use_vision=False
        include_recent_events=self.include_recent_events,
    )
    
    # 按页面 URL 过滤可用动作
    await self._update_action_models_for_page(browser_state_summary.url)
    page_filtered_actions = self.tools.registry.get_prompt_description(browser_state_summary.url)
    
    # 构建状态消息
    self._message_manager.create_state_messages(
        browser_state_summary=browser_state_summary,
        model_output=self.state.last_model_output,
        result=self.state.last_result,
        use_vision=self.settings.use_vision,
        page_filtered_actions=page_filtered_actions,
        sensitive_data=self.sensitive_data,
        plan_description=plan_description,
    )
    
    # 注入各种提示(预算警告、重规划、循环检测)
    await self._inject_budget_warning(step_info)
    self._inject_replan_nudge()
    self._inject_loop_detection_nudge()

7. 系统提示工程(SystemPrompt)

系统提示根据模型类型和运行模式动态选择模板,支持 8+ 种变体。模板以 Markdown 文件存储,通过 importlib.resources 加载。

# agent/prompts.py - SystemPrompt 模板选择
class SystemPrompt:
    def _load_prompt_template(self) -> None:
        if self.is_browser_use_model:
            if self.flash_mode:
                template_filename = 'system_prompt_browser_use_flash.md'
            elif self.use_thinking:
                template_filename = 'system_prompt_browser_use.md'
            else:
                template_filename = 'system_prompt_browser_use_no_thinking.md'
        elif self.is_anthropic_4_5 and self.flash_mode:
            template_filename = 'system_prompt_anthropic_flash.md'
        elif self.flash_mode and self.is_anthropic:
            template_filename = 'system_prompt_flash_anthropic.md'
        elif self.flash_mode:
            template_filename = 'system_prompt_flash.md'
        elif self.use_thinking:
            template_filename = 'system_prompt.md'
        else:
            template_filename = 'system_prompt_no_thinking.md'

浏览器状态描述包含丰富的页面统计信息:

# agent/prompts.py - 页面统计提取
def _extract_page_statistics(self) -> dict[str, int]:
    stats = {
        'links': 0, 'iframes': 0,
        'shadow_open': 0, 'shadow_closed': 0,
        'scroll_containers': 0, 'images': 0,
        'interactive_elements': 0, 'total_elements': 0,
        'text_chars': 0,
    }
    # 遍历 SimplifiedNode 树统计各类元素
    def traverse_node(node: SimplifiedNode) -> None:
        original = node.original_node
        if original.node_type == NodeType.ELEMENT_NODE:
            tag = original.tag_name.lower()
            if tag == 'a': stats['links'] += 1
            elif tag in ('iframe', 'frame'): stats['iframes'] += 1
            if node.is_interactive: stats['interactive_elements'] += 1
            if node.is_shadow_host: stats['shadow_open' if ... else 'shadow_closed'] += 1

8. 循环检测与规划系统

Browser-Use 内置了行为循环检测器(ActionLoopDetector)和规划系统,防止 Agent 陷入无效重复。

# agent/views.py - ActionLoopDetector
class ActionLoopDetector(BaseModel):
    """Tracks action repetition and page stagnation to detect behavioral loops."""
    window_size: int = 20
    recent_action_hashes: list[str] = Field(default_factory=list)
    recent_page_fingerprints: list[PageFingerprint] = Field(default_factory=list)
    max_repetition_count: int = 0
    consecutive_stagnant_pages: int = 0

    def get_nudge_message(self) -> str | None:
        """Return an escalating awareness nudge based on repetition severity."""
        if self.max_repetition_count >= 12:
            return f'Heads up: you have repeated a similar action {self.max_repetition_count} times...'
        elif self.max_repetition_count >= 8:
            return f'Heads up: you have repeated a similar action {self.max_repetition_count} times...'
        elif self.max_repetition_count >= 5:
            return f'Heads up: you have repeated a similar action {self.max_repetition_count} times...'
        
        if self.consecutive_stagnant_pages >= 5:
            return 'The page content has not changed across 5 consecutive actions...'
        return None

动作哈希归一化确保相似操作被正确识别为重复:

# agent/views.py - 动作归一化
def _normalize_action_for_hash(action_name: str, params: dict[str, Any]) -> str:
    if action_name == 'search':
        tokens = sorted(set(re.sub(r'[^ws]', ' ', query.lower()).split()))
        return f'search|{engine}|{"|".join(tokens)}'
    if action_name in ('click', 'input'):
        return f'click|{params.get("index")}'
    if action_name == 'navigate':
        return f'navigate|{params.get("url", "")}'

规划系统支持 enable_planning=True,在连续失败时触发重规划(planning_replan_on_stall=3),在长时间无计划时提示制定计划(planning_exploration_limit=5)。


9. 安全机制

Browser-Use 在多个层面实现了安全防护,特别是敏感数据处理和域名隔离。

# agent/service.py - 敏感数据安全检查
if self.sensitive_data:
    has_domain_specific_credentials = any(isinstance(v, dict) for v in self.sensitive_data.values())
    
    if not self.browser_profile.allowed_domains:
        self.logger.warning(
            '⚠️ Agent(sensitive_data=••••••••) was provided but Browser(allowed_domains=[...]) is not locked down! ⚠️n'
            '          ☠️ If the agent visits a malicious website and encounters a prompt-injection attack, '
            'your sensitive_data may be exposed!'
        )
    elif has_domain_specific_credentials:
        # 验证域名模式是否被 allowed_domains 覆盖
        for domain_pattern in domain_patterns:
            for allowed_domain in self.browser_profile.allowed_domains:
                if pattern_domain.endswith('.' + allowed_domain_part[2:]):
                    is_allowed = True
                    break

动作超时保护防止 CDP 操作挂起:

# tools/service.py - 全局动作超时
_ACTION_TIMEOUT_FALLBACK_S = 180.0

def _parse_env_action_timeout(raw: str | None) -> float:
    """Parse BROWSER_USE_ACTION_TIMEOUT_S defensively.
    Accepts only finite positive values."""
    parsed = float(raw)
    if not math.isfinite(parsed) or parsed <= 0:
        return _ACTION_TIMEOUT_FALLBACK_S
    return parsed

敏感数据替换支持 TOTP 2FA 验证码自动注入:

# tools/registry/service.py - 2FA 自动注入
if placeholder_name.endswith('bu_2fa_code'):
    totp = pyotp.TOTP(applicable_secrets[placeholder_name], digits=6)
    replacement_value = totp.now()

10. 事件驱动架构与弹性设计

Browser-Use 的事件系统基于 bubus 库的 EventBus,所有浏览器操作(导航、点击、标签切换)都以事件形式分发和处理。

# browser/session.py - ResilientEventBus
class ResilientEventBus(EventBus):
    """EventBus whose step()/wait_until_idle() no-op on a torn-down bus instead of asserting."""
    async def step(self, event=None, timeout=None, wait_for_timeout=0.1):
        if self._on_idle is None or self.event_queue is None:
            return None
        return await super().step(event, timeout, wait_for_timeout)

    async def wait_until_idle(self, timeout=None):
        if self._on_idle is None or self.event_queue is None:
            return None
        return await super().wait_until_idle(timeout)

新标签页自动检测与切换:

# tools/service.py - 自动检测新标签
async def _detect_new_tab_opened(browser_session, tabs_before) -> str:
    await asyncio.sleep(0.05)  # 等待 CDP Target.attachedToTarget 事件传播
    tabs_after = await browser_session.get_tabs()
    new_tabs = [t for t in tabs_after if t.target_id not in tabs_before]
    if new_tabs:
        new_tab = new_tabs[0]
        # 自动切换到新标签
        switch_event = browser_session.event_bus.dispatch(
            SwitchTabEvent(target_id=new_tab.target_id)
        )
        await switch_event
        return f'. Automatically switched to new tab (tab_id: {new_tab.target_id[-4:]}).'

Agent 支持暂停/恢复、外部中断、follow-up 任务续接等弹性能力:

# agent/service.py - 暂停/停止检查
async def _check_stop_or_pause(self) -> None:
    if self.register_should_stop_callback:
        if await self.register_should_stop_callback():
            self.state.stopped = True
            raise InterruptedError
    if self.state.stopped:
        raise InterruptedError
    if self.state.paused:
        raise InterruptedError

架构图

┌─────────────────────────────────────────────────────────┐
│                      User Task                          │
│                  "Find stars of repo"                   │
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│                    Agent (service.py)                    │
│  ┌─────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│  │ SystemPrompt│ │MessageManager│ │ ActionLoopDetector│ │
│  │  (prompts)  │ │  (messages)  │ │   (views.py)     │ │
│  └─────────────┘ └──────────────┘ └──────────────────┘ │
│  ┌─────────────────────────────────────────────────────┐│
│  │         step() → _prepare_context()                 ││
│  │              → _get_next_action() (LLM)             ││
│  │              → _execute_actions() → multi_act()     ││
│  │              → _post_process()                      ││
│  └─────────────────────────────────────────────────────┘│
└──────────────────────┬──────────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────────┐
│              Tools / Registry (tools/)                   │
│  ┌──────────────┐  ┌─────────────────────────────────┐  │
│  │   Registry   │  │  Built-in Actions:              │  │
│  │  @action()   │  │  search, click, input, scroll,  │  │
│  │  decorator   │  │  navigate, extract, done, ...   │  │
│  └──────────────┘  └─────────────────────────────────┘  │
│  ┌──────────────────────────────────────────────────────┐│
│  │  Domain filtering │ Sensitive data replacement       ││
│  │  TOTP 2FA injection │ Coordinate conversion          ││
│  └──────────────────────────────────────────────────────┘│
└──────────────────────┬──────────────────────────────────┘
                       │ EventBus dispatch
┌──────────────────────▼──────────────────────────────────┐
│           BrowserSession (browser/session.py)            │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
│  │ SessionManager│  │ CDPClient    │  │ BrowserProfile│ │
│  │ (targets)     │  │ (cdp_use)    │  │ (config)      │ │
│  └──────────────┘  └──────────────┘  └───────────────┘ │
│  ┌──────────────────────────────────────────────────────┐│
│  │  Local Browser ←→ Cloud Browser (use_cloud=True)     ││
│  │  DOM Watchdog │ Captcha Watchdog │ Demo Mode         ││
│  └──────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────┘

关键设计决策总结

维度 设计选择 权衡

|——|———|——|

输出约束 Pydantic 动态模型 + Union 类型 类型安全 vs 灵活性
浏览器控制 CDP 直连(非 Playwright) 性能 vs 易用性
事件系统 EventBus 解耦 可扩展性 vs 复杂度
循环检测 软检测(nudge 不 block) 安全性 vs 自主性
消息压缩 LLM 驱动摘要 上下文保真 vs 窗口效率
安全模型 allowed_domains + 敏感数据隔离 安全性 vs 易用性
多模型适配 模型检测 + 自动配置 通用性 vs 优化空间
云端支持 BrowserSession 双模(local/cloud) 统一 API vs 底层差异

Browser-Use 的架构精髓在于:将浏览器自动化从 DOM 选择器层面提升到语义理解层面。Agent 不依赖固定的 CSS 选择器,而是通过 LLM 理解页面语义,动态规划操作序列。这使得它能处理 CAPTCHA、动态页面、登录流程等传统自动化工具难以应对的场景。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注