Agent一句话截真机屏截图Tool这样写

合集 - LangChain× DeepAgents × Hypium:打造鸿蒙自动化测试 Agent(5)

1

2 2行配置!让AI Agent自动读取需求文档

08-05

2

3 1个Tool!让Agent自动生成测试用例

08-06

3

4 鸿蒙真机变开发机:从开箱到usb连通,一步不落

08-06

4

5 PyCharm×鸿蒙Hypium:从0到跑通,少踩90%坑

08-06

5

6 Agent一句话截真机屏:截图Tool这样写

08-07

收起

配套动画视频在:《6 Agent一句话截真机屏:截图Tool这样写》。

前面几节把读文档、生成用例这条链路打通了。www.ycsjb.com自动化测试里还有一块绕不开:对着真机截屏。这一节就写截图 Tool——Agent 说一句话,就能把当前屏幕落到本地。

整条链路可以先记个轮廓:先在 env 里配好 HDC_PATH,再在 json 里写上设备 sn;配置就绪后解析 hdc、检查设备——没设备就失败返回,有设备就截图并拉回本地;文件为空也算失败,www.jlygroup.net成功则保存到 screenshots。Agent 一句话截屏,跑的就是这条链路。下面按关键段落过一遍。

image

先找 hdc 在哪。_get_hdc_cmd 优先看 HDC_PATH(文件或目录),找不到再退回系统 PATH 里的 hdc:

def _get_hdc_cmd() -> list:


    return ['hdc']

核心截图函数 _do_take_screenshot 一进来,先拼好 hdc 命令;有 device_id 就加上 -t:

hdc_cmd = _get_hdc_cmd()

hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])

本地保存路径也要处理好。传进来的若是目录,就自动建目录并用时间戳拼文件名;若是文件路径,只保证父目录存在:

if save_fileis_dir() or (save_filesuffix == '' and not '' in save_filename):


    save_file = save_file / fname

else:

    save_fileparentmkdir(parents=True, exist_ok=True)

设备端截图时,临时路径固定。先试新命令 snapshot_display,再试旧命令 screenshot;两个都失败就返回 None:

复制代码

remote_path = "/data/local/tmp/tmp_screenshotjpeg"

for cmd in [

    hdc_prefix + ['shell', 'snapshot_display', '-f', remote_path],

    hdc_prefix + ['shell', 'screenshot', remote_path],

]:


if not screenshot_ok:

    return None

复制代码

截完后用 file recv 拉回电脑。本地文件存在且非空才返回路径,异常统一返回 None:

复制代码

subprocessrun(

    hdc_prefix + ['file', 'recv', remote_path, str(save_file)],


)

if save_fileexists() and save_filestat()st_size > 0:

    return str(save_file)

return None

except Exception:

复制代码

工具侧,TakeScreenshotInput 声明参数 schema:设备 ID 可选,保存路径必填。TakeScreenshotTool 名字叫 take_screenshot,description 里写清干什么、参数怎么用,Agent 才知道何时调用。

_run 里先做路径规范化——相对路径一律接到项目根目录,已是绝对路径则原样使用:

base_dir = ospathdirname(ospathabspath(__file__))

if save_pathstartswith('/'):

    save_path = ospathnormpath(ospathjoin(base_dir, save_pathlstrip('/')))

elif not ospathisabs(save_path):

    save_path = ospathnormpath(ospathjoin(base_dir, save_path))

接着做前置检查:跑一遍 hdc list targets。工具不可用,或没连上设备,直接返回中文错误:

复制代码

check = subprocessrun(

    hdc_prefix + ['list', 'targets'],


)

if checkreturncode != 0:

    return f"错误:hdc工具不可用 - "

if not output or 'No device' in output or '无设备' in output:

    return "错误:未检测到连接的HarmonyOS设备"

复制代码

检查通过后调用底层截图。成功报路径和大小;失败、超时、hdc 找不到,则分别返回对应提示:

复制代码

result_path = _do_take_screenshot(save_path, device_id)

if result_path:


    return f"截图成功,保存至:{result_path} ({file_size} bytes)"

else:

    return "错误:截图失败"

except subprocessTimeoutExpired:

    return "错误:截图超时(>15秒)"

复制代码

熟悉截图 Tool 之后,改一下触发工具的提示词,表达清楚就可以了:

复制代码

def main():

    agent = create_test_agent()

    result = agentinvoke({

        "messages": [{"role": "user", "content": "使用 take_screenshot 工具对当前设备屏幕截一张图,保存到 testcases/screenshots/app_screenshotpng 。"}]

    })

    print(result["messages"][-1]content[0]["text"])

复制代码

完整test_toolspy的代码如下:

复制代码

import os

import subprocess

from typing import Optional, Type

from pathlib import Path

