void test_conv3d () { const char * filename_input = "smallCube.sgy" ;   //输入文件名 const char * filename_output = "conv_small.sgy" ;             //输出文件名 FILE* fp_input = nullptr;                                 //输入文件的文件指针 FILE* fp_output = nullptr;                                 //输出文件的文件指针 bhed fileheader;                                           //定义文件头 segy* traceheader_array = nullptr;                                 //道头数组指针 float * datacube_input = nullptr;                         //输入地震数据的三维指针 float * datacube_output = nullptr;                       //输出地震数据的三维指针 unsigned int size_fileheader = sizeof (fileheader); unsigned int size_traceheader = sizeof (segy); unsigned int nline = 0 ;                                         //三维地震数据的测线数 unsigned int ncdp = 0 ;                                         //三维地震数据的CDP数 unsigned int nt = 0 ;                                         //三维地震数据的时间采样数 long long ntrace = 0 ; long long size_inputcube = 0 ; long long size_trace = 0 ; nline = 100 ; ncdp = 500 ; fp_input = fopen(filename_input, "rb" ); if (fp_input == nullptr) { printf ( "Cannot open this input file!!!\n" ); qDebug() << "Cannot open this input file!!!\n" ; system( "pause" ); return ; } //end if(fp_input==nullptr) fp_output = fopen(filename_output, "wb" ); if (fp_output == nullptr) { printf ( "Cannot open\create this output file!!!\n" );        qDebug() << "Cannot open\create this output file!!!\n" ; system( "pause" ); return ; } //end if(fp_output==nullptr) /*判断地震数据是否为规则的*/ fread(&fileheader, size_fileheader, 1 , fp_input); nt = fileheader.hns; size_trace = size_traceheader + nt * sizeof ( float ); _fseeki64(fp_input, 0 , SEEK_END); size_inputcube = _ftelli64(fp_input); ntrace = (size_inputcube - size_fileheader) / (size_trace); _fseeki64(fp_input, size_fileheader, SEEK_SET); if (ntrace != (nline * ncdp)) { printf ( "Cube is not regularized!!!\n" ); return ; } /*读取规则的地震数据体及处理，输出*/ int input_dims[ 3 ] = { nline, ncdp, nt }; int nAllSize = input_dims[ 0 ] * input_dims[ 1 ] * input_dims[ 2 ]; datacube_input = new float [nAllSize]; memset (datacube_input, 0 , nAllSize * sizeof ( float )); traceheader_array = (segy*) calloc (nline * ncdp, size_traceheader); datacube_output = new float [nAllSize]; memset (datacube_output, 0 , nline * ncdp * nt * sizeof ( float )); for ( int iline = 0 ; iline < nline; iline++) {   //读取三维地震属性体 for ( int icdp = 0 ; icdp < ncdp; icdp++) { int nIndex = (iline * ncdp + icdp) * nt; //printf("iline=%d,icdp=%d,nIndex=%d\n", iline, icdp, nIndex); fread(&traceheader_array[iline * ncdp + icdp], size_traceheader, 1 , fp_input); fread(&datacube_input[nIndex], nt * sizeof ( float ), 1 , fp_input); } //end for(int icdp = 0; icdp < ncdp;icdp++) } //end for(int iline = 0; iline < nline; iline++) int cube2_dims[ 3 ] = { 21 , 21 , 21 }; int nCube2Size = cube2_dims[ 0 ] * cube2_dims[ 1 ] * cube2_dims[ 2 ]; float * pCube2 = new float [nCube2Size]; memset (pCube2, 0 , sizeof ( float ) * nCube2Size); getCube2(cube2_dims[ 0 ], cube2_dims[ 1 ], cube2_dims[ 2 ], pCube2); //开始计时 clock_t start_clock, end_clock; start_clock = clock(); // 初始化输入数据（主机到设备） // -------------------------- 8. 执行卷积并计时 -------------------------- cudnn_conv_3d(datacube_input, pCube2, datacube_output, input_dims, cube2_dims); //结束计时 end_clock = clock(); double endtime = ( double )(end_clock - start_clock) / CLOCKS_PER_SEC; std :: cout << "Total time:" << endtime << " s" << std :: endl ; //ms为单位 qDebug() << "Total time:" << endtime << " s" << endl ; fwrite(&fileheader, size_fileheader, 1 , fp_output); for ( int iline = 0 ; iline < nline; iline++) {   //写出三维地震属性体 for ( int icdp = 0 ; icdp < ncdp; icdp++) { int nIndex = (iline * ncdp + icdp) * nt; fwrite(&traceheader_array[iline * ncdp + icdp], size_traceheader, 1 , fp_output); fwrite(&datacube_output[nIndex], nt * sizeof ( float ), 1 , fp_output); } //end for(int icdp = 0; icdp < ncdp;icdp++) } //end for(int iline = 0; iline < nline; iline++) // -------------------------- 9. 释放资源 -------------------------- fclose(fp_input); fclose(fp_output); delete[]datacube_input; delete[]datacube_output; free (traceheader_array);

阿里云百炼平台的 API Key ，用于调用大模型能力 # 请前往 https : //bailian.console.aliyun.com/?tab=app#/api-key 创建和查看 DASHSCOPE_API_KEY =sk-your-bailian-api-key # 阿里云账号的访问密钥 ID 和访问密钥 Secret ，用于 AgentRun SDK 鉴权 ALIBABA_CLOUD_ACCESS_KEY_ID =your-ak ALIBABA_CLOUD_ACCESS_KEY_SECRET =your-sk ALIBABA_CLOUD_ACCOUNT_ID =your-main-account-id ALIBABA_CLOUD_REGION =cn-hangzhou # browser sandbox 模板的名称，可以在 https : //functionai.console.aliyun.com/cn-hangzhou/agent/runtime/sandbox 控制台创建 BROWSER_TEMPLATE_NAME =sandbox-your-template-name # agentrun 的控制面和数据面的 API 端点请求地址，默认cn-hangzhou AGENTRUN_CONTROL_ENDPOINT =agentrun. cn -hangzhou. aliyuncs . com

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_type= TemplateType . BROWSER ,                template_name=template_name,                sandbox_idle_timeout_seconds=idle_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 f "Sandbox 已销毁: {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

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_name=template_name, idle_timeout=idle_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 (         default = "load" ,        description= "等待页面加载的状态: load, domcontentloaded, networkidle"    )     timeout : int = Field (         default = 30000 ,        description= "超时时间（毫秒），默认 30000"    ) @tool (args_schema= NavigateInput ) 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_until=wait_until, timeout=timeout)                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 (path=filename)                 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 (        model=model_name,        api_key=api_key,        base_url=base_url,        temperature= 0.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 (        model=model,        tools=tools,        system_prompt=system_prompt,    )         return agent def get_available_tools ():     "" "获取所有可用的工具列表" ""     return [

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, directory= str (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 (target=run_server, daemon= True )                thread. start ()                                _http_server = server         cpnapt.com        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"m.cpnapt.com                # 检查文件是否存在         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  jrszb.cpnapt.com            http_url = f "http://localhost:{port}/vnc.html?url={encoded_url}"                        # 打开浏览器             print (f "\n正在打开 VNC 查看器..." )             print (f "HTTP 服务器运行在: http://localhost:{port}" )             print (f "VNC URL: {vnc_url[:80]}..." )             print (f "完整 URL: {http_url[:100]}..." )            webbrowser. open (http_url)             print (f "VNC 查看器已打开" )             print (f "VNC URL 已通过 URL 参数自动设置，页面加载后会自动连接" )         else :            # 如果 HTTP 服务器启动失败，尝试使用 file : // 协议             print (f "HTTP 服务器启动失败，尝试使用文件协议..." )            encoded_url = urllib. parse . quote (vnc_url, safe= '' )            file_url = f "file://{vnc_html_path}?url={encoded_url}"            webbrowser. open (file_url)               print (f "VNC 查看器已打开（使用文件协议）" )             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  ijs.zw0769.com         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  jrs.qdhyyy.cn        # 防止重复清理     if _cleanup_done : jhp.zw0769.com        return        _cleanup_done = True         try :        manager = get_sandbox_manager ()         if manager. is_active ():             print ( "\n" + "=" * 60 )             print ( "正在清理 sandbox..." )             print ( "=" * 60 )            result = manager. destroy ()    qkk.zw0769.com          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):     "" "    信号处理器，处理 Ctrl+C (SIGINT) 和其他信号        Args:        signum: 信号编号        frame: 当前堆栈帧    " ""     print ( "\n\n收到中断信号，正在清理资源..." )     cleanup_sandbox ()     print ( "清理完成" )    sys. exit ( 0 ) def main ():     "" "主函数" ""     global _cleanup_done        # 重置清理标志    _cleanup_done = False        # 注册信号处理器，处理 Ctrl +C ( 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 ({          hpm.zw0769.com           "messages" : [{ "role" : "user" , "content" : query}]                })                                # 提取最后一条消息的内容                output = result. get ( "messages" , [])[- 1 ]. content if isinstance (result. get ( "messages" ), list) else result. get ( "output" , str (result))      fwn.zw0769.com              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 ()       qca.zw0769.com                  if manager. is_active ():                            info = manager. get_info ()                             if info. get ( 'vnc_url' ):                                 print_vnc_info (info[ 'vnc_url' ])    ytm.zw0769.com                 except :                        pass                        except Exception as e :                 print (f "查询失败: {str(e)}\n" )                 import traceback                traceback. print_exc ()                # 交互式查询         print ( "\n" + "=" * 60 )         print ( "进入交互模式（输入 'quit' 或 'exit' 退出，Ctrl+C 或 Ctrl+D 中断）" )         print ( "=" * 60 + "\n" )        tms.zw0769.com         while True :             try :                user_input = input ( "请输入您的查询: " ). strip ()            except EOFError :                # 处理 Ctrl +D ( EOF )    kib.zw0769.com               print ( "\n\n检测到输入结束 (Ctrl+D)，正在清理资源..." )                 cleanup_sandbox ()                 print ( "清理完成" )                 break            except KeyboardInterrupt :                # 处理 Ctrl +C (在 input 调用期间)                 print ( "\n\n检测到中断信号 (Ctrl+C)，正在清理资源..." )                 cleanup_sandbox ()                 print ( "清理完成" )                 break                         if not user_input :                 continue                         if user_input. lower () in [ 'quit' , 'exit' , '退出' ]:                 print ( "\nBye" )                # 退出前清理 sandbox       ddy.zw0769.com            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 )                         m.cpnapt.com                        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 :              zol.zw0769.com           print (f "打开 VNC 查看器时出错: {str(e)}" )                         import traceback                        traceback. print_exc ()                elif "sandbox" in user_input_lower or "vnc" in user_input_lower :                    # 其他情况只显示信息      dsv.zw0769.com               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 :jhp.zw0769.com        # 处理顶层 KeyboardInterrupt ( Ctrl +C)         print ( "\n\n检测到中断信号 (Ctrl+C)，正在清理资源..." )         cleanup_sandbox ()         print ( "清理完成" )        sys. exit ( 0 )    except EOFError :        # 处理顶层 EOFError ( Ctrl +D)         print ( "\n\n检测到输入结束 (Ctrl+D)，正在清理资源..." )         cleanup_sandbox ()         print ( "清理完成" )        sys. exit ( 0 )    except ValueError as e :     dtt.zw0769.com      print (f "配置错误: {str(e)}" )         print ( "\n提示: 请确保已设置以下环境变量:" )         print ( "   - DASHSCOPE_API_KEY: DashScope API Key" )         print ( "   - ALIBABA_CLOUD_ACCOUNT_ID: 阿里云账号 ID" )