基于Qwen-VL的手机智能体开辟

先上Demo：
     vl_agent_demo
  
代码如下：
0 设置工作目次：
    你的工作目次需要如下：
   此中utils文件夹和qwenvl_agent.py均参考自
  GitHub - QwenLM/Qwen2.5-VL: Qwen2.5-VL is the multimodal large language model series developed by Qwen team, Alibaba Cloud.Qwen2.5-VL is the multimodal large language model series developed by Qwen team, Alibaba Cloud. - QwenLM/Qwen2.5-VL

https://github.com/QwenLM/Qwen2.5-VL
   YourProj(文件夹)：
                  utils(文件夹)
                        agent_function_call.py
                  mobile_agent.py
                  qwenvl_agent.py
  （1）运行代码mobile_agent.py：

1. import os
2. import time
3. import json
4. from ppadb.client import Client as AdbClient
5. import uiautomator2 as u2
6. import base64
7. from qwenvl_agent import perform_gui_grounding_with_api
10. class Android_VL_Agent:
12.     def __init__(self):
13.         self.client = AdbClient(host="127.0.0.1", port=5037)
14.         self.device_serial = None
15.         self.u2_device = None
16.         self.SCREENSHOT_PATH = None
17.         self.QWEN_MODEL_ID = 'qwen2.5-vl-7b-instruct'
18.         self.__set_up()
20.     @staticmethod
21.     def check_adb_service():
22.         try:
23.             result = os.popen("adb devices").read()
24.             if "List of devices attached" in result:
25.                 return True
26.             else:
27.                 os.system("adb start-server")
28.                 time.sleep(5)  # 等待 ADB 服务启动
29.                 result = os.popen("adb devices").read()
30.                 if "List of devices attached" in result:
31.                     return True
32.                 else:
33.                     return False
34.         except Exception:
35.             print("ADB服务启动失败")
36.             return False
37.     @staticmethod
38.     def encode_image(image_path):
39.         with open(image_path, "rb") as image_file:
40.             return base64.b64encode(image_file.read()).decode("utf-8")
42.     @staticmethod
43.     def info_parser(info):
44.         try:
45.             body = info.split("<tool_call>")[1].split("</tool_call>")[0]
46.             return json.loads(body)
47.         except Exception as e:
48.             print(f"解析失败: {str(e)}")
49.             return None
51.     # 启动
52.     def __set_up(self):
53.         assert self.check_adb_service()
54.         devices = self.client.devices()
55.         self.device_serial = devices[0].serial if devices else None
56.         self.u2_device = u2.connect(self.device_serial)
57.         self.SCREENSHOT_PATH = "screenshot.png"
59.     # 定义单点事件
60.     def __single_point_event(self,x,y):
61.         try:
62.             self.u2_device.click(x, y)
63.             return True
64.         except Exception as e:
65.             print(f"单点失败: {str(e)}")
66.             return False
68.     # 定义输入内容
69.     def __input_content(self,content):
70.         try:
71.             self.u2_device.send_keys(content)
72.             return True
73.         except Exception as e:
74.             print(f"输入失败: {str(e)}")
75.             return False
77.     # 截图并保存
78.     def __screenshot(self):
79.         try:
80.             # 清除之前的截图
81.             if os.path.exists(self.SCREENSHOT_PATH):
82.                 os.remove(self.SCREENSHOT_PATH)
83.             screenshot = self.u2_device.screenshot()
84.             screenshot.save(self.SCREENSHOT_PATH)
85.             # screenshot.show()
86.             return True
87.         except Exception as e:
88.             print(f"截图失败: {str(e)}")
89.             return False
91.     def __Qwen_vl_agent(self, query):
92.         output_info = perform_gui_grounding_with_api(self.SCREENSHOT_PATH, query, self.QWEN_MODEL_ID)
93.         # print(output_info)
94.         result = self.info_parser(str(output_info))["arguments"]
95.         return result
97.     def __action(self,result):
98.         if "click" in result["action"]:
99.             coordinate = result["coordinate"]
100.             self.__single_point_event(coordinate[0],coordinate[1])
101.         elif "type" in result["action"]:
102.             self.__input_content(result["text"])
104.     def run(self,query):
105.         # 重新连接
106.         self.u2_device = u2.connect(self.device_serial)
107.         # 感知
108.         self.__screenshot()
109.         # 理解
110.         result = self.__Qwen_vl_agent(query)
111.         print(result)
112.         # 执行
113.         self.__action(result)
115.     def __call__(self,query):
116.         self.run(query)
118. if __name__ == "__main__":
119.     agent = Android_VL_Agent()
120.     # timestep
121.     timestep = 2
122.     name = "名字"
123.     message = "信息"
125.     agent.run("打开微信")
126.     time.sleep(timestep)
127.     agent.run(f"点击和{name}聊天框的的顶部区域进入聊天界面")
128.     time.sleep(timestep)
129.     agent.run("点击屏幕底部的输入框部分进入输入界面")
130.     time.sleep(timestep)
131.     agent.run(f"在聊天框输入内容：{message}")
132.     time.sleep(timestep)
133.     agent.run("点击右侧发送按钮中心位置发送消息")

复制代码