from datetime import datetime

from pydantic import BaseModel, Field

from langchain_coretools import BaseTool

import yaml

"""

B站、抖音:@老陈说编程

"""

class GenerateTestCasesInput(BaseModel):

    test_cases: list = Field(description="测试用例列表,每个用例含 case_id, case_name, description, steps")

    output_path: str = Field(description="YAML文件输出路径")

class GenerateTestCasesTool(BaseTool):

    name: str = "generate_test_cases"

    description: str = "将测试用例列表保存为YAML文件。用例格式:每个用例包含case_id(用例ID)、case_name(用例名称)、description(描述)、steps(步骤列表)。每个step包含action(操作类型:click/input/clear/assert_exists/assert_not_exists/assert_text/assert_toast/wait/screenshot/swipe/back)和params(参数字典,含selector元素ID)"

    args_schema: Type[BaseModel] = GenerateTestCasesInput

    def _run(self, test_cases: list, output_path: str) -> str:

        try:

            base_dir = ospathdirname(ospathabspath(__file__))

            if output_pathstartswith('/'):

                output_path = ospathnormpath(ospathjoin(base_dir, output_pathlstrip('/')))

            elif not ospathisabs(output_path):

                output_path = ospathnormpath(ospathjoin(base_dir, output_path))

            for i, case in enumerate(test_cases):

                if 'case_id' not in case or 'steps' not in case:

                    return f"错误:第{i + 1}个用例缺少case_id或steps"

            output_file = Path(output_path)

            output_fileparentmkdir(parents=True, exist_ok=True)

            with open(output_file, 'w', encoding='utf-8') as f:

                yamldump({'test_cases': test_cases}, f, allow_unicode=True, default_flow_style=False)

            return f"成功保存{len(test_cases)}个用例至:{output_path}"

        except Exception as e:

            return f"保存失败:{str(e)}"

"""

B站、抖音:@老陈说编程

"""

def _get_hdc_cmd() -> list:

    """

    获取 hdc 命令路径

    hdc(HarmonyOS Device Connector)是 HarmonyOS 的命令行工具,用于与设备通信。

    本函数按以下优先级查找 hdc:

    1 优先使用环境变量 HDC_PATH 指定的路径(可以是可执行文件或目录)

    2 如果未配置环境变量或找不到,直接使用 'hdc'(依赖系统 PATH)

    Returns:

        list: hdc 命令列表,例如 ['hdc'] 或 ['C:\\path\\to\\hdcexe']

    """

    # 从环境变量获取 HDC_PATH 配置

    hdc_path = osenvironget("HDC_PATH", "")

    if hdc_path:

        p = Path(hdc_path)

        # 如果 HDC_PATH 直接指向一个文件,返回该文件路径

        if pis_file():

            return [str(p)]

        # 如果 HDC_PATH 是一个目录,在目录下查找 hdcexe(Windows)或 hdc

        if pis_dir():

            for name in ['hdcexe', 'hdc']:

                exe = p / name

                if exeexists():

                    return [str(exe)]

    # 默认使用系统 PATH 中的 hdc 命令

    return ['hdc']

