一、从一次失败的景点采集说起
在 AI 自动化任务中,记忆模块是实现复杂任务处理的关键组件。Browser-use 项目通过引入记忆模块,解决了 LLM 在连续性任务中的无状态性问题,使代理能够维持上下文连贯性,执行复杂多步骤任务,并从错误中学习和恢复。
上一篇(公众号首发)-揭秘AI自动化框架Browser-use(三):Browser-use控制浏览器的核心机制
设想一个场景:你正在开发一个自动采集四川著名景点信息的程序,目标是收集九寨沟、峨眉山等 10 个景点的基本信息。初始方案可能是这样的:
async def collect_landscape_info():# 访问九寨沟页面await llm.invoke("在页面上查找九寨沟的门票价格")# ✓ 找到了"220元"# 继续查找开放时间await llm.invoke("查找开放时间") # ❌ 失败:LLM不知道在查找谁的开放时间# 尝试获取交通信息await llm.invoke("查找如何到达") # ❌ 失败:LLM不知道要到达哪里
这个简单的尝试很快就失败了,原因在于 LLM 是"无状态"的,每次调用都是独立的,没有上下文的概念。
人类在完成类似任务时,会依赖记忆能力:
- 记住当前目标:"我正在查找九寨沟的信息"
- 追踪进度:"已经找到门票价格,还需要查找开放时间"
- 关联信息:"这个价格是旺季的,淡季应该更便宜"
- 错误恢复:"这个页面打不开,我换个网站试试"
LLM 的无状态性使其在连续性任务中表现不佳,每次调用都像是一个"失忆"的助手,需要重新解释任务。记忆模块的引入使代理能够维持上下文连贯性,执行复杂多步骤任务,并从错误中学习和恢复。
二、记忆模块的必要性
1维持上下文连贯性
LLM 的无状态性使其在连续性任务中表现不佳。记忆模块存储了整个交互历史,使模型能够理解当前状态与之前操作的关系,避免重复已执行的操作,并基于过去的结果调整策略。
处理复杂多步骤任务
浏览任务通常需要多个步骤才能完成,如搜索信息、导航到特定页面、填写表单等。没有记忆,代理将无法执行需要多个连续步骤的任务。
错误恢复和学习
记忆模块记录了错误和失败的尝试,使代理能够避免重复相同的错误,从失败中学习并尝试替代方法,在遇到问题时回溯到之前的状态。
令牌管理和优化
LLM 有输入令牌限制,记忆模块通过智能管理历史信息来裁剪不必要的历史,保持在令牌限制内,优先保留重要信息。
状态持久化和恢复
记忆模块允许保存代理状态到文件,在会话之间恢复状态,或在系统崩溃后恢复执行。
长期规划能力
通过记忆,代理能够制定多步骤计划,跟踪计划的执行进度,根据新信息调整计划。
多模态信息整合
Browser-use 需要整合多种信息类型,记忆模块将这些不同类型的信息组织成结构化的历史,使模型能够全面理解网页环境。
三、记忆模块的技术实现解析
1. 核心记忆组件
1.1 MessageManager
MessageManager
是记忆模块的核心组件,负责管理与 LLM 的对话历史。它维护与 LLM 的对话历史,添加浏览器状态信息到对话中,添加模型输出到对话中,并管理令牌限制,必要时裁剪历史。
#browser_use/agent/message_manager/service.py
class MessageManager:def __init__(self,task: str,system_message: str,settings: MessageManagerSettings,state: Optional[MessageManagerState] = None,):# 初始化消息历史self.task = taskself.system_message = system_messageself.settings = settingsself.state = state or MessageManagerState()self._messages: List[BaseMessage] = []self._init_messages()
1.2 MessageManagerSettings
消息管理的核心配置由 MessageManagerSettings
定义,决定了最大输入令牌数、图片令牌预算、需要包含的 HTML 属性列表、敏感数据处理方式等。
#browser_use/agent/message_manager/service.py
class MessageManagerSettings(BaseModel):max_input_tokens: int = 128000estimated_characters_per_token: int = 3image_tokens: int = 800include_attributes: list[str] = []message_context: Optional[str] = Nonesensitive_data: Optional[Dict[str, str]] = Noneavailable_file_paths: Optional[List[str]] = None
1.3 AgentState
AgentState
类存储代理的当前状态,包括代理 ID、步骤数、连续失败次数、历史记录和消息管理状态。
class AgentState:agent_id: str = Field(default_factory=lambda: str(uuid.uuid4()))n_steps: int = 0consecutive_failures: int = 0history: AgentHistoryList = Field(default_factory=AgentHistoryList)message_manager_state: MessageManagerState = Field(default_factory=MessageManagerState)
1.4 AgentHistoryList
AgentHistoryList
存储执行历史的列表,提供了多种辅助方法来分析执行过程,如获取 URL、截图、操作名称、提取的内容、错误、模型操作等。
class AgentHistoryList:history: List[AgentHistory] = Field(default_factory=list)# 提供多种辅助方法def urls(self) -> List[str]def screenshots(self) -> List[str]def action_names(self) -> List[str]def extracted_content(self) -> List[str]def errors(self) -> List[str]def model_actions(self) -> List[Dict[str, Any]]def final_result(self) -> Optional[str]def is_done(self) -> booldef has_errors(self) -> booldef model_thoughts(self) -> List[str]def action_results(self) -> List[Dict[str, Any]]
1.5 BrowserStateHistory
保存浏览器状态的历史记录,包括 URL、标题、标签页、交互元素和截图。
@dataclass
class BrowserStateHistory:url: strtitle: strtabs: list[TabInfo]interacted_element: list[DOMHistoryElement | None] | list[None]screenshot: Optional[str] = None
2. 记忆管理流程
2.1 初始化记忆
在 Agent
类的初始化方法中,创建 MessageManager
并初始化系统提示和任务。
self._message_manager = MessageManager(task=task,system_message=SystemPrompt(action_description=self.available_actions,max_actions_per_step=self.settings.max_actions_per_step,override_system_message=override_system_message,extend_system_message=extend_system_message,).get_system_message(),settings=MessageManagerSettings(max_input_tokens=self.settings.max_input_tokens,include_attributes=self.settings.include_attributes,message_context=self.settings.message_context,sensitive_data=sensitive_data,available_file_paths=self.settings.available_file_paths,),state=self.state.message_manager_state,
)
2.2 记录浏览器状态
每个步骤中,通过 add_state_message
方法将当前浏览器状态添加到记忆中。
def add_state_message(self,state: BrowserState,result: Optional[List[ActionResult]] = None,step_info: Optional[AgentStepInfo] = None,use_vision=True,
) -> None:# 如果需要保留在记忆中,直接添加到历史if result:for r in result:if r.include_in_memory:if r.extracted_content:msg = HumanMessage(content='Action result: ' + str(r.extracted_content))self._add_message_with_tokens(msg)if r.error:# 获取错误的最后一行last_line = r.error.split('\n')[-1]msg = HumanMessage(content='Action error: ' + last_line)self._add_message_with_tokens(msg)result = None # 如果结果已添加到历史,不再重复添加# 创建状态消息state_message = AgentMessagePrompt(state,result,include_attributes=self.settings.include_attributes,step_info=step_info,).get_user_message(use_vision)self._add_message_with_tokens(state_message)
2.3 记录模型输出
模型的输出通过 add_model_output
方法添加到记忆中。
def add_model_output(self, model_output: AgentOutput) -> None:tool_calls = [{'name': 'AgentOutput','args': model_output.model_dump(mode='json', exclude_unset=True),'id': str(self.state.tool_id),'type': 'tool_call',}]msg = AIMessage(content='',tool_calls=tool_calls,)self._add_message_with_tokens(msg)# 空工具响应self.add_tool_message(content='')
2.4 创建历史记录项
在每个步骤结束时,通过 _make_history_item
方法创建历史记录。
def _make_history_item(self,model_output: AgentOutput | None,state: BrowserState,result: list[ActionResult],metadata: Optional[StepMetadata] = None,
) -> None:if model_output:interacted_elements = AgentHistory.get_interacted_element(model_output, state.selector_map)else:interacted_elements = [None]state_history = BrowserStateHistory(url=state.url,title=state.title,tabs=state.tabs,interacted_element=interacted_elements,screenshot=state.screenshot,)history_item = AgentHistory(model_output=model_output, result=result, state=state_history, metadata=metadata)self.state.history.history.append(history_item)
2.5 令牌管理
当达到令牌限制时,MessageManager
会裁剪历史,优先保留系统消息,移除最旧的非系统消息,必要时移除最后的状态消息。
if 'Max token limit reached' in error_msg:# 从历史中裁剪令牌self._message_manager.settings.max_input_tokens = self.settings.max_input_tokens - 500logger.info(f'Cutting tokens from history - new max input tokens: {self._message_manager.settings.max_input_tokens}')self._message_manager.cut_messages()
3. 记忆持久化
3.1 保存对话历史
可以通过 save_conversation_path
参数指定保存对话历史的路径。
if self.settings.save_conversation_path:logger.info(f'Saving conversation to {self.settings.save_conversation_path}')
3.2 生成 GIF 记录
可以通过 generate_gif
参数生成任务执行的 GIF 记录。
if self.settings.generate_gif:output_path: str = 'agent_history.gif'if isinstance(self.settings.generate_gif, str):output_path = self.settings.generate_gifcreate_history_gif(task=self.task, history=self.state.history, output_path=output_path)
4. 记忆的使用
4.1 规划器使用记忆
规划器使用完整的消息历史来分析状态并建议下一步操作。
async def _run_planner(self) -> Optional[str]:# 创建规划器消息历史,使用完整的消息历史planner_messages = [PlannerPrompt(self.controller.registry.get_prompt_description()).get_system_message(),*self._message_manager.get_messages()[1:], # 使用完整的消息历史,除了第一条]# 获取规划器输出response = await self.settings.planner_llm.ainvoke(planner_messages)plan = str(response.content)
4.2 重放历史步骤
可以通过 _execute_history_step
方法重放历史步骤。
async def _execute_history_step(self, history_item: AgentHistory, delay: float) -> list[ActionResult]:state = await self.browser_context.get_state()if not state or not history_item.model_output:raise ValueError('Invalid state or model output')updated_actions = []for i, action in enumerate(history_item.model_output.action):updated_action = await self._update_action_indices(history_item.state.interacted_element[i],action,state,)updated_actions.append(updated_action)if updated_action is None:raise ValueError(f'Could not find matching element {i} in current page')result = await self.multi_act(updated_actions)await asyncio.sleep(delay)return result
5. 错误处理与恢复机制
Agent 类中内置了完整的错误处理机制,关键配置包括最大失败次数、重试延迟、输出验证等。
#browser_use/agent/service.pydef __init__(self,task: str,llm: BaseChatModel,# Optional parametersbrowser: Browser | None = None,browser_context: BrowserContext | None = None,controller: Controller[Context] = Controller(),# Initial agent run parameterssensitive_data: Optional[Dict[str, str]] = None,initial_actions: Optional[List[Dict[str, Dict[str, Any]]]] = None,# Cloud Callbacksregister_new_step_callback: Callable[['BrowserState', 'AgentOutput', int], Awaitable[None]] | None = None,register_done_callback: Callable[['AgentHistoryList'], Awaitable[None]] | None = None,register_external_agent_status_raise_error_callback: Callable[[], Awaitable[bool]] | None = None,# Agent settingsuse_vision: bool = True,use_vision_for_planner: bool = False,save_conversation_path: Optional[str] = None,save_conversation_path_encoding: Optional[str] = 'utf-8',max_failures: int = 3,retry_delay: int = 10,override_system_message: Optional[str] = None,extend_system_message: Optional[str] = None,max_input_tokens: int = 128000,validate_output: bool = False,message_context: Optional[str] = None,
总结
Browser-use 项目的记忆模块设计非常全面,主要特点包括分层记忆结构、智能令牌管理、丰富的历史分析方法、DOM 元素追踪、多种持久化选项等。这种设计使得代理能够有效地利用历史信息来指导未来的行动,同时提供了丰富的调试和分析功能。
想了解更多技术实现细节和源码解析,欢迎关注我的微信公众号【松哥ai自动化】。每周我都会带来一篇深度技术文章,从源码角度剖析各种实用工具的实现原理。
通过这些实践建议和最佳实践,开发者可以借鉴 Browser-use 的记忆模块,构建更可靠、更高效的自动化任务。
朋友们,有啥想要分析的,请在评论区发我,我会优先分析。下一篇我们将深入分析一个最近很火的ai自动化框架,你知道是谁吗?