站竣凰 发表于 4 天前

【GUI-Agent】阶跃星辰 GUI-MCP 解读---(2)---决策层

【GUI-Agent】阶跃星辰 GUI-MCP 解读---(2)---决策层


目录

[*]【GUI-Agent】阶跃星辰 GUI-MCP 解读---(2)---决策层

[*]0x00 摘要
[*]0x01 LocalServer

[*]1.1 核心功能概述
[*]1.2 详细功能说明
[*]1.3 在系统架构中的位置
[*]1.4 实现

[*]automate_step

[*]流程图
[*]代码

[*]ask_llm_anything

[*]流程图
[*]代码



[*]0x02 Parser0920Summary 的功能

[*]2.1 核心功能概览
[*]2.2 详细功能说明
[*]2.3 重点函数分析

[*]2.3.1 str2action

[*]核心步骤
[*]使用场景
[*]代码

[*]2.3.2 make_status_prompt

[*]2.4 THINK 标签
[*]2.5 COT 输出与执行流程
[*]2.6 支持的动作类型

[*]动作空间定义
[*]提示词引导
[*]动作解析
[*]动作验证
[*]实际执行流程


[*]0x03 交互
[*]0x04 模型分发

[*]4.1 模型分发机制
[*]4.2 本地模型流程
[*]4.3 云端模型流程

[*]0x05 过渡到执行层
[*]0xFF 参考


0x00 摘要

25年底,阶跃星辰升级发布了全新的AI Agent系列模型Step-GUI,包括云端模型Step-GUI、首个面向GUI Agent的MCP协议:GUI-MCP(Graphical User Interface - Model Context Protocol),这是首个专为图形用户界面自动化而设计的 MCP 实现,兼顾标准化与隐私保护。因此,我们就来解读这个MCP协议,顺便看看端侧Agent的实现架构。
本文是第二篇,主要是介绍决策层,本层在任何情况下(是/非MCP)都会用到。
因为是反推解读,而且时间有限,所以可能会有各种错误,还请大家不吝指出。
0x01 LocalServer

LocalServer 是本地 GUI Agent 服务器实现,作为系统核心组件之一,负责协调模型推理、环境管理和任务执行。总的来说,LocalServer 是 GUI Agent 系统的大脑,负责将高层任务请求转化为具体的模型推理和设备操作序列。
1.1 核心功能概述


[*]会话管理
[*]模型推理调度,即负责与 LLM 交互,解析模型输出为可执行动作
[*]环境状态跟踪
[*]日志记录与持久化
1.2 详细功能说明

会话管理(get_session 方法)

[*]创建新的任务会话并分配唯一 session_id(UUID 保证唯一性)
[*]初始化会话配置(任务描述、任务类型、模型配置等)
[*]记录会话开始日志
自动化步骤执行(automate_step 方法):这是 LocalServer 的核心方法,负责执行单步自动化操作:
① 环境状态读取:读取历史日志,获取之前的环境和动作记录

[*]解析当前观察(主要是屏幕截图)
[*]支持图像缩放预处理
② 消息构建:

[*]使用 Parser 将任务、环境和历史动作转换为模型输入消息
[*]支持图像缩放预处理
③ 模型调用:调用 ask_llm_anything 函数与 LLM 交互,并记录模型推理时间和响应
④ 动作解析:使用 Parser 将模型响应解析为可执行动作
⑤ 日志记录:

[*]保存完整的交互日志(环境、动作、模型输入输出等)
[*]支持调试模式下的详细输出
日志管理系统:使用 LocalServerLogger 管理会话日志,提供:

[*]日志与图像文件分离存储
[*]支持日志读取和图像保存
1.3 在系统架构中的位置

LocalServer 在整个系统中起到承上启下的协调作用:
MCP 接口 / 命令行工具
    ↓
(协调层)
    ↓             ↓
Parser(环境转换)LLM 模型(推理决策)
                  ↓
设备执行层(act_on_device 等)1.4 实现

automate_step

automate_step将 GUI 交互的观测信息(截图 + 文本)转换为可执行的动作指令,是 GUI-Agent 决策的核心环节。通过解析器适配不同任务类型,图片预处理适配不同模型,日志全流程记录保证可溯源;
流程图


[*]数据预处理阶段:核心是将用户的截图 + 查询转换为 LLM 可识别的格式,同时通过日志器保存原始数据,保证可溯源;
[*]适配优化阶段:通过图片预处理(缩放)适配不同 LLM 的输入要求,深拷贝原始消息避免预处理修改日志数据;
[*]LLM 交互阶段:封装模型调用参数,记录调用全生命周期耗时,便于性能监控和调优;
[*]结果输出阶段:将 LLM 的文本响应解析为结构化动作,同时返回步骤数,适配 Copilot 服务的多步执行逻辑;