def _do_take_screenshot(save_path: str, device_id: Optional[str] = None, label: str = "") -> Optional[str]:

    """

    统一的截图底层实

    完整流程:

    1 解析保存路径(支持传目录或完整文件路径)

    2 在设备端执行截图命令(优先 snapshot_display,兼容旧版 screenshot)

    3 验证截图是否在设备端成功生成

    4 将截图从设备拉取到本地

    5 验证本地文件有效性

    Args:

        save_path: 保存文件路径或目录

            - 如果是目录(如 'testcases/screenshots/'),自动生成带时间戳的文件名

            - 如果是完整路径(如 'testcases/screenshots/beforepng'),直接使用该路径

        device_id: 设备ID(可选)

            - 如果有多个设备连接,通过此参数指定目标设备

            - 如果为 None,使用默认设备

        label: 截图标签(仅当 save_path 是目录时使用)

            - 用于自动生成文件名,例如 label='before' → 'before_20260723_160000png'

    Returns:

        成功: 返回截图文件的绝对路径(字符串)

        失败: 返回 None

    """

    try:

        # 获取 hdc 基础命令

        hdc_cmd = _get_hdc_cmd()

        # 构建设备前缀命令:如果指定了 device_id,添加 '-t <device_id>' 参数

        # 例如 ['hdc', '-t', '2L1111110223916']

        hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])

        # ========== 步骤1:解析并准备本地保存路径 ==========

        save_file = Path(save_path)

        # 判断 save_path 是目录还是文件路径:

        # - 如果是已存在的目录

        # - 或者路径没有后缀名(没有 ''),也视为目录

        if save_fileis_dir() or (save_filesuffix == '' and not '' in save_filename):

            # 自动创建目录(parents=True 允许递归创建,exist_ok=True 目录已存在不报错)

            save_filemkdir(parents=True, exist_ok=True)

            # 生成时间戳,格式:年月日_时分秒(例如 20260723_160000)

            timestamp = datetimenow()strftime('%Y%m%d_%H%M%S')

            # 生成文件名:有 label 时用 label_时间戳png,否则用 screenshot_时间戳png

            fname = f"{label}_{timestamp}png" if label else f"screenshot_{timestamp}png"

            # 拼接完整文件路径

            save_file = save_file / fname

        else:

            # 如果是文件路径,确保父目录存在

            save_fileparentmkdir(parents=True, exist_ok=True)

        # ========== 步骤2:设备端临时路径 ==========

        # HarmonyOS 设备上的临时截图保存位置(应用沙盒外的公共临时目录)

        remote_path = "/data/local/tmp/tmp_screenshotjpeg"

        # ========== 步骤3:在设备端执行截图命令(兼容新旧版本) ==========

        screenshot_ok = False

        # 尝试两种截图命令:

        # 1 snapshot_display:新版 HarmonyOS 的截图命令(API 9+)

        # 2 screenshot:旧版兼容命令

        for cmd in [

            hdc_prefix + ['shell', 'snapshot_display', '-f', remote_path],

            hdc_prefix + ['shell', 'screenshot', remote_path],

        ]:

            # 执行截图命令,超时时间 15 秒

            r = subprocessrun(

                cmd, capture_output=True, text=True, timeout=15,

                encoding='utf-8', errors='replace'

            )

            # 合并 stdout 和 stderr,转为小写用于错误检测

            combined = (rstdout + rstderr)lower()

            # 检查命令是否执行成功:

            # - 返回码为 0

            # - 输出中不包含 'fail' 或 'error' 关键字

            if rreturncode == 0 and 'fail' not in combined and 'error' not in combined:

                # ========== 步骤4:验证设备端截图文件确实生成了 ==========

                # 执行 ls -l 命令检查文件是否存在、大小是否正常

                check = subprocessrun(

                    hdc_prefix + ['shell', 'ls', '-l', remote_path],

                    capture_output=True, text=True, timeout=5,

                    encoding='utf-8', errors='replace'

                )

                # 再次验证:ls 命令成功、没有 "No such file"、文件名出现在输出中

                if checkreturncode == 0 and 'No such file' not in checkstdout and remote_pathsplit('/')[-1] in checkstdout:

                    screenshot_ok = True

                    # 截图成功,跳出循环

                    break

        # 如果两种命令都失败了,直接返回 None

        if not screenshot_ok:

            return None

        # ========== 步骤5:将截图从设备拉取到本地 ==========

        # 使用 hdc file recv 命令:从设备接收文件到本地

        # 命令格式:hdc file recv <设备端路径> <本地路径>

        subprocessrun(

            hdc_prefix + ['file', 'recv', remote_path, str(save_file)],

            capture_output=True, text=True, timeout=10,

            encoding='utf-8', errors='replace'

        )

        # ========== 步骤6:验证本地文件有效性 ==========

        # 检查文件是否存在,并且文件大小大于 0(避免空文件)

        if save_fileexists() and save_filestat()st_size > 0:

            # 返回绝对路径字符串

            return str(save_file)

        # 文件无效,返回 None

        return None

    except Exception:

        # 捕获所有异常(超时、文件IO错误等),统一返回 None

        return None

class TakeScreenshotInput(BaseModel):

    """

    TakeScreenshotTool 的参数定义模型

    使用 Pydantic BaseModel 定义工具的输入参数 schema,

    LangChain 会自动根据这个模型生成参数描述和验证。

    """

    # 设备ID,可选参数,默认 None(使用默认设备)

    device_id: Optional[str] = Field(default=None, description="设备ID(可选)")

    # 截图保存路径,必填参数

    save_path: str = Field(description="截图保存路径")