（2）方法代码qwenvl_agent.py

1. import json
2. import base64
3. from openai import OpenAI
4. from qwen_agent.llm.fncall_prompts.nous_fncall_prompt import (
5.     NousFnCallPrompt,
6.     Message,
7.     ContentItem,
8. )
9. from PIL import Image, ImageDraw, ImageColor
10. from transformers.models.qwen2_vl.image_processing_qwen2_vl_fast import smart_resize
11. import warnings
12. warnings.filterwarnings("ignore")
13. from utils.agent_function_call import ComputerUse
15. def draw_point(image: Image.Image, point: list, color=None):
16.     if isinstance(color, str):
17.         try:
18.             color = ImageColor.getrgb(color)
19.             color = color + (128,)
20.         except ValueError:
21.             color = (255, 0, 0, 128)
22.     else:
23.         color = (255, 0, 0, 128)
25.     overlay = Image.new('RGBA', image.size, (255, 255, 255, 0))
26.     overlay_draw = ImageDraw.Draw(overlay)
27.     radius = min(image.size) * 0.05
28.     x, y = point
30.     overlay_draw.ellipse(
31.         [(x - radius, y - radius), (x + radius, y + radius)],
32.         fill=color)
34.     center_radius = radius * 0.1
35.     overlay_draw.ellipse(
36.         [(x - center_radius, y - center_radius),
37.          (x + center_radius, y + center_radius)],
38.         fill=(0, 255, 0, 255))
40.     image = image.convert('RGBA')
41.     combined = Image.alpha_composite(image, overlay)
42.     return combined.convert('RGB')
44. def encode_image(image_path):
45.     with open(image_path, "rb") as image_file:
46.         return base64.b64encode(image_file.read()).decode("utf-8")
48. def perform_gui_grounding_with_api(screenshot_path, user_query, model_id, min_pixels=3136, max_pixels=12845056):
49.     """
50.     Perform GUI grounding using Qwen model to interpret user query on a screenshot.
52.     Args:
53.         screenshot_path (str): Path to the screenshot image
54.         user_query (str): User's query/instruction
55.         model: Preloaded Qwen model
56.         min_pixels: Minimum pixels for the image
57.         max_pixels: Maximum pixels for the image
59.     Returns:
60.         tuple: (output_text, display_image) - Model's output text and annotated image
61.     """
63.     # Open and process image
64.     input_image = Image.open(screenshot_path)
65.     base64_image = encode_image(screenshot_path)
66.     client = OpenAI(
67.         # If the environment variable is not configured, please replace the following line with the Dashscope API Key: api_key="sk-xxx". Access via https://bailian.console.alibabacloud.com/?apiKey=1 "
68.         api_key="xxx",
69.         base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
70.     )
71.     resized_height, resized_width = smart_resize(
72.         input_image.height,
73.         input_image.width,
74.         min_pixels=min_pixels,
75.         max_pixels=max_pixels,
76.     )
78.     # Initialize computer use function
79.     computer_use = ComputerUse(
80.         cfg={"display_width_px": resized_width, "display_height_px": resized_height}
81.     )
83.     # Build messages
84.     system_message = NousFnCallPrompt.preprocess_fncall_messages(
85.         messages=[
86.             Message(role="system", content=[ContentItem(text="You are a helpful assistant.")]),
87.         ],
88.         functions=[computer_use.function],
89.         lang=None,
90.     )
91.     system_message = system_message[0].model_dump()
92.     messages = [
93.         {
94.             "role": "system",
95.             "content": [
96.                 {"type": "text", "text": msg["text"]} for msg in system_message["content"]
97.             ],
98.         },
99.         {
100.             "role": "user",
101.             "content": [
102.                 {
103.                     "type": "image_url",
104.                     "min_pixels": min_pixels,
105.                     "max_pixels": max_pixels,
106.                     # Pass in BASE64 image data. Note that the image format (i.e., image/{format}) must match the Content Type in the list of supported images. "f" is the method for string formatting.
107.                     # PNG image:  f"data:image/png;base64,{base64_image}"
108.                     # JPEG image: f"data:image/jpeg;base64,{base64_image}"
109.                     # WEBP image: f"data:image/webp;base64,{base64_image}"
110.                     "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"},
111.                 },
112.                 {"type": "text", "text": user_query},
113.             ],
114.         }
115.     ]
116.     # print(json.dumps(messages, indent=4))
117.     completion = client.chat.completions.create(
118.         model=model_id,
119.         messages=messages,
120.     )
121.     output_text = completion.choices[0].message.content
123.     # Parse action and visualize
124.     # print(output_text)
125.     action = json.loads(output_text.split('<tool_call>\n')[1].split('\n</tool_call>')[0])
126.     # display_image = input_image.resize((resized_width, resized_height))
127.     # display_image = draw_point(input_image, action['arguments']['coordinate'], color='green')
129.     return output_text
132. if __name__ == "__main__":
133.     screenshot = "screenshot.png"
134.     user_query = '在聊天框输入内容：下午好！'
135.     model_id = "qwen2.5-vl-7b-instruct"
136.     output_text = perform_gui_grounding_with_api(screenshot, user_query, model_id)
137.     print(output_text)

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。