代码

automate_step 的代码如下:
def automate_step(self, payload: dict) -> dict:
    """
    Automate a step in the Copilot service.
    核心功能:执行Copilot服务的单步自动化逻辑,输入观测信息(截图+查询),输出下一步操作指令
    """
    # 从请求载荷中提取会话ID(用于日志隔离和溯源)
    session_id = payload["session_id"]

    # 初始化本地服务器日志器:指定日志目录、图片目录、会话ID,用于保存步骤截图和日志
    logger = LocalServerLogger({
      "log_dir": self.server_config["log_dir"],
      "image_dir": self.server_config["image_dir"],
      "session_id": session_id
    })

    # 计算当前步骤数:日志列表长度-1(logs为全局/类内维护的历史日志,0号为配置日志)
    current_ste = len(logs) - 1

    # 读取0号日志(配置日志),解析核心配置信息
    config_log = logs
    config_dict = config_log['message']

    # 提取任务类型、模型配置、核心任务描述
    task_type = config_dict['task_type']
    model_config = config_dict['model_config']
    task = config_dict['task']

    # 从观测信息中获取当前步骤的核心输入:截图+用户查询
    observation = payload['observation']
    # 提取截图的URL并读取图片数据
    image_url = observation['screenshot']['image_url']['url']
    image = read_from_url(image_url)
    # 保存图片到本地,生成内部可访问的图片URL(用于日志和LLM输入)
    image_inner_url = logger.save_image(image, f"step_{current_ste+1}")
    # 提取用户查询(无则为空字符串)
    query = observation.get('query', '')

    # 定义内部函数:从历史日志中提取环境信息和动作序列(logs为非配置日志)
    def get_envs_acts_from_logs(logs):
      environments = []# 存储历史环境信息(截图+用户输入)
      actions = []       # 存储历史动作指令
      for log in logs:
            msg = log['message']
            environments.append(msg['environment'])
            actions.append(msg['action'])
      return environments, actions

    # 提取历史环境和动作,构建上下文
    environments, actions = get_envs_acts_from_logs(logs)

    # 构建当前步骤的环境信息(本地图片URL+用户查询)
    current_env = {
      "image": image_inner_url,
      "user_comment": query
    }
    # 将当前环境加入历史列表,用于LLM上下文构建
    environments.append(current_env)      
    # 根据任务类型获取对应的解析器(适配不同任务的消息构建/动作解析逻辑)
    parser = get_parser(task_type)
    # 调用解析器:将任务、历史环境/动作转换为LLM可识别的消息格式
    messages_to_ask = parser.env2messages4ask(
      task = task,
      environments = environments,
      actions = actions,
    )

    # 深拷贝原始消息(用于日志记录,避免后续图片预处理修改原始数据)
    asked_messages = deepcopy(messages_to_ask)

    # 提取模型配置:模型名称、提供商(默认eval)
    model_name = model_config['model_name']
    model_provider = model_config.get('model_provider', 'eval')

    # 提取LLM调用参数,设置默认值(保证参数完整性)
    args = model_config.get('args', {
      "temperature": 0.1,    # 低温度保证输出稳定
      "top_p": 1.0,          # 采样策略
      "frequency_penalty": 0.0,# 重复惩罚
      "max_tokens": 512,   # 最大生成令牌数
    })

    # 提取图片预处理配置(None则不处理)
    image_preprocess = model_config.get('image_preprocess', None)

    # 图片预处理:若指定目标尺寸,缩放消息中的图片
    if image_preprocess is not None:
      if "target_image_size" in image_preprocess:
            target_image_size = image_preprocess["target_image_size"]
            
            # 定义内部函数:遍历消息,缩放所有图片URL
            def resize_image_in_messages(messages, target_size):
                for msg in messages:
                  # 跳过纯文本消息
                  if type(msg['content']) == str:
                        continue
                  # 遍历消息内容,处理图片URL
                  for content in msg['content']:
                        if content['type'] == "text":
                            continue
                        # 提取图片URL并生成缩放后的Base64 URL
                        image_url = content['image_url']['url']
                        image_resize_url = make_b64_url(image_url, resize_config={
                            "is_resize": True,
                            "target_image_size": target_size
                        })
                        # 替换为缩放后的URL(适配LLM的图片输入要求)
                        content['image_url']['url'] = image_resize_url
            
            # 执行图片缩放:修改待发送给LLM的消息中的图片URL
            resize_image_in_messages(messages_to_ask, target_image_size)
   
    # 记录LLM调用开始时间(监控耗时)
    llm_start_time = time.time()
    # 调用LLM:传入提供商、模型名、消息、参数,获取响应
    response = ask_llm_anything(
      model_provider=model_provider,
      model_name=model_name,
      messages=messages_to_ask,
      args=args
    )
    # 记录LLM调用结束时间
    llm_end_time = time.time()

    # 调用解析器:将LLM文本响应解析为结构化的动作指令
    action = parser.str2action(response)

    # 构建当前步骤的日志消息(完整记录上下文、模型响应、耗时等)
    log_message = {
      "environment": current_env,          # 当前环境信息
      "action": action,                  # 解析后的动作指令
      "asked_messages": asked_messages,    # 原始LLM请求消息(未预处理)
      "model_response": response,          # LLM原始响应
      "model_config": model_config,      # 模型配置
      "llm_cost": {                        # LLM调用耗时统计
            "llm_time": llm_end_time - llm_start_time,
            "llm_start_time": llm_start_time,
            "llm_end_time": llm_end_time
      },
    }
   
    # 返回核心结果:下一步动作指令、当前步骤数
    return {
      "action": action,
      "current_step": current_ste + 1
    }ask_llm_anything

