LangChain + AgentRun 浏览器沙箱极简集成指南

AgentRun Browser Sandbox 介绍什么是 Browser Sandbox?Browser Sandbox 是 AgentRun 平台提供的云原生无头浏览器沙箱服务,基于阿里云函数计算FC构建。它为智能体提供了一个安全隔离的浏览器执行环境,支持通过标准的 Chrome DevTools Protocol (CDP) 远程控制浏览器实例。核心特性无头浏览器能力内置 Chromium/Chrome 浏览器,支持完整的 Web 标准原生兼容 Puppeteer、Playwright 等主流自动化框架支持通过 CDP 协议进行精细化控制实时可视化内置 VNC 服务,支持实时查看浏览器界面提供操作录制功能,方便调试和回放支持通过 noVNC 客户端在网页中直接观看安全与隔离每个沙箱实例运行在独立的容器环境中文件系统和进程空间完全隔离支持 WSS 加密传输,确保数据安全Serverless 架构按需创建,按量付费,无需提前预置资源快速弹性伸缩,支持高并发场景零运维,无需管理服务器和浏览器依赖主要应用场景AI Agent 赋能: 为大模型提供眼睛和手,执行网页浏览、信息提取、在线操作等任务自动化测试: 在云端运行端到端E2E测试和视觉回归测试数据采集: 稳定、高效地进行网页抓取,应对动态加载和反爬虫挑战内容生成: 自动化生成网页截图或 PDF 文档上手使用 Agentrun Browser SandboxAgentRun SDK 快速介绍后续的内容将基于 Agentrun SDK 进行因此我们先对 SDK 进行简要介绍AgentRun SDK 是一个开源的 Python 工具包,旨在简化智能体与 AgentRun 平台各种服务包括 Browser Sandbox的集成。它提供了统一的接口,让您可以用几行代码就将沙箱能力集成到现有的 Agent 框架中。SDK 的核心功能如下统一集成接口提供对 LangChain、AgentScope 等主流框架的开箱即用支持统一的模型代理接口,简化多模型管理标准化的工具注册机制Sandbox 生命周期管理自动创建和销毁沙箱实例支持会话级别的状态保持灵活的资源配置和超时控制安装 AgentRun SDKpip install agentrun-sdk[playwright,server]注意: 确保您的 Python 环境版本在 3.10 及以上。基本使用示例以下是使用 AgentRun SDK 创建和管理 Browser Sandbox 的核心代码from agentrun.sandbox import Sandbox, TemplateType from playwright.sync_api import sync_playwright # 创建 Browser Sandbox sandbox Sandbox.create( template_typeTemplateType.BROWSER, template_nameyour-template-name, sandbox_idle_timeout_seconds300 ) # 获取 CDP URL用于 Playwright 连接 cdp_url sandbox.get_cdp_url() # 使用 Playwright 连接并操作 with sync_playwright() as p: browser p.chromium.connect_over_cdp(cdp_url) page browser.contexts[0].pages[0] page.goto(https://www.example.com) page.screenshot(pathscreenshot.png) browser.close() # 销毁 Sandbox sandbox.delete()关键概念template_name: 控制台创建的浏览器环境模板cdp_url: 用于 Playwright/Puppeteer 连接vnc_url: 用于实时查看浏览器画面可通过sandbox.get_cdp_url()获取注意: 由于所有浏览器操作都在云端进行您无需在本地安装浏览器。Playwright 仅用于通过 CDP 协议连接到云端的浏览器实例。如何创建 sandbox 模板使用 Browser Sandbox 需要新建 Sandbox 模板您需要访问 Agentrun 控制台网站并按照如下步骤创建模板:在顶部菜单栏选择“运行时与沙箱”在左侧边栏选择“Sandbox沙箱”点击右上角“创建沙箱模板”选择“浏览器”在弹出的抽屉对话框中填写和选择您的模板的规格、网络等配置并复制模板名称6. 点击“创建浏览器” 等待其就绪即可。从零开始用 LangChain 创建 Browser Sandbox 智能体本教程将指导您从零开始创建一个完整的 Browser Sandbox 智能体项目。基于 LangChain 集成 Browser Sandbox本教程将详细讲解如何使用 LangChain 创建 Browser Sandbox 相关的 tools 并集成到 Agent 中。项目结构为了保持代码的内聚性和可维护性我们将代码拆分为以下模块模块职责划分sandbox_manager.py: 负责 Sandbox 的创建、管理和销毁提供统一的接口langchain_agent.py: 负责创建 LangChain tools 和 Agent集成 VNC 信息main.py: 作为入口文件演示如何使用上述模块步骤 1: 创建项目并安装依赖首先创建项目目录如果还没有mkdir -p langchain-demo cd langchain-demo创建 requirements.txt 文件内容如下# LangChain 核心库 langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.20 # AgentRun SDK agentrun-sdk[playwright,server]0.0.8 # 浏览器自动化 playwright1.40.0 # 环境变量管理 python-dotenv1.0.0然后安装依赖pip install -r requirements.txt主要依赖说明langchain 和 langchain-openai: LangChain 核心库agentrun-sdk[playwright,server]: AgentRun SDK用于 Sandbox 管理playwright: 浏览器自动化库 python-dotenv: 环境变量管理步骤 2: 配置环境变量在项目根目录创建 .env 文件配置以下环境变量# 阿里云百炼平台的 API Key用于调用大模型能力 # 请前往 https://bailian.console.aliyun.com/?tabapp#/api-key 创建和查看 DASHSCOPE_API_KEYsk-your-bailian-api-key # 阿里云账号的访问密钥 ID 和访问密钥 Secret用于 AgentRun SDK 鉴权 ALIBABA_CLOUD_ACCESS_KEY_IDyour-ak ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-sk ALIBABA_CLOUD_ACCOUNT_IDyour-main-account-id ALIBABA_CLOUD_REGIONcn-hangzhou # browser sandbox 模板的名称可以在 https://functionai.console.aliyun.com/cn-hangzhou/agent/runtime/sandbox 控制台创建 BROWSER_TEMPLATE_NAMEsandbox-your-template-name # agentrun 的控制面和数据面的 API 端点请求地址默认cn-hangzhou AGENTRUN_CONTROL_ENDPOINTagentrun.cn-hangzhou.aliyuncs.com AGENTRUN_DATA_ENDPOINThttps://${your-main-account-id}.agentrun-data.cn-hangzhou.aliyuncs.com步骤 3: 创建 Sandbox 生命周期管理模块创建 sandbox_manager.py 文件负责 Sandbox 的创建、管理和销毁。核心代码如下 Sandbox 生命周期管理模块 负责 AgentRun Browser Sandbox 的创建、管理和销毁。 提供统一的接口供 LangChain Agent 使用。 import os from typing import Optional, Dict, Any from dotenv import load_dotenv # 加载环境变量 load_dotenv() class SandboxManager: Sandbox 生命周期管理器 def __init__(self): self._sandbox: Optional[Any] None self._sandbox_id: Optional[str] None self._cdp_url: Optional[str] None self._vnc_url: Optional[str] None def create( self, template_name: Optional[str] None, idle_timeout: int 3000 ) - Dict[str, Any]: 创建或获取一个浏览器 sandbox 实例 Args: template_name: Sandbox 模板名称如果为 None 则从环境变量读取 idle_timeout: 空闲超时时间秒默认 3000 秒 Returns: dict: 包含 sandbox_id, cdp_url, vnc_url 的字典 Raises: RuntimeError: 创建失败时抛出异常 try: from agentrun.sandbox import Sandbox, TemplateType # 如果已有 sandbox直接返回 if self._sandbox is not None: return self.get_info() # 从环境变量获取模板名称 if template_name is None: template_name os.getenv( BROWSER_TEMPLATE_NAME, sandbox-browser-demo ) # 创建 sandbox self._sandbox Sandbox.create( template_typeTemplateType.BROWSER, template_nametemplate_name, sandbox_idle_timeout_secondsidle_timeout ) self._sandbox_id self._sandbox.sandbox_id self._cdp_url self._get_cdp_url() self._vnc_url self._get_vnc_url() return self.get_info() except ImportError as e: print(e) raise RuntimeError( agentrun-sdk 未安装请运行: pip install agentrun-sdk[playwright,server] ) except Exception as e: raise RuntimeError(f创建 Sandbox 失败: {str(e)}) def get_info(self) - Dict[str, Any]: 获取当前 sandbox 的信息 Returns: dict: 包含 sandbox_id, cdp_url, vnc_url 的字典 Raises: RuntimeError: 如果没有活动的 sandbox if self._sandbox is None: raise RuntimeError(没有活动的 sandbox请先创建) return { sandbox_id: self._sandbox_id, cdp_url: self._cdp_url, vnc_url: self._vnc_url, } def get_cdp_url(self) - Optional[str]: 获取 CDP URL return self._sandbox.get_cdp_url() def get_vnc_url(self) - Optional[str]: 获取 VNC URL return self._sandbox.get_vnc_url() def get_sandbox_id(self) - Optional[str]: 获取 Sandbox ID return self._sandbox_id def destroy(self) - str: 销毁当前的 sandbox 实例 Returns: str: 操作结果描述 if self._sandbox is None: return 没有活动的 sandbox try: sandbox_id self._sandbox_id # 尝试销毁 sandbox if hasattr(self._sandbox, delete): self._sandbox.delete() elif hasattr(self._sandbox, stop): self._sandbox.stop() elif hasattr(self._sandbox, destroy): self._sandbox.destroy() # 清理状态 self._sandbox None self._sandbox_id None self._cdp_url None self._vnc_url None return fSandbox 已销毁: {sandbox_id} except Exception as e: # 即使销毁失败也清理本地状态 self._sandbox None self._sandbox_id None self._cdp_url None self._vnc_url None return f销毁 Sandbox 时出错: {str(e)} def is_active(self) - bool: 检查 sandbox 是否活跃 return self._sandbox is not None def __enter__(self): 上下文管理器入口 return self def __exit__(self, exc_type, exc_val, exc_tb): 上下文管理器退出自动销毁 self.destroy() return False # 全局单例可选用于简单场景 _global_manager: Optional[SandboxManager] None def get_global_manager() - SandboxManager: 获取全局 SandboxManager 单例 global _global_manager if _global_manager is None: _global_manager SandboxManager() return _global_manager def reset_global_manager(): 重置全局 SandboxManager global _global_manager if _global_manager: _global_manager.destroy() _global_manager None关键功能创建 Sandbox: 使用 AgentRun SDK 创建浏览器 Sandbox获取连接信息: 自动获取 CDP URL 和 VNC URL支持多种属性名兼容生命周期管理: 提供销毁方法确保资源正确释放步骤 4: 创建 LangChain Tools 和 Agent创建langchain_agent.py文件定义 LangChain tools 并创建 Agent。核心代码如下 LangChain Agent 和 Tools 注册模块 负责创建 LangChain Agent注册 Sandbox 相关的 tools并集成 VNC 可视化。 本模块使用 sandbox_manager.py 中封装的 SandboxManager 来管理 sandbox 生命周期。 import os from dotenv import load_dotenv from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_agent from pydantic import BaseModel, Field # 导入 sandbox 管理器 from sandbox_manager import SandboxManager # 加载环境变量 load_dotenv() # 全局 sandbox 管理器实例单例模式 _sandbox_manager: SandboxManager | None None def get_sandbox_manager() - SandboxManager: 获取 sandbox 管理器实例单例模式 global _sandbox_manager if _sandbox_manager is None: _sandbox_manager SandboxManager() return _sandbox_manager # LangChain Tools 定义 tool def create_browser_sandbox( template_name: str None, idle_timeout: int 3000 ) - str: 创建或获取一个浏览器 sandbox 实例。 当需要访问网页、执行浏览器操作时首先需要创建 sandbox。 创建成功后会返回 sandbox 信息包括 VNC URL 用于可视化。 Args: template_name: Sandbox 模板名称如果不提供则从环境变量 BROWSER_TEMPLATE_NAME 读取 idle_timeout: 空闲超时时间秒默认 3000 秒 Returns: Sandbox 信息字符串包括 ID、CDP URL、VNC URL try: manager get_sandbox_manager() # 如果 template_name 为空字符串转换为 None 以便从环境变量读取 if template_name : template_name None info manager.create(template_nametemplate_name, idle_timeoutidle_timeout) result f✅ Sandbox 创建成功 Sandbox 信息: - ID: {info[sandbox_id]} - CDP URL: {info[cdp_url]} vnc_url info.get(vnc_url) if vnc_url: result f- VNC URL: {vnc_url}\n\n result 提示: VNC 查看器应该已自动打开您可以在浏览器中实时查看浏览器操作。 else: result \n警告: 未获取到 VNC URL可能无法使用可视化功能。 return result except Exception as e: return f 创建 Sandbox 失败: {str(e)} tool def get_sandbox_info() - str: 获取当前 sandbox 的详细信息包括 ID、CDP URL、VNC URL 等。 当需要查看当前 sandbox 状态或获取 VNC 连接信息时使用此工具。 Returns: Sandbox 信息字符串 try: manager get_sandbox_manager() info manager.get_info() result f 当前 Sandbox 信息: - Sandbox ID: {info[sandbox_id]} - CDP URL: {info[cdp_url]} if info.get(vnc_url): result f- VNC URL: {info[vnc_url]}\n\n result 您可以使用 VNC URL 在浏览器中实时查看操作过程。\n result 推荐使用 vnc.html 文件或 noVNC 客户端。 return result except RuntimeError as e: return f {str(e)} except Exception as e: return f 获取 Sandbox 信息失败: {str(e)} class NavigateInput(BaseModel): 浏览器导航输入参数 url: str Field(description要访问的网页 URL必须以 http:// 或 https:// 开头) wait_until: str Field( defaultload, description等待页面加载的状态: load, domcontentloaded, networkidle ) timeout: int Field( default30000, description超时时间毫秒默认 30000 ) tool(args_schemaNavigateInput) def navigate_to_url(url: str, wait_until: str load, timeout: int 30000) - str: 使用 sandbox 中的浏览器导航到指定 URL。 当用户需要访问网页时使用此工具。导航后可以在 VNC 中实时查看页面。 Args: url: 要访问的网页 URL wait_until: 等待页面加载的状态load/domcontentloaded/networkidle timeout: 超时时间毫秒 Returns: 导航结果描述 try: manager get_sandbox_manager() if not manager.is_active(): return 错误: 请先创建 sandbox # 验证 URL if not url.startswith((http://, https://)): return f 错误: 无效的 URL 格式: {url} cdp_url manager.get_cdp_url() if not cdp_url: return 错误: 无法获取 CDP URL # 使用 Playwright 连接浏览器并导航 try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser p.chromium.connect_over_cdp(cdp_url) pages browser.contexts[0].pages if browser.contexts else [] if pages: page pages[0] else: page browser.new_page() page.goto(url, wait_untilwait_until, timeouttimeout) title page.title() return f已成功导航到: {url}\n 页面标题: {title}\n 您可以在 VNC 中查看页面内容。 except ImportError: return f导航指令已发送: {url}\n 提示: 安装 playwright 以启用实际导航功能 (pip install playwright) except Exception as e: return f 导航失败: {str(e)} except Exception as e: return f 操作失败: {str(e)} tool(browser_screenshot, description在浏览器 sandbox 中截取当前页面截图) def take_screenshot(filename: str screenshot.png) - str: 截取浏览器当前页面的截图。 Args: filename: 截图文件名默认 screenshot.png Returns: 操作结果 try: manager get_sandbox_manager() if not manager.is_active(): return 错误: 请先创建 sandbox cdp_url manager.get_cdp_url() if not cdp_url: return 错误: 无法获取 CDP URL try: from playwright.sync_api import sync_playwright with sync_playwright() as p: browser p.chromium.connect_over_cdp(cdp_url) pages browser.contexts[0].pages if browser.contexts else [] if pages: page pages[0] else: return 错误: 没有打开的页面 page.screenshot(pathfilename) return f截图已保存: {filename} except ImportError: return 错误: 需要安装 playwright (pip install playwright) except Exception as e: return f 截图失败: {str(e)} except Exception as e: return f 操作失败: {str(e)} tool(destroy_sandbox, description销毁当前的 sandbox 实例释放资源。注意仅在程序退出或明确需要释放资源时使用不要在一轮对话后销毁。) def destroy_sandbox() - str: 销毁当前的 sandbox 实例。 重要提示此工具应该仅在以下情况使用 - 程序即将退出 - 明确需要释放资源 - 用户明确要求销毁 不要在一轮对话完成后就销毁 sandbox因为 sandbox 可以在多轮对话中复用。 Returns: 操作结果 try: manager get_sandbox_manager() result manager.destroy() return result except Exception as e: return f 销毁失败: {str(e)} # Agent 创建 def create_browser_agent(system_prompt: str None): 创建带有 sandbox 工具的 LangChain Agent Args: system_prompt: 自定义系统提示词如果为 None 则使用默认提示词 Returns: LangChain Agent 实例 # 配置 DashScope API api_key os.getenv(DASHSCOPE_API_KEY) if not api_key: raise ValueError(请设置环境变量 DASHSCOPE_API_KEY) base_url https://dashscope.aliyuncs.com/compatible-mode/v1 model_name os.getenv(QWEN_MODEL, qwen-plus) # 创建 LLM model ChatOpenAI( modelmodel_name, api_keyapi_key, base_urlbase_url, temperature0.7, ) # 创建工具列表 tools [ create_browser_sandbox, get_sandbox_info, navigate_to_url, take_screenshot, destroy_sandbox, ] # 默认系统提示词 if system_prompt is None: system_prompt 你是一个浏览器自动化助手可以使用 sandbox 来访问和操作网页。 当用户需要访问网页时请按以下步骤操作 1. 首先创建或获取 sandbox如果还没有 2. 使用 navigate_to_url 导航到目标网页 3. 执行用户请求的操作 4. 如果需要可以截取截图 重要提示 - 创建 sandbox 后会返回 VNC URL用户可以使用它实时查看浏览器操作 - 所有操作都会在 VNC 中实时显示方便调试和监控 - sandbox 可以在多轮对话中复用不要在一轮对话完成后就销毁 - 只有在用户明确要求销毁时才使用 destroy_sandbox 工具 - 不要主动建议用户销毁 sandbox除非用户明确要求 - 请始终用中文回复确保操作准确、高效。 # 创建 Agent agent create_agent( modelmodel, toolstools, system_promptsystem_prompt, ) return agent def get_available_tools(): 获取所有可用的工具列表 return [ create_browser_sandbox, get_sandbox_info, navigate_to_url, take_screenshot, destroy_sandbox, ]关键要点Tool 定义: 使用tool装饰器定义 LangChain tools类型提示: 所有参数必须有类型提示用于生成工具 schema文档字符串: 详细的文档字符串帮助 LLM 理解何时使用工具单例模式: 使用全局管理器实例确保 Sandbox 在会话中复用步骤 5: 创建主入口文件创建main.py文件作为程序入口。核心代码如下 LangChain AgentRun Browser Sandbox 集成示例 主入口文件演示如何使用 LangChain Agent 与 AgentRun Browser Sandbox 集成。 import os import sys import signal import webbrowser import urllib.parse import threading import http.server import socketserver from pathlib import Path from dotenv import load_dotenv from langchain_agent import create_browser_agent, get_sandbox_manager # 加载环境变量 load_dotenv() # 全局 HTTP 服务器实例 _http_server None _http_port 8080 # 全局清理标志用于防止重复清理 _cleanup_done False def start_http_server(): 启动一个简单的 HTTP 服务器来提供 vnc.html global _http_server if _http_server is not None: return _http_port try: current_dir Path(__file__).parent.absolute() class VNCRequestHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directorystr(current_dir), **kwargs) def log_message(self, format, *args): # 静默日志避免输出过多信息 pass # 尝试启动服务器 for port in range(_http_port, _http_port 10): try: server socketserver.TCPServer((, port), VNCRequestHandler) server.allow_reuse_address True # 在后台线程中运行服务器 def run_server(): server.serve_forever() thread threading.Thread(targetrun_server, daemonTrue) thread.start() _http_server server return port except OSError: continue return None except Exception as e: print(f启动 HTTP 服务器失败: {str(e)}) return None def open_vnc_viewer(vnc_url: str): 自动打开 VNC 查看器并设置 VNC URL Args: vnc_url: VNC WebSocket URL if not vnc_url: return try: # 获取当前文件所在目录 current_dir Path(__file__).parent.absolute() vnc_html_path current_dir / vnc.html # 检查文件是否存在 if not vnc_html_path.exists(): print(f警告: vnc.html 文件不存在: {vnc_html_path}) print_vnc_info(vnc_url) return # 启动 HTTP 服务器 port start_http_server() if port: # 编码 VNC URL 作为 URL 参数 encoded_url urllib.parse.quote(vnc_url, safe) # 构建 HTTP URL http_url fhttp://localhost:{port}/vnc.html?url{encoded_url} # 打开浏览器 print(f\n正在打开 VNC 查看器...) print(fHTTP 服务器运行在: http://localhost:{port}) print(fVNC URL: {vnc_url[:80]}...) print(f完整 URL: {http_url[:100]}...) webbrowser.open(http_url) print(fVNC 查看器已打开) print(fVNC URL 已通过 URL 参数自动设置页面加载后会自动连接) else: # 如果 HTTP 服务器启动失败尝试使用 file:// 协议 print(fHTTP 服务器启动失败尝试使用文件协议...) encoded_url urllib.parse.quote(vnc_url, safe) file_url ffile://{vnc_html_path}?url{encoded_url} webbrowser.open(file_url) print(fVNC 查看器已打开使用文件协议) print(f提示: 如果无法自动连接请手动复制 VNC URL 到输入框) except Exception as e: print(f自动打开 VNC 查看器失败: {str(e)}) print_vnc_info(vnc_url) def print_vnc_info(vnc_url: str): 打印 VNC 连接信息 if not vnc_url: return print(\n * 60) print(VNC 可视化连接信息) print( * 60) print(f\nVNC URL: {vnc_url}) print(\n使用方式:) print( 1. 使用 noVNC 客户端连接) print( 2. 或在浏览器中访问 VNC 查看器页面) print( 3. 实时查看浏览器操作过程) print(\n * 60 \n) def cleanup_sandbox(): 清理 sandbox 资源 这个函数可以被信号处理器、异常处理器和正常退出流程调用 global _cleanup_done # 防止重复清理 if _cleanup_done: return _cleanup_done True try: manager get_sandbox_manager() if manager.is_active(): print(\n * 60) print(正在清理 sandbox...) print( * 60) result manager.destroy() print(f清理结果: {result}\n) else: print(\n没有活动的 sandbox 需要清理\n) except Exception as e: print(f\n清理 sandbox 时出错: {str(e)}\n) def signal_handler(signum, frame): 信号处理器处理 CtrlC (SIGINT) 和其他信号 Args: signum: 信号编号 frame: 当前堆栈帧 print(\n\n收到中断信号正在清理资源...) cleanup_sandbox() print(清理完成) sys.exit(0) def main(): 主函数 global _cleanup_done # 重置清理标志 _cleanup_done False # 注册信号处理器处理 CtrlC (SIGINT) signal.signal(signal.SIGINT, signal_handler) # 在 Windows 上SIGBREAK 也可以处理 if hasattr(signal, SIGBREAK): signal.signal(signal.SIGBREAK, signal_handler) print( * 60) print(LangChain AgentRun Browser Sandbox 集成示例) print( * 60) print() try: # 创建 Agent print(正在初始化 LangChain Agent...) agent create_browser_agent() print(Agent 初始化完成\n) # 示例查询 queries [ 创建一个浏览器 sandbox, 获取当前 sandbox 的信息包括 VNC URL, 导航到 https://www.aliyun.com, 截取当前页面截图, ] # 执行查询 for i, query in enumerate(queries, 1): print(f\n{ * 60}) print(f查询 {i}: {query}) print(f{ * 60}\n) try: result agent.invoke({ messages: [{role: user, content: query}] }) # 提取最后一条消息的内容 output result.get(messages, [])[-1].content if isinstance(result.get(messages), list) else result.get(output, str(result)) print(f\n结果:\n{output}\n) # 如果是创建 sandbox自动打开 VNC 查看器 if i 1: try: # 等待一下确保 sandbox 完全创建 import time time.sleep(1) manager get_sandbox_manager() if manager.is_active(): info manager.get_info() vnc_url info.get(vnc_url) if vnc_url: print(f\n检测到 VNC URL: {vnc_url[:80]}...) open_vnc_viewer(vnc_url) print_vnc_info(vnc_url) else: print(\n警告: 未获取到 VNC URL请检查 sandbox 创建是否成功) except Exception as e: print(f打开 VNC 查看器时出错: {str(e)}) import traceback traceback.print_exc() # 如果是获取信息显示 VNC 信息 elif i 2: try: manager get_sandbox_manager() if manager.is_active(): info manager.get_info() if info.get(vnc_url): print_vnc_info(info[vnc_url]) except: pass except Exception as e: print(f查询失败: {str(e)}\n) import traceback traceback.print_exc() # 交互式查询 print(\n * 60) print(进入交互模式输入 quit 或 exit 退出CtrlC 或 CtrlD 中断) print( * 60 \n) while True: try: user_input input(请输入您的查询: ).strip() except EOFError: # 处理 CtrlD (EOF) print(\n\n检测到输入结束 (CtrlD)正在清理资源...) cleanup_sandbox() print(清理完成) break except KeyboardInterrupt: # 处理 CtrlC (在 input 调用期间) print(\n\n检测到中断信号 (CtrlC)正在清理资源...) cleanup_sandbox() print(清理完成) break if not user_input: continue if user_input.lower() in [quit, exit, 退出]: print(\nBye) # 退出前清理 sandbox cleanup_sandbox() break try: result agent.invoke({ messages: [{role: user, content: user_input}] }) output result.get(messages, [])[-1].content if isinstance(result.get(messages), list) else result.get(output, str(result)) print(f\n结果:\n{output}\n) # 检查是否需要打开或显示 VNC 信息 user_input_lower user_input.lower() if 创建 in user_input_lower and sandbox in user_input_lower: # 如果是创建 sandbox自动打开 VNC 查看器 try: # 等待一下确保 sandbox 完全创建 import time time.sleep(1) manager get_sandbox_manager() if manager.is_active(): info manager.get_info() vnc_url info.get(vnc_url) if vnc_url: print(f\n检测到 VNC URL: {vnc_url[:80]}...) open_vnc_viewer(vnc_url) print_vnc_info(vnc_url) else: print(\n警告: 未获取到 VNC URL请检查 sandbox 创建是否成功) except Exception as e: print(f打开 VNC 查看器时出错: {str(e)}) import traceback traceback.print_exc() elif sandbox in user_input_lower or vnc in user_input_lower: # 其他情况只显示信息 try: manager get_sandbox_manager() if manager.is_active(): info manager.get_info() if info.get(vnc_url): print_vnc_info(info[vnc_url]) except: pass except Exception as e: print(f查询失败: {str(e)}\n) import traceback traceback.print_exc() # 清理资源仅在程序正常退出时 cleanup_sandbox() except KeyboardInterrupt: # 处理顶层 KeyboardInterrupt (CtrlC) print(\n\n检测到中断信号 (CtrlC)正在清理资源...) cleanup_sandbox() print(清理完成) sys.exit(0) except EOFError: # 处理顶层 EOFError (CtrlD) print(\n\n检测到输入结束 (CtrlD)正在清理资源...) cleanup_sandbox() print(清理完成) sys.exit(0) except ValueError as e: print(f配置错误: {str(e)}) print(\n提示: 请确保已设置以下环境变量:) print( - DASHSCOPE_API_KEY: DashScope API Key) print( - ALIBABA_CLOUD_ACCOUNT_ID: 阿里云账号 ID) print( - ALIBABA_CLOUD_ACCESS_KEY_ID: 访问密钥 ID) print( - ALIBABA_CLOUD_ACCESS_KEY_SECRET: 访问密钥 Secret) print( - ALIBABA_CLOUD_REGION: 区域默认: cn-hangzhou) except Exception as e: print(f发生错误: {str(e)}) import traceback traceback.print_exc() # 发生错误时也尝试清理 cleanup_sandbox() if __name__ __main__: main()关键功能VNC 自动打开: 创建 Sandbox 后自动打开 VNC 查看器信号处理: 捕获 CtrlC确保资源正确清理交互模式: 支持持续对话复用 Sandbox 实例VNC 可视化集成VNCVirtual Network Computing功能允许您实时查看和监控浏览器在 Sandbox 中的操作过程这对于调试和监控 Agent 行为非常有用。获取 VNC URL创建 Sandbox 后可以通过get_sandbox_infotool 获取 VNC URL# 通过 Agent 调用 result agent.invoke({ messages: [{role: user, content: 获取 sandbox 信息}] }) # 或直接通过管理器获取 manager get_sandbox_manager() info manager.get_info() vnc_url info[vnc_url]自动打开 VNC 查看器在main.py中我们实现了自动打开 VNC 查看器的功能import webbrowser import urllib.parse from pathlib import Path def open_vnc_viewer(vnc_url: str): 自动打开 VNC 查看器 current_dir Path(__file__).parent.absolute() vnc_html_path current_dir / vnc.html if vnc_html_path.exists(): # 通过 URL 参数传递 VNC URL encoded_url urllib.parse.quote(vnc_url, safe) file_url ffile://{vnc_html_path}?url{encoded_url} webbrowser.open(file_url)VNC HTML 页面vnc.html页面会从 URL 参数中读取 VNC URL并自动连接到 VNC 服务器。页面包含以下核心功能noVNC 库加载: 从 CDN 动态加载 noVNC 客户端库自动连接: 读取 URL 参数中的 VNC URL 并自动连接状态显示: 显示连接状态连接中、已连接、已断开手动控制: 支持手动输入 VNC URL、断开重连等操作核心 JavaScript 代码片段// 从 URL 参数获取 VNC URL const urlParams new URLSearchParams(window.location.search); const vncUrl urlParams.get(url); // 加载 noVNC 库 async function loadNoVNC() { const module await import(https://cdn.jsdelivr.net/gh/novnc/noVNCv1.4.0/core/rfb.js); return module.default; } // 连接 VNC async function connectVNC(url) { const RFB await loadNoVNC(); rfb new RFB(vncScreen, url, { shared: true, credentials: { password: } }); rfb.addEventListener(connect, () { console.log(VNC 连接成功); }); }完整的vnc.html文件可以在示例代码仓库中获取。手动使用 VNC 查看器如果自动打开失败您也可以手动使用 VNC 查看器使用 noVNC 在线客户端:访问 noVNC 在线客户端在连接设置中填入 VNC URL点击连接使用本地 VNC HTML 页面:打开vnc.html输入 VNC URL点击连接按钮实时监控功能所有浏览器操作都会在 VNC 中实时显示可以看到 Agent 的每一步操作导航、点击、输入等方便调试和监控 Agent 行为支持交互式操作在 VNC 中直接操作浏览器运行和测试python main.py程序会自动创建 Browser Sandbox打开 VNC 查看器实时查看浏览器操作执行预设查询进入交互模式工作原理为了更好地理解系统架构我们将工作流程拆分为两个部分LangChain Agent 工作流程和SandboxManager 生命周期管理。1. LangChain Agent 工作流程下图展示了 LangChain Agent 如何处理用户请求并调用相应的 Tools首次使用导航网页截取屏幕查询状态清理资源多轮对话用户发起请求例: 访问网页并截图LangChain Agent分析用户意图选择合适的 Toolcreate_browser_sandboxnavigate_to_urltake_screenshotget_sandbox_infodestroy_sandbox调用 SandboxManager调用 SandboxManager调用 SandboxManager调用 SandboxManager调用 SandboxManagerSandboxManager单例实例Tool 返回结果Agent 处理结果生成响应返回用户友好的响应Agent 工作流程说明请求接收用户发起自然语言请求如访问淘宝首页并截图意图分析Agent 分析用户意图决定需要调用哪些 ToolsTool 调用根据任务需求顺序或组合调用多个 ToolsManager 交互所有 Tools 都通过 SandboxManager 单例实例操作 Sandbox结果处理Agent 将 Tool 返回的结果整合成用户友好的响应多轮对话Sandbox 在整个会话中保持活跃支持多轮对话