class TakeScreenshotTool(BaseTool):

    """

    HarmonyOS 截图工具(LangChain Tool 封装)

    这是供 DeepAgent 调用的工具类,Agent 通过调用 take_screenshot 工具来截取设备屏幕。

    工具会:

    1 自动处理相对路径/绝对路径转换(基于项目根目录)

    2 检查 hdc 工具可用性

    3 检查设备连接状态

    4 调用底层 _do_take_screenshot 执行实际截图

    5 返回友好的成功/失败信息(包含文件路径和大小)

    """

    # 工具名称(Agent 通过此名称调用工具)

    name: str = "take_screenshot"

    # 工具描述(Agent 会根据此描述理解工具用途和参数)

    description: str = "对HarmonyOS设备当前屏幕截图并保存到本地文件。参数save_path为截图保存的本地路径(建议testcases/screenshots/目录下)。成功返回文件路径和大小,失败返回错误信息。"

    # 参数 schema,指向上面定义的 TakeScreenshotInput

    args_schema: Type[BaseModel] = TakeScreenshotInput

    def _run(self, save_path: str, device_id: Optional[str] = None) -> str:

        """

        工具执行入口(LangChain 调用此方法)

        Args:

            save_path: 截图保存路径(可以是相对路径或绝对路径)

            device_id: 设备ID(可选)

        Returns:

            str: 执行结果消息

                - 成功:"截图成功,保存至:<绝对路径> (<文件大小> bytes)"

                - 失败:"错误:<错误原因>"

        """

        try:

            # ========== 步骤1:路径规范化处理 ==========

            # 获取当前文件所在目录(项目根目录)

            base_dir = ospathdirname(ospathabspath(__file__))

            # 将传入的路径转换为绝对路径:

            if save_pathstartswith('/'):

                # 如果以 '/' 开头,视为相对于项目根目录的路径(去掉开头的 / 后拼接)

                save_path = ospathnormpath(ospathjoin(base_dir, save_pathlstrip('/')))

            elif not ospathisabs(save_path):

                # 如果是相对路径(不以 / 开头,也不是 Windows 绝对路径如 C:\),

                # 基于项目根目录拼接成绝对路径

                save_path = ospathnormpath(ospathjoin(base_dir, save_path))

            # 如果已经是绝对路径(Windows 如 E:\),直接使用

            # ========== 步骤2:准备 hdc 命令 ==========

            hdc_cmd = _get_hdc_cmd()

            hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])

            # ========== 步骤3:前置检查 - hdc 可用性和设备连接 ==========

            # 执行 'hdc list targets' 命令列出已连接设备

            check = subprocessrun(

                hdc_prefix + ['list', 'targets'],

                capture_output=True, text=True, timeout=5,

                encoding='utf-8', errors='replace'

            )

            # 检查 hdc 命令本身是否可用

            if checkreturncode != 0:

                return f"错误:hdc工具不可用 - {checkstderrstrip() or checkstdoutstrip()}"

            output = checkstdoutstrip()

            # 检查是否有设备连接

            # hdc list targets 在无设备时可能输出 "No device"、空行、或 "无设备"

            if not output or 'No device' in output or '无设备' in output:

                return "错误:未检测到连接的HarmonyOS设备"

            # ========== 步骤4:执行截图 ==========

            result_path = _do_take_screenshot(save_path, device_id)

            if result_path:

                # 截图成功:获取文件大小,返回成功消息

                save_file = Path(result_path)

                file_size = save_filestat()st_size

                return f"截图成功,保存至:{result_path} ({file_size} bytes)"

            else:

                # 截图失败

                return "错误:截图失败"

        except subprocessTimeoutExpired:

            # 命令执行超时

            return "错误:截图超时(>15秒)"

        except FileNotFoundError:

            # hdc 命令未找到(系统找不到 hdcexe)

            return "错误:hdc命令未找到,请确保已安装HarmonyOS SDK并配置HDC_PATH环境变量"

        except Exception as e:

            # 其他未知异常

            return f"截图失败:{str(e)}"

def get_all_tools() -> list:

    return [

        GenerateTestCasesTool(),

        TakeScreenshotTool()

    ]

复制代码

完整run_automation_testpy的代码如下:

复制代码

import os

from deepagentsbackends import LocalShellBackend

from dotenv import load_dotenv

from deepagents import create_deep_agent

from test_tools import get_all_tools

"""

B站、抖音:@老陈说编程

"""

load_dotenv()

osenviron["OPENAI_API_KEY"] = osgetenv("API_KEY")

osenviron["OPENAI_BASE_URL"] = osgetenv("BASE_URL")

model = f"openai:{osgetenv('MODEL')}"

def create_test_agent():

    return create_deep_agent(

        model=model,

        system_prompt="HarmonyOS 自动化测试专家。",

        tools=get_all_tools(),

        skills=["/skills/"],

        backend=LocalShellBackend(root_dir=osgetcwd(), virtual_mode=True, inherit_env=True),

        debug=True

    )

def main():

    agent = create_test_agent()

    result = agentinvoke({

        "messages": [{"role": "user", "content": "使用take_screentshot工具对当前设备屏幕截一张,"

                                                "保存到 testcases/screenshots/app_screenshotpng 。"}]

    })

    print(result["messages"][-1]content[0]["text"])

if __name__ == '__main__':

    main()

复制代码

跑起来后,Agent 会调用 take_screenshot,把当前屏幕保存到指定路径。有了这步,后面的启停应用、测前测后截图才有落点。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

友情链接更多精彩内容