ask_llm_anything 是阶跃星辰 GUI-Agent 中通用的 LLM 调用核心函数,专门负责将 GUI 场景下的多模态消息(文本 + 图片)标准化处理后调用指定 LLM 并返回结果;其核心特色包括:

[*]① 多源图片适配,自动将 URL 形式的图片转换为 Base64 编码,兼容 PNG/JPEG 格式,同时支持image_b64格式自动转换为image_url格式,统一 LLM 输入规范;
[*]② 灵活的图片预处理能力,支持按指定尺寸缩放图片,且缩放后统一转换为 JPEG 格式并压缩质量,适配不同 LLM 的图片输入限制;
[*]③ 模型配置解耦,通过读取model_config.yaml文件动态加载不同提供商的 API 地址和密钥,无需硬编码;
[*]④ 完整的调用监控,记录 LLM 推理耗时和请求 ID,便于性能分析和问题排查;
[*]⑤ 兼容推理过程返回,若 LLM 返回reasoning_content则拼接至结果前,保留模型思考过程,提升 GUI-Agent 决策的可解释性;
[*]⑥ 鲁棒的参数处理,为温度、最大令牌数等核心参数设置默认值,避免调用异常;
流程图

流程图核心说明

[*]初始化阶段:优先加载模型配置文件,校验提供商合法性,避免无效 API 调用;
[*]消息预处理核心:统一图片格式:将image_url(URL/Base64)、image_b64 全部转换为标准image_url(Base64 格式),将图片按需缩放至目标尺寸,统一转换为 JPEG 格式压缩,适配 LLM 输入限制;
[*]LLM 交互阶段:记录调用全生命周期耗时,打印关键监控信息(耗时 / 请求 ID),便于问题排查;
[*]结果处理:兼容reasoning_content(模型思考过程),拼接后返回,提升 GUI-Agent 决策的可解释性;

代码

