拆解 browser-use 项目——深入理解 Agent 层

这是我们拆解 browser-use 项目的最后一篇文章了，我们在本文中将之前我们分析的功能点整合到一起构成完整的 Agent 项目。

核心架构概览

在深入细节之前，我们先来鸟瞰一下整个系统的架构。browser-use 主要由以下几个关键部分组成：

1. Agent (智能体核心)：作为系统的“大脑”和“指挥官”，负责接收任务、协调其他组件、执行决策循环（观察-思考-行动）。
2. Browser / BrowserContext (浏览器交互层)：提供与浏览器（如 Playwright）进行交互的接口，负责执行动作（点击、输入等）并获取浏览器当前状态（URL、DOM 结构、截图等）。
3. Controller (动作控制器)：管理和执行具体的浏览器动作（如 click_element, input_text, go_to_url 等）。它知道有哪些动作可用，以及如何执行它们。
4. MessageManager (消息管理器)：负责构建和管理与 LLM 的对话历史。这是至关重要的，因为它需要将复杂的浏览器状态转换为 LLM 能理解的格式，并处理上下文长度限制。
5. LLM (大型语言模型)：提供智能决策能力。Agent 将当前状态和任务信息通过 MessageManager 发送给 LLM，LLM 则返回下一步的思考和行动指令。
6. AgentHistory (历史记录器)：记录智能体执行过程中的每一步状态、思考、行动和结果，用于调试、分析和回放。
7. Prompts (提示工程)：精心设计的提示词，用于指导 LLM 的行为，确保其理解任务、遵循规则并以正确的格式输出结果。
8. 视图 (views.py) 和工具 (utils.py)：定义数据结构（如 AgentOutput, ActionResult, BrowserState）和提供辅助功能（如 JSON 解析、消息转换）。

整个系统的工作流程可以概括为：Agent 接收任务 -> 通过 BrowserContext 获取当前浏览器状态 -> MessageManager 将状态和历史格式化为提示 -> Agent 调用 LLM 获取思考和行动指令 -> Agent 指挥 Controller 执行 Browser 动作 -> 记录 AgentHistory -> 重复此循环，直到任务完成或达到限制。

指挥官：Agent 服务 (agent/service.py)

Agent 类是整个系统的核心，它像一个项目经理，负责统筹全局。

1. 初始化 (__init__)：搭建舞台

创建 Agent 实例时，需要提供核心组件（任务描述 task、语言模型 llm）以及各种配置。