ask_llm_anything 的代码如下:
def ask_llm_anything(model_provider, model_name, messages, args= {
    "max_tokens": 256,
    "temperature": 0.5,
    "top_p": 1.0,
    "frequency_penalty": 0.0,
}, resize_config=None):
    """
    通用LLM调用函数:处理多模态消息(文本+图片),调用指定LLM并返回结果
    :param model_provider: 模型提供商(如openai/self-host等)
    :param model_name: 模型名称(如gpt-4v/llama3-vision等)
    :param messages: 待发送的多模态消息列表
    :param args: LLM调用参数(温度、最大令牌数等)
    :param resize_config: 图片缩放配置(是否缩放、目标尺寸)
    :return: LLM返回的文本结果(含可选的推理过程)
    """
    # 读取模型配置文件:加载不同提供商的API地址和密钥
    with smart_open("model_config.yaml", "r") as f:
      model_config = yaml.safe_load(f)
   
    # 配置当前提供商的API基础地址和密钥
    if model_provider in model_config:
      openai.api_base = model_config["api_base"]
      openai.api_key = model_config["api_key"]
    else:
      # 未知提供商则抛出异常,避免无效调用
      raise ValueError(f"Unknown model provider: {model_provider}")
   
    # 定义内部函数:预处理消息中的图片,统一转换为Base64格式的image_url
    def preprocess_messages(messages):
      # 遍历每条消息
      for msg in messages:
            # 跳过纯文本消息(content为字符串)
            if type(msg['content']) == str:
                continue
            # 断言content为列表(多模态内容格式)
            assert type(msg['content']) == list
            # 遍历消息内的每个内容块(文本/图片)
            for content in msg['content']:
                # 跳过文本内容块
                if content['type'] == "text":
                  continue
                # 处理image_url类型的图片
                if content['type'] == "image_url":
                  url = content['image_url']['url']
                  # 已为Base64格式则跳过
                  if url.startswith("data:image/"):
                        continue
                  else:
                        # 从URL读取图片字节数据
                        image_bytes = smart_open(url, mode="rb").read()
                        # 转换为Base64编码
                        b64 = base64.b64encode(image_bytes).decode('utf-8')
                        # 判断图片格式,设置对应的Base64前缀
                        if image_bytes == b"\x89PNG":
                            content['image_url']['url'] = "data:image/png;base64," + b64
                        elif image_bytes == b"\xff\xd8":
                            content['image_url']['url'] = "data:image/jpeg;base64," + b64
                        else:
                            # 未知格式默认按PNG处理
                            content['image_url']['url'] = "data:image/png;base64," + b64
                else:
                  # 处理image_b64类型,转换为统一的image_url格式
                  assert content['type'] == "image_b64"
                  b64 = content['image_b64']['b64_json']
                  # 删除原image_b64字段
                  del content['image_b64']
                  # 新增image_url字段,统一格式
                  content['image_url'] = {"url": "data:image/png;base64," + b64}
                  content['type'] = "image_url"
               
                # 图片缩放处理:若配置了缩放且开启
                if resize_config is not None and resize_config.get("is_resize", False) == True:
                  # 提取Base64编码部分(去掉前缀)
                  image_url = content['image_url']['url']
                  image_b64_url = image_url.split(",", 1)
                  # 解码Base64为图片字节数据
                  image_data = base64.b64decode(image_b64_url)
                  # 打开图片并缩放至目标尺寸
                  image = Image.open(io.BytesIO(image_data))
                  image = image.resize(size= resize_config['target_image_size'])
                  # 转换为RGB并保存为JPEG格式(压缩质量85)
                  image_data = io.BytesIO()
                  image = image.convert('RGB')
                  image.save(image_data, format="JPEG", quality=85)
                  image_data = image_data.getvalue()
                  # 重新编码为Base64并更新URL
                  b64_image = base64.b64encode(image_data).decode('utf-8')
                  content['image_url']['url'] = f"data:image/jpeg;base64,{b64_image}"

      return messages
   
    # 执行消息预处理:统一图片格式和尺寸
    messages = preprocess_messages(messages)

    # 记录LLM调用开始时间
    start_time = time.time()
    # 调用OpenAI兼容的ChatCompletion接口
    completion = openai.ChatCompletion.create(
      api_key=openai.api_key,          # 传入API密钥
      api_base = openai.api_base,      # 传入API基础地址
      model=model_name,                # 指定模型名称
      messages=messages,               # 预处理后的多模态消息
      temperature=args.get("temperature", 0.5),          # 采样温度(默认0.5)
      top_p=args.get("top_p", 1.0),                      # 核采样(默认1.0)
      frequency_penalty=args.get("frequency_penalty", 0.0),# 重复惩罚(默认0.0)
      max_tokens=args.get("max_tokens", 100),             # 最大生成令牌数(默认100)
    )
    # 记录LLM调用结束时间
    end_time = time.time()
    # 打印推理耗时(便于性能监控)
    print(f"LLM {model_name} inference time: {end_time - start_time:.2f} seconds")   
    # 提取LLM返回的核心结果
    result = completion.choices.message['content']
    # 打印请求ID(便于问题排查)
    print("llm ask id:", completion['id'])

    # 提取可选的推理过程内容(reasoning_content)
    reasoning = completion.choices.message.get("reasoning_content", "")
    # 若有推理过程,拼接至结果前(格式:推理内容\n结果)
    if reasoning is not None and len(reasoning) > 0:
      result = "" + reasoning + "<|FunctionCallEnd|>0x02 Parser0920Summary 的功能

如上一篇所言,GUI-MCP 提供一套跨平台的标准协议,将设备能力抽象为少量原子级和复合级工具。其分层双栈架构把“低层 MCP”与“高层 MCP”结合起来:前者提供细粒度操作(如点击、滑动、文本输入);后者则把整个任务委派给本地部署的 GUI 专家模型(如 Step-GUI-4B)。该设计让主 LLM 只需关注高层规划,而将常规 GUI 操作卸载给本地模型。尤为关键的是,GUI-MCP 支持高隐私执行模式:原始截图与敏感状态始终留在设备端,仅将语义摘要发送给外部 LLM,从而在利用云端推理能力的同时有效保护用户隐私。
因此,图像摘要/文本摘要/任务描述 是一个及其重要的功能。
Parser0920Summary 是一个专用于 GUI Agent 交互的解析器类, 在 GUI Agent 体系中充当关键桥梁,负责把自然语言任务和环境信息转换成模型可理解的消息,并将模型输出解析为可执行的动作。
2.1 核心功能概览

Parser0920Summary 的核心功能如下。

[*]构造提供给 LLM 的提示信息:环境信息 → 消息格式,即 环境信息 + 任务描述 → → 模型输入消息
[*]解析 LLM 返回的动作指令:模型输出 → 结构化动作,即 模型输出字符串 → → 结构化动作 → 设备执行
[*]动作格式标准化
[*]历史交互摘要维护
通过上述双向转换,Parser0920Summary 实现自然语言与设备操作之间的无缝衔接,是 GUI Agent 理解并执行复杂任务的核心组件。
2.2 详细功能说明

构建提示词(make_status_prompt ):该函数是整个 GUI Agent 系统中人机交互的关键环节,确保模型能够获得足够的上下文信息来做出正确的操作决策。

[*]整合任务上下文信息

[*]当前用户任务描述
[*]历史操作摘要(summary_history)
[*]用户评论/指令(user_comment)
[*]当前屏幕截图

[*]格式化提示内容

[*]将所有信息组织成结构化的文本格式
[*]添加必要的引导说明,指导模型如何响应
[*]包含动作空间定义和输出格式要求

[*]处理历史对话

[*]整合之前的问答历史(historical_qa)
[*]特别强调用户的最新指令

环境转消息(env2messages4ask):将任务描述、历史环境和动作转换为可供模型理解的消息格式

[*]整合任务描述与用户指令
[*]添加历史操作摘要
[*]包含当前屏幕截图
[*]处理用户与 Agent 的问答历史
[*]生成符合模型输入要求的多模态消息
字符串到动作解析(str2action):将模型输出的字符串解析为结构化的动作对象

[*]解析
[*]提取动作类型和参数
[*]兼容多种动作格式
[*]标准化坐标点
[*]容错错误格式
动作到字符串转换(action2str):将结构化动作转换为标准化字符串格式

[*]生成含思考过程的完整输出
[*]格式化各类动作参数
[*]添加动作摘要信息
动作标准化(action2action):确保动作对象符合标准格式

[*]根据动作类型(CLICK、TYPE、AWAKE、INFO、WAIT、COMPLETE、ABORT、SLIDE、LONGPRESS)验证特定字段
[*]统一不同动作类型的参数
[*]处理字段命名兼容性问题
2.3 重点函数分析

我们要把几个函数单独拿出来分析。
2.3.1 str2action

str2action 函数负责把模型生成的原始字符串解析成可执行的结构化指令,使系统能够理解并执行模型的决策。
核心步骤

核心步骤如下:

[*]提取思考过程(CoT)

[*]捕获内容;
[*]容错修正标签拼写(如 )

[*]解析动作参数

[*]按制表符分隔提取键值对;
[*]获取 action、explain、point/point1/point2 等关键字段;
[*]处理制表符分隔的键值对格式。

[*]数据类型转换

[*]将坐标字符串 "100,200" 转为整数列表 ;
[*]对其他参数执行相应类型转换。

[*]容错处理

[*]应对模型输出格式不规范;
[*]返回错误信息,便于调试解析失败。

使用场景

在 LocalServer 的 automate_step 方法中调用:
response = ask_llm_anything(...)      # 获取模型响应
action = parser.str2action(response)# 解析为结构化动作输入(模型字符串):
<THINK>我需要点击搜索按钮来查找相关信息</THINK>
explain:点击搜索按钮以开始搜索|taction:CLICK|tpoint:500,320|tsummary:已点击搜索按钮输出(结构化字典):
OrderedDict([
    ("cot", "我需要点击搜索按钮来查找相关信息"),
    ("explain", "点击搜索按钮以开始搜索"),
    ("action", "CLICK"),
    ("point", ),
    ("summary", "已点击搜索按钮")
])代码

    def str2action(self, command_str):
      command_str = command_str.strip()
      
      # Normalize THINK tags: fix typos, case, and spacing
      command_str = (
            command_str
            .replace("<TINK>", "<THINK>").replace("</TINK>", "</THINK>")
            .replace("<think>", "<THINK>").replace("</think>", "</THINK>")
      )
      command_str = re.sub(r"<\s*/?THINK\s*>", lambda m: "<THINK>" if "/" not in m.group() else "</THINK>", command_str, flags=re.IGNORECASE)
      
      # Extract CoT and key-value parts
      # Expected format: <THINK> cot </THINK>\nexplain:xxx\taction:xx\tvalue:xxx\tsummary:xxx
      try:
            cot_part = command_str.split("<THINK>").split("</THINK>").strip()
            kv_part = command_str.split("</THINK>").strip()
      except IndexError:
            print(f" Missing <THINK> tags, treating entire response as kv")
            kv_part = command_str
            cot_part = ""

      action = OrderedDict()
      action['cot'] = cot_part
      
      # Error split by \n, should split by tab separator
      kvs =

      for kv in kvs:
            if ":" not in kv:
                continue

            key = kv.split(":", 1).strip()
            value = kv.split(":", 1).strip()

            if key == "action":
                action['action'] = value
            elif key == "summary":
                action['summary'] = value
            elif "point" in key:
                # Parse point format: "x,y" or "x y"
                try:
                  # Replace comma with space for unified processing
                  coords = value.replace(",", " ").split()
                  if len(coords) < 2:
                        raise ValueError(f"Expected 2 coordinates, got {len(coords)}")
                  
                  x, y = int(coords), int(coords)
                  action =
                  
                except (ValueError, IndexError) as e:
                  raise ValueError(
                        f" Failed to parse point '{value}' for key '{key}': {str(e)}. "
                        f"Expected format: 'x,y' or 'x y' with integer values"
                  ) from e
            else:
                action = value

      return action2.3.2 make_status_prompt