# Agent 初始化（简化示意）
class Agent(Generic[Context]):def __init__(self,task: str,llm: BaseChatModel,browser: Browser | None = None,browser_context: BrowserContext | None = None,controller: Controller[Context] = Controller(),settings: AgentSettings = AgentSettings(), # 包含各种配置sensitive_data: Optional[Dict[str, str]] = None,initial_actions: Optional[List[Dict[str, Dict[str, Any]]]] = None,# ... 其他参数如回调函数、状态注入等):self.task = taskself.llm = llmself.controller = controllerself.settings = settingsself.state = AgentState() # 智能体运行时状态# 设置动态动作模型 (基于 Controller 注册的动作)self._setup_action_models()# 设置模型名称、工具调用方式等self._set_model_names()self.tool_calling_method = self._set_tool_calling_method()# 初始化消息管理器self._message_manager = MessageManager(...)# 设置浏览器实例和上下文self.browser, self.browser_context = self._setup_browser(browser, browser_context)# ... 初始化回调、上下文、遥测等

值得注意的是 _setup_action_models 方法。它会根据 Controller 中注册的可用动作（如 click_element, input_text, done 等）动态地创建 Pydantic 模型 (ActionModel 和 AgentOutput)。这意味着智能体的能力（能执行哪些动作）是可扩展的，并且 LLM 的输出会被严格校验。

2. 核心循环 (step)：观察-思考-行动

step 方法是智能体执行任务的基本单元，体现了经典的“观察-思考-行动”循环。

# Agent step 方法（简化示意）
async def step(self, step_info: Optional[AgentStepInfo] = None) -> None:try:# 1. 观察 (Observe): 获取当前浏览器状态state = await self.browser_context.get_state()# 检查是否暂停或停止await self._raise_if_stopped_or_paused()# 2. 思考准备 (Prepare Thought): 将状态添加到消息历史self._message_manager.add_state_message(state, self.state.last_result, step_info, self.settings.use_vision)# (可选) 运行规划器 (Planner)if self.settings.planner_llm and ...:plan = await self._run_planner()self._message_manager.add_plan(plan, position=-1) # 插入规划建议# 3. 思考 (Think): 调用 LLM 获取下一步行动input_messages = self._message_manager.get_messages()model_output: AgentOutput = await self.get_next_action(input_messages)# 移除临时的状态消息，避免历史记录过长self._message_manager._remove_last_state_message()# 将模型的输出（思考和行动）加入历史self._message_manager.add_model_output(model_output)# 4. 行动 (Act): 执行 LLM 返回的动作序列result: list[ActionResult] = await self.multi_act(model_output.action)self.state.last_result = result# 处理成功/失败状态self.state.consecutive_failures = 0except Exception as e:# 错误处理result = await self._handle_step_error(e)self.state.last_result = resultfinally:# 5. 记录 (Record): 将这一步的完整信息存入历史if state:metadata = StepMetadata(...)self._make_history_item(model_output, state, result, metadata)

• 观察 (get_state): 获取当前页面的 URL、标题、标签页、DOM 结构（特别是可交互元素）以及可选的页面截图。
• 思考准备 (add_state_message): MessageManager 将观察到的信息，连同上一步的结果/错误，格式化成 LLM 能理解的消息（详见 MessageManager 部分）。
• 规划器 (_run_planner): 一个可选的高级功能。如果配置了 planner_llm，智能体会定期调用一个“规划师”LLM，对当前整体进展进行分析，并给出更高层次的策略建议。这就像在执行具体步骤时，定期向导师请教大方向。
• 思考 (get_next_action): 将构建好的消息列表发送给 LLM。LLM 返回一个结构化的 AgentOutput 对象，包含它的“思考”（对上一步的评估 evaluation_previous_goal、当前的记忆 memory、下一步的目标 next_goal）和“行动”(action 列表)。
• 行动 (multi_act): Agent 指挥 Controller 按照 AgentOutput 中的 action 列表顺序执行浏览器动作。
• 记录 (_make_history_item): 将这一步的所有信息（LLM 的思考、选择的动作、动作执行结果、浏览器状态快照、耗时等）打包成 AgentHistory 对象，存入 AgentState 的 history 列表中。

3. 与 LLM 对话 (get_next_action)

这个方法负责调用 LLM。它会根据配置 (tool_calling_method) 选择不同的方式与 LLM 交互：

• function_calling / Tool Calling: 利用 OpenAI 等模型原生的工具调用能力。AgentOutput 被定义为一个“工具”，LLM 会直接返回调用该工具所需的参数。这是推荐的方式，因为模型经过专门训练，效果通常最好。
• json_mode: 强制 LLM 输出 JSON 格式的字符串，然后代码库再解析这个 JSON 字符串并验证其是否符合 AgentOutput 的结构。适用于支持 JSON 模式但不支持原生工具调用的模型。
• raw: LLM 直接输出包含 JSON 的文本（可能混杂其他思考过程），代码库需要从中提取 JSON 部分（如 extract_json_from_model_output）并解析。这是兼容性最广但也最容易出错的方式，需要 LLM 严格遵循输出格式。
• None (或 Langchain 的 with_structured_output): 使用 Langchain 提供的结构化输出功能，它会根据模型能力自动选择最佳策略（可能是工具调用、JSON 模式或内部提示调整）来获取符合 AgentOutput 结构的输出。

# get_next_action 简化逻辑
async def get_next_action(self, input_messages: list[BaseMessage]) -> AgentOutput:# (可选) 转换消息格式以适应特定模型input_messages = self._convert_input_messages(input_messages)if self.tool_calling_method == 'raw':output = self.llm.invoke(input_messages)# 提取并解析 JSONparsed_json = extract_json_from_model_output(str(output.content))parsed = self.AgentOutput(**parsed_json)elif self.tool_calling_method is None:# 使用 Langchain 的 with_structured_output (自动选择方法)structured_llm = self.llm.with_structured_output(self.AgentOutput, include_raw=True)response = await structured_llm.ainvoke(input_messages)parsed = response['parsed']else: # 'function_calling' or 'json_mode'# 使用 Langchain 的 with_structured_output (指定方法)structured_llm = self.llm.with_structured_output(self.AgentOutput, include_raw=True, method=self.tool_calling_method)response = await structured_llm.ainvoke(input_messages)parsed = response['parsed']# ... 处理解析结果，可能裁剪过多动作 ...log_response(parsed) # 记录 LLM 的思考和决策return parsed

4. 执行动作序列 (multi_act)

LLM 可能一次返回多个需要连续执行的动作（例如，填写用户名、填写密码、点击登录）。multi_act 负责按顺序执行这些动作。

一个重要的细节是：在执行每个动作（除了第一个）之前，它会检查页面上的元素是否发生了变化（通过比较元素的路径哈希 branch_path_hash）。如果页面出现了新的元素，意味着之前的动作可能触发了动态加载、弹窗或其他变化，此时再按旧的元素索引执行后续动作可能会出错。因此，它会中断执行序列，让 Agent 在下一个 step 中基于新的页面状态重新决策。

想象你告诉朋友：“去冰箱拿牛奶，然后关上冰箱门，再把牛奶递给我。” 如果朋友打开冰箱门时，里面突然跳出一只猫，他可能就不会继续执行“关门”和“递牛奶”了，而是会先处理这个意外情况（猫！）。multi_act 中的检查机制就类似这个过程，确保智能体在环境变化时能及时停下，重新评估。

# multi_act 简化逻辑
async def multi_act(self, actions: list[ActionModel], ...) -> list[ActionResult]:results = []# 获取当前页面元素哈希作为基准cached_selector_map = await self.browser_context.get_selector_map()cached_path_hashes = set(e.hash.branch_path_hash for e in cached_selector_map.values())for i, action in enumerate(actions):# 如果动作需要元素索引，并且不是第一个动作if action.get_index() is not None and i != 0:# 获取新状态并比较元素哈希new_state = await self.browser_context.get_state()new_path_hashes = set(e.hash.branch_path_hash for e in new_state.selector_map.values())# 如果出现新元素 (新哈希不在旧哈希集合中)，则中断if check_for_new_elements and not new_path_hashes.issubset(cached_path_hashes):msg = f'页面在动作 {i}/{len(actions)} 后发生变化，中断序列'logger.info(msg)results.append(ActionResult(extracted_content=msg, include_in_memory=True))break# 执行单个动作result = await self.controller.act(action, ...)results.append(result)# 如果动作完成任务、出错或已是最后一个动作，则结束if results[-1].is_done or results[-1].error or i == len(actions) - 1:break# 等待一小段时间await asyncio.sleep(self.browser_context.config.wait_between_actions)return results

5. 运行与控制 (run, pause, resume, stop)

run 方法是执行整个任务的入口点。它在一个循环中调用 step，直到任务完成（is_done 状态为 True）、达到最大步数限制或连续失败次数过多。它还处理初始动作执行、最终的清理工作（关闭浏览器）以及可选的 GIF 生成。pause, resume, stop 方法则提供了对智能体执行流程的外部控制。

信使：MessageManager (message_manager/service.py)

LLM 本身是无状态的，它们无法记住之前的对话。为了让 LLM 能够理解上下文并做出连贯的决策，我们需要将相关的对话历史和当前状态信息一起发送给它。MessageManager 就是负责管理这个“记忆”和“沟通”过程的关键组件。

1. 为何需要它？

• 上下文管理: 追踪对话历史，包括系统指令、用户任务、智能体的思考、执行的动作、动作结果以及浏览器状态。
• 状态格式化: 将复杂的浏览器状态（DOM 树、截图）转换为 LLM 能够理解的文本和图像格式。
• Token 限制: LLM 的输入有长度限制（Token Limit）。当对话历史变得过长时，需要有策略地进行删减，以确保最重要的信息得以保留。
• 敏感数据处理: 在将信息发送给 LLM 之前，自动过滤或替换掉配置中的敏感信息（如密码、API 密钥）。

2. 初始化 (_init_messages)：设定初始语境

MessageManager 在初始化时，会构建一个基础的对话开端，通常包括：

• 系统消息 (SystemMessage): 来自 system_prompt.md，定义了智能体的角色、能力、规则和输出格式。
• 上下文消息 (HumanMessage): (可选) 任务的额外背景信息。
• 任务消息 (HumanMessage): 清晰地陈述最终目标。
• 敏感数据提示 (HumanMessage): (可选) 告知 LLM 存在敏感数据占位符以及如何使用它们（例如 <secret>password</secret>）。
• 示例输出 (HumanMessage, AIMessage, ToolMessage): 给 LLM 一个清晰的、符合格式要求的输出示例，帮助它理解期望的 JSON 结构和动作调用方式。
• 历史占位符 (HumanMessage): 标记任务历史记忆的开始位置。
• 可用文件路径 (HumanMessage): (可选) 告知 LLM 可以访问哪些本地文件。

这个精心设计的开场白对于引导 LLM 正确理解任务和格式至关重要。

3. 构建状态消息 (add_state_message)：向 LLM 描述世界

这是 MessageManager 最核心的功能之一。当 Agent 获取到新的浏览器状态 (BrowserState) 时，此方法会将其转换为一条 HumanMessage。

# add_state_message 简化逻辑
def add_state_message(self, state: BrowserState, result: Optional[List[ActionResult]] = None, ...) -> None:# 处理上一步的结果/错误，如果需要直接加入记忆if result:for r in result:if r.include_in_memory:# 将需要记忆的结果/错误直接添加为单独的 HumanMessageif r.extracted_content:self._add_message_with_tokens(HumanMessage(content='Action result: ' + str(r.extracted_content)))if r.error:self._add_message_with_tokens(HumanMessage(content='Action error: ' + r.error.split('\n')[-1]))result = None # 标记已处理，避免重复添加# 使用 AgentMessagePrompt 格式化当前状态state_message_prompt = AgentMessagePrompt(state, result, ...)# 获取包含文本描述和可选截图的 HumanMessagestate_message = state_message_prompt.get_user_message(use_vision)# 将格式化后的状态消息添加到历史记录self._add_message_with_tokens(state_message)

AgentMessagePrompt (在 prompts.py 中定义) 负责具体的格式化工作。它会提取：

• 当前 URL
• 打开的标签页列表
• 视口内可交互元素的文本表示（例如 [1]<button>Login</button>, [2]<input type="text">Username</input>），并根据配置包含指定的 HTML 属性 (include_attributes)。
• 页面滚动信息（例如 ... 500 pixels above ..., ... 1200 pixels below ...）。
• 当前步骤信息和时间。
• (如果 use_vision=True 且存在截图) 截图的 Base64 编码，嵌入到消息中。

4. Token 管理：应对有限的记忆

LLM 的上下文窗口（能处理的最大 Token 数）是有限的。随着对话进行，消息历史会越来越长，最终可能超出限制。MessageManager 通过以下方式处理这个问题：

• Token 计算 (_count_tokens, _add_message_with_tokens): 在添加每条消息时，估算其 Token 数量（文本按字符数估算，图片按固定值 image_tokens 计算），并累加到 MessageHistory 的 current_tokens 中。
• 历史裁剪 (cut_messages): 当 current_tokens 超过 max_input_tokens 时，触发裁剪逻辑：
  1. 优先移除图片: 如果最后一条消息包含图片，先将其移除，因为图片通常占用大量 Token（例如 800 个）。
  2. 截断最后一条消息: 如果移除图片后仍然超限，则按比例截断最后一条（通常是状态消息）的文本内容，直到 Token 总数低于限制。

为了方便理解我打个比方，有一个非常健谈但记忆力有限的人。当你和他长时间交谈时，为了继续对话，他可能会先忘记你刚才给他看的照片（移除图片），如果还不行，他可能会忘记你刚刚说的长篇大论的最后一部分（截断文本）。cut_messages 的逻辑与此类似，旨在牺牲部分细节以维持对话。

5. 敏感数据过滤 (_filter_sensitive_data)

在将任何消息添加到历史记录之前，_add_message_with_tokens 会调用 _filter_sensitive_data。这个方法会遍历消息内容，查找 settings.sensitive_data 字典中定义的值（例如，用户密码明文），并将其替换为占位符（例如 <secret>user_password</secret>）。这样可以防止敏感信息被发送给 LLM 或记录在日志/历史文件中。

6. 数据结构 (message_manager/views.py)

• ManagedMessage: 将 Langchain 的 BaseMessage（如 HumanMessage, AIMessage）与 MessageMetadata（目前只有 tokens）封装在一起。它还实现了自定义的 Pydantic 序列化/反序列化逻辑，利用 Langchain 的 dumpd 和 load 来正确处理各种消息类型。
• MessageHistory: 存储 ManagedMessage 列表，并维护 current_tokens 总数。提供了添加、获取、移除消息的方法。
• MessageManagerState: 保存 MessageManager 的整体状态，主要是 MessageHistory 和下一个 ToolMessage 需要使用的 tool_id。

记忆与重现：AgentHistory (agent/views.py)

AgentHistoryList 和 AgentHistory 就像是智能体的飞行记录仪，忠实地记录了它执行任务过程中的每一个细节。

• AgentHistory: 代表智能体执行的一个单步记录。它包含：

  • model_output: LLM 在这一步的输出 (AgentOutput)，包括它的思考（current_state）和建议的动作 (action)。
  • result: 实际执行 action 后得到的结果 (list[ActionResult])。
  • state: 执行动作前的浏览器状态快照 (BrowserStateHistory，是 BrowserState 的简化版，只包含 URL、标题、标签页、交互元素和截图等关键信息，避免存储整个 DOM 树）。特别地，interacted_element 记录了该步骤中实际交互的元素信息（来自 state.selector_map），用于后续分析或回放。
  • metadata: 这一步的元数据 (StepMetadata)，如开始/结束时间、消耗的 Token 数。

• AgentHistoryList: 包含一个 AgentHistory 对象的列表，代表了智能体完整或部分的执行轨迹。它提供了许多实用的方法来分析这次运行：

  • total_duration_seconds(): 计算总耗时。
  • total_input_tokens(): 计算总 Token 消耗（基于 MessageManager 的估算）。
  • errors(): 获取每一步的错误信息。
  • final_result(): 获取最后一个 done 动作提取的内容。
  • is_done(), is_successful(): 判断任务是否完成以及是否成功。
  • urls(), screenshots(), action_names(), model_thoughts(), extracted_content(): 方便地提取运行过程中的各种信息。
  • save_to_file(), load_from_file(): 将历史记录序列化为 JSON 文件保存，或从文件加载。加载时会使用当前的 AgentOutput 模型来验证和丰富动作数据。

价值:

• 调试: 当智能体行为异常时，可以检查历史记录，了解它在每一步看到了什么（state），想了什么（model_output.current_state），做了什么（model_output.action），以及结果如何（result）。
• 分析: 可以统计 Token 消耗、耗时、错误率、常用动作等，用于评估和优化智能体性能。
• 回放与复现 (rerun_history): Agent 服务提供了 rerun_history 和 load_and_rerun 方法。可以加载保存的 AgentHistoryList，并尝试在当前的浏览器环境中重新执行其中的动作序列。这对于复现问题或在相似场景下重用动作流非常有用。回放时，_update_action_indices 方法会尝试根据历史记录中的元素特征在当前页面上找到对应的元素，即使其索引 (highlight_index) 可能已经改变，增加了回放的鲁棒性。
• 可视化 (gif.py): create_history_gif 函数可以读取 AgentHistoryList，提取其中的截图，并在截图上叠加任务目标、当前步骤、LLM 的下一步目标等信息，生成一个动态展示智能体执行过程的 GIF 动画。这对于演示和直观理解智能体行为非常有帮助。

指南针：提示工程 (prompts.py, system_prompt.md)

与 LLM 的有效沟通在很大程度上依赖于精心设计的提示（Prompts）。

• system_prompt.md: 这是提供给 LLM 的核心指令集。它定义了：

  • 角色: AI 代理，自动化浏览器任务。
  • 输入格式: 解释 LLM 将会收到的信息格式（任务、历史、URL、标签、元素列表等）。
  • 输出格式: 强制要求 LLM 必须以特定的 JSON 格式响应，包含 current_state (evaluation, memory, next_goal) 和 action 列表。这是确保代码能正确解析 LLM 输出的关键。
  • 动作规则: 如何使用动作、动作序列的限制（如 max_actions）、何时中断序列等。
  • 元素交互: 如何使用元素索引。
  • 导航与错误处理: 遇到问题（无合适元素、卡住、弹窗、验证码）时的建议策略。
  • 任务完成: 何时以及如何使用 done 动作，强调要在 done 动作中包含最终结果，以及处理重复性任务的计数要求。
  • 视觉上下文: 如何利用截图信息。
  • 表单填写: 提示注意填写时可能出现的动态变化。
  • 长任务: 强调在 memory 中跟踪状态和子结果。
  • 提取: 如何使用 extract_content。

  这个文件就像是智能体的“宪法”或“操作手册”，LLM 需要严格遵守。SystemPrompt 类负责加载和格式化这个 Markdown 文件。

• AgentMessagePrompt: 如前所述，它负责动态生成描述当前浏览器状态的 HumanMessage。它将 BrowserState 对象翻译成 LLM 易于理解的文本描述，并可选地附带截图。

• PlannerPrompt: 为可选的“规划师”LLM 设计的系统提示。它指导规划师分析历史和当前状态，评估进展，识别挑战，并提出高层次的后续步骤建议，同样要求以 JSON 格式输出。

良好的提示工程是确保智能体能够理解任务、遵循指令、与环境有效互动的基石。

工具箱：utils.py 及其他

• utils.py:

  • extract_json_from_model_output: 处理 LLM 可能以代码块形式返回 JSON 的情况。
  • convert_input_messages: 针对某些不支持标准工具调用或不允许连续相同角色消息的特定模型（如 deepseek-reasoner），转换消息格式（例如，将 ToolMessage 转为 HumanMessage，将 AIMessage 的工具调用转为内容字符串，合并连续的 HumanMessage 或 AIMessage）。这是确保与不同 LLM 兼容的重要环节。
  • save_conversation: 将某一步的完整输入消息和 LLM 响应保存到文本文件，方便调试提示和模型响应。

• views.py: 除了上面提到的 MessageManager 和 AgentHistory 相关视图外，还定义了：

  • AgentSettings: 智能体的配置选项。
  • AgentState: 智能体的运行时状态。
  • ActionResult: 单个动作的执行结果。
  • AgentOutput: LLM 的标准输出结构，使用 Pydantic 的 create_model 实现动态动作字段。
  • AgentError: 错误处理相关的常量和格式化函数。

总结

我们深入探讨了 browser-use 代码库所展示的一个浏览器 AI 智能体的内部架构。通过 Agent 的协调，MessageManager 的精心沟通，Controller 和 Browser 的精确执行，以及 AgentHistory 的详细记录，系统能够有效地将 LLM 的智能应用于复杂的浏览器自动化任务中。

这个架构为构建更强大、更通用的 AI 智能体奠定了坚实的基础。未来的发展可能包括更智能的规划能力、更强的错误恢复机制、对复杂交互（如拖拽、Canvas 操作）的支持、以及与其他工具和服务的集成。

希望这这个拆解系列文章详尽的分析能帮助你理解构建此类智能体的复杂性与精妙之处，并为你开发自己的 AI Agent 应用带来启发。好的，感谢您的耐心收看我们下个拆解系列再见。