该函数在 Parser0920Summary 类的 env2messages4ask 方法中被调用,用于生成发送给 LLM 的消息内容:
# 在 parser_0920_summary.py 中
conversations = [
    {
      "type": "text",
      "text": task_define_prompt          # 任务定义和动作空间说明
      + make_status_prompt(            # 当前状态信息
            task,
            current_env['image'],
            hints,
            summary_history,
            qa_prompt
      )
    }
]函数返回一个包含以下元素的列表:

[*]任务描述和历史操作:用户任务与已完成操作的文本描述
[*]当前截图:以图像 URL 形式提供的当前屏幕截图
[*]操作指导:指导模型如何思考和回应的说明文本
代码如下:
def make_status_prompt(task, current_image, hints, summary_history="", user_comment=""):

    if len(hints) == 0:
      hint_str = ""
    else:
      hint_str = "\n".join()
      hint_str = f"### HINT:\n{hint_str}\n"

    if user_comment == "":
      history_display = summary_history if summary_history.strip() else "暂无历史操作"
    else:
      history_display = summary_history + user_comment if summary_history.strip() else "暂无历史操作"

    user_instruction = f'''\n\n{user_comment}\n\n''' if user_comment != "" else ""
    task = task + user_instruction + "指令结束\n\n"
   
    status_conversation = [
      {
            "type": "text",
            "text": f'''
已知用户指令为:{task}
已知已经执行过的历史动作如下:{history_display}
当前手机屏幕截图如下:
'''
      },
      {
            "type": "image_url",
            "image_url": {"url": current_image}
      },
      {
            "type": "text",
            "text": f'''
在执行操作之前,请务必回顾你的历史操作记录和限定的动作空间,先进行思考和解释然后输出动作空间和对应的参数:
1. 思考(THINK):在 <THINK> 和 </THINK> 标签之间。
2. 解释(explain):在动作格式中,使用 explain: 开头,简要说明当前动作的目的和执行方式。
在执行完操作后,请输出执行完当前步骤后的新历史总结。
输出格式示例:
<THINK> 思考的内容 </THINK>
explain:解释的内容\taction:动作空间和对应的参数\tsummary:执行完当前步骤后的新历史总结
'''
      }
    ]

    return status_conversation2.4 THINK 标签

既然上面代码中提到了THINK 标签,我们本小节就进行分析。
定义和结构
在 Parser0920Summary 类中定义了 THINK 标签的使用方式

[*]THINK 标签用于封装 Agent 的思维过程,格式为思考的内容
[*]在 make_status_prompt 函数中明确要求 Agent 在执行操作前进行思考
解析
在 str2action 方法中解析包含 THINK 标签的响应

[*]使用正则表达式提取 THINK 标签中的内容作为COT(Chain of Thought)字段
[*]str2action 方法中包含错误处理逻辑,处理标签格式不规范的情况
在 LLM 交互中的应用

[*]在 ask_llm_v2.py 中,当 LLM 响应包含 THINK 标签时,会提取 `` 之后的内容
[*] ask_llm_anything 函数会处理 reasoning_content 字段,将其包装在 THINK 标签中
在 Agent 决策中的作用

[*]在 gui_agent_loop 中,每个步骤的 action 都包含 cot 字段(即 THINK 标签内容)
[*]COT内容在日志中输出,便于调试和理解 Agent 的决策过程
在 INFO 动作中的应用

[*]当 Agent 执行 INFO 动作时,auto_reply 函数会分析 THINK 标签内容来理解 Agent 的问题
[*]auto_reply 使用 THINK 标签中的内容作为上下文,生成合适的回复
2.5 COT 输出与执行流程

GUI-Agent 不是将 COT 输出作为具体执行流程。COT 输出主要作用为:

[*]COT 不直接参与设备操作的执行,实际执行的动作由 action 字段定义(如 CLICK、TYPE 等)
[*]COT 记录并展示 Agent 的思考过程,让用户了解 Agent 如何分析界面元素和决策点击位置
[*]COT 可以协助任务规划

[*]Agent 需要分析当前屏幕截图和任务,COT 可以帮助 Agent 系统化地分析视觉信息和任务需求
[*]Agent 需要将复杂任务分解为一系列动作,COT 可以帮助 Agent 规划执行路径
[*]Agent 需要将视觉信息转换为具体操作,COT 可以帮助 Agent 理解界面元素与目标之间的关系,有助于精确定位点击坐标(x,y 为 0-1000 范围)
[*]有助于处理复杂的界面布局和交互场景

[*]COT 提供决策的可解释性,开源辅助调试和分析

[*]当 Agent 做出错误决策时,可以通过 THINK 内容追踪错误原因
[*]便于调试和改进 Agent 的决策逻辑

[*]在需要时辅助生成对用户问题的回复,即在 INFO 动作中,COT 信息可用于生成对用户的回复
具体如下:

[*]COT 是在标签中生成的,用于解释 Agent 的决策逻辑
[*]在 str2action 方法中,COT 从 LLM 响应中提取并存储为 COT 字段
[*]在 execute_task 的返回结果中,中间日志会包含 COT信息
[*]intermediate_logs 中包含每个步骤的 COT 字段
[*]在最终结果中,final_action 也包含 COT 信息
2.6 支持的动作类型

动作空间定义

Parser0920Summary 支持完整的 Android GUI 操作集合,例如:

[*]CLICK:点击屏幕坐标
[*]OTYPE:输入文本
[*]COMPLETE:标记任务完成
[*]WAIT:等待指定时间
[*]AWAKE:唤醒指定应用
[*]INFO:向用户询问信息
[*]ABORT:终止任务
[*]SLIDE:滑动操作
[*]LONGPRESS:长按操作
提示词引导

在 task_define_prompt 中明确定义了可用的动作空间和参数格式。
task_define_prompt = """你是一个手机 GUI-Agent 操作专家,你需要根据用户下发的任务、手机屏幕截图和交互操作的历史记录,借助既定的动作空间与手机进行交互,从而完成用户的任务。
请牢记,手机屏幕坐标系以左上角为原点,x轴向右,y轴向下,取值范围均为 0-1000。

# 行动原则:
1. 你需要明确记录自己上一次的action,如果是滑动,不能超过5次。
2. 你需要严格遵循用户的指令,如果你和用户进行过对话,需要更遵守最后一轮的指令

# Action Space:
在 Android 手机的场景下,你的动作空间包含以下9类操作,所有输出都必须遵守对应的参数要求:
1. CLICK:点击手机屏幕坐标,需包含点击的坐标位置 point。
例如:action:CLICK\tpoint:x,y
2. TYPE:在手机输入框中输入文字,需包含输入内容 value、输入框的位置 point。
例如:action:TYPE\tvalue:输入内容\tpoint:x,y
3. COMPLETE:任务完成后向用户报告结果,需包含报告的内容 value。
例如:action:COMPLETE\treturn:完成任务后向用户报告的内容
4. WAIT:等待指定时长,需包含等待时间 value(秒)。
例如:action:WAIT\tvalue:等待时间
5. AWAKE:唤醒指定应用,需包含唤醒的应用名称 value。
例如:action:AWAKE\tvalue:应用名称
6. INFO:询问用户问题或详细信息,需包含提问内容 value。
例如:action:INFO\tvalue:提问内容
7. ABORT:终止当前任务,仅在当前任务无法继续执行时使用,需包含 value 说明原因。
例如:action:ABORT\tvalue:终止任务的原因
8. SLIDE:在手机屏幕上滑动,滑动的方向不限,需包含起点 point1 和终点 point2。
例如:action:SLIDE\tpoint1:x1,y1\tpoint2:x2,y2
9. LONGPRESS:长按手机屏幕坐标,需包含长按的坐标位置 point。
例如:action:LONGPRESS\tpoint:x,y
"""动作解析


[*]通过 str2action 方法解析模型输出的动作
[*]该方法会从模型响应中提取 action 类型、参数等信息
[*]解析时会验证动作类型是否在预定义的动作空间内
动作验证


[*]使用 action_assertion 函数验证动作格式
[*]验证每个动作是否包含必需的字段,如:

[*]CLICK 动作必须包含 point 参数
[*]TYPE 动作必须包含 value 参数
[*]SLIDE 动作必须包含 point1 和 point2 参数

实际执行流程

实际生成动作的流程如下:

[*]LocalServer 的 automate_step 方法接收 observation 并生成动作
[*]gui_agent_loop 调用 automate_step 获取动作
[*]动作通过 uiTars_to_frontend_action 转换为前端动作
[*]act_on_device 执行具体的设备操作
0x03 交互

LocalServer.automate_step 与 Parsero920Summary 之间的逻辑关系如下:

[*]LocalServer.automate_step 负责协调整个流程,从日志读取历史环境和动作
[*]通过 get_parser 获取 Parsero920Summary 实例
[*]调用 env2messages4ask 构建模型输入消息
Parsero920Summary 内部函数协作

[*]env2messages4ask 接收任务、环境和动作参数,整合历史动作和当前状态
[*]通过 make_status_prompt 构建状态提示,包含任务描述、历史动作和当前截图
[*]str2action 解析模型输出,action2action 验证动作格式
模型交互流程

[*]env2messages4ask 返回构建好的消息格式
[*]LocalServer 通过 ask_llm_anything 调用模型
[*]模型返回字符串后,str2action 解析成动作对象
[*]处理
[*]LocalServer 将动作和当前步骤信息记录到日志
数据流向

[*]环境数据从 LocalServer 传递给 Parsero920Summary
[*]Parsero920Summary 构建消息并返回给 LocalServer
[*]LocalServer 调用模型并接收响应
[*]Parsero920Summary 解析模型响应并返回动作给 LocalServer
[*]LocalServer 记录结果并返回给调用者
Parser0920Summary 与 LocalServer 交互流程图如下:

0x04 模型分发

论文中提到:GUI-MCP 提供一套标准化、跨平台的协议,将设备能力抽象为少量原子及组合工具。其分层双栈架构结合:“低层 MCP”提供细粒度操作(点击、滑动、文本输入等),与“高层 MCP”将整个任务委派给本地部署的 GUI 专有模型(如 Step-GUI-4B)。该设计使主语言模型专注于高层规划,同时将常规 GUI 操作卸载至本地模型。尤为关键的是,GUI-MCP 支持高隐私执行模式:原始截图与敏感状态留在设备端,仅语义摘要流向外部语言模型,从而在利用云端推理能力的同时有效保护用户隐私。
本小节就来看看如何进行模型分发(没有实际串联的调用代码,只是反推)。
4.1 模型分发机制


[*]统一接口:ask_llm_anything 函数抽象了不同模型提供商的差异
[*]配置驱动:通过 model_config.yaml 文件配置不同模型提供商
[*]灵活切换:在任务配置中指定使用哪个模型提供商

4.2 本地模型流程

模型配置

[*]run_single_task.py 中的 local_model_config 定义了模型配置
[*]配置中指定 model_provider: "local" 表示使用本地模型
[*]model_config.yaml 文件中定义了本地模型的 API 基础地址和密钥
模型推理流程

[*]evaluate_task_on_device 函数启动任务评估
[*]LocalServer 类作为本地服务器
[*]automate_step 方法处理单步推理
[*]get_parser 获取解析器(Parser0920Summary)
[*]env2messages4ask 生成模型输入消息
[*]ask_llm_anything 函数调用模型 API
本地模型调用链

[*]ask_llm_anything 从 model_config.yaml 读取配置
[*]设置 openai.api_base 和 openai.api_key
[*]通过 ChatCompletion.create 发起请求
4.3 云端模型流程

云端模型配置

[*]model_config.yaml 文件可配置多个模型提供商
[*]例如 openai、anthropic 等云端服务提供商
[*]配置相应的 API 基础地址和密钥
云端模型调用

[*]调用流程与本地模型相同,通过 ask_llm_anything 函数
[*]根据 model_provider 参数切换不同模型提供商
[*]通过网络请求访问云端模型服务
0x05 过渡到执行层

下面流程展示了完整的从输入到最终执行操作的流程:

[*]make_status_prompt 整合任务、截图、历史、QA → 多模态消息
[*]LLM 接收消息 → 返回自然语言动作字符串
[*]LLM 输出:模型生成包含思考过程和动作指令的文本响应
[*]解析响应:使用 str2action 函数解析文本,提取结构化信息
[*]动作验证:验证解析出的动作是否符合预定义的动作类型和参数要求
[*]前端动作转换:将标准化动作转换为设备可执行的前端动作格式
[*]设备执行:通过 ADB 命令在 Android 设备上执行具体操作 (点击、滑动、输入等)
[*]循环回到步骤 1,直至任务完成或达到最大步数。
具体参见下图。

0xFF 参考

从豆包手机谈起:端侧智能的愿景与路线图

来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
页: [1]
查看完整版本: 【GUI-Agent】阶跃星辰 GUI-MCP 解读---(2)---决策层