323 lines
10 KiB
Python
323 lines
10 KiB
Python
"""
|
||
AI Agent - 智能意图解析与自动化执行
|
||
类似豆包手机的AI自动化功能
|
||
|
||
功能：
|
||
1. 接收语音/文本命令
|
||
2. 调用大模型解析意图
|
||
3. 生成自动化操作序列
|
||
4. 通过ADB/uiautomator2执行
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import httpx
|
||
import asyncio
|
||
import logging
|
||
from typing import Dict, List, Optional, Any
|
||
from dataclasses import dataclass
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class UIAction:
|
||
"""UI操作动作"""
|
||
action: str # click, swipe, input_text, open_app, back, home, etc.
|
||
params: Dict[str, Any]
|
||
description: str = ""
|
||
|
||
|
||
class AIAgent:
|
||
"""
|
||
AI自动化代理
|
||
|
||
使用大模型解析用户意图，生成操作序列
|
||
"""
|
||
|
||
def __init__(self):
|
||
# v0 API配置
|
||
self.api_url = os.getenv("AI_API_URL", "https://api.v0.dev/v1")
|
||
self.api_key = os.getenv("AI_API_KEY", "")
|
||
self.model = os.getenv("AI_MODEL", "claude-opus")
|
||
|
||
# 系统提示词
|
||
self.system_prompt = """你是一个手机自动化助手，负责将用户的语音指令转换为具体的手机操作序列。
|
||
|
||
你需要返回一个JSON数组，包含要执行的操作。每个操作格式如下：
|
||
|
||
可用的操作类型：
|
||
1. open_app: 打开应用
|
||
{"action": "open_app", "params": {"package": "com.tencent.mm"}, "description": "打开微信"}
|
||
|
||
2. click: 点击屏幕坐标
|
||
{"action": "click", "params": {"x": 540, "y": 1200}, "description": "点击位置"}
|
||
|
||
3. swipe: 滑动屏幕
|
||
{"action": "swipe", "params": {"x1": 540, "y1": 1500, "x2": 540, "y2": 500, "duration": 300}, "description": "向上滑动"}
|
||
|
||
4. input_text: 输入文字
|
||
{"action": "input_text", "params": {"text": "你好"}, "description": "输入文字"}
|
||
|
||
5. key_event: 按键事件
|
||
{"action": "key_event", "params": {"keycode": "KEYCODE_BACK"}, "description": "返回"}
|
||
|
||
6. back: 返回
|
||
{"action": "back", "params": {}, "description": "点击返回"}
|
||
|
||
7. home: 回到桌面
|
||
{"action": "home", "params": {}, "description": "回到桌面"}
|
||
|
||
8. wait: 等待
|
||
{"action": "wait", "params": {"seconds": 2}, "description": "等待2秒"}
|
||
|
||
9. screenshot: 截图
|
||
{"action": "screenshot", "params": {"path": "/sdcard/screenshot.png"}, "description": "截图"}
|
||
|
||
常用应用包名：
|
||
- 微信: com.tencent.mm
|
||
- 抖音: com.ss.android.ugc.aweme
|
||
- 支付宝: com.eg.android.AlipayGphone
|
||
- 淘宝: com.taobao.taobao
|
||
- 微博: com.sina.weibo
|
||
- QQ: com.tencent.mobileqq
|
||
- 设置: com.android.settings
|
||
- 相机: com.android.camera
|
||
- 浏览器: com.android.chrome
|
||
|
||
只返回JSON数组，不要其他解释。如果无法理解指令，返回空数组 []。
|
||
"""
|
||
|
||
async def parse_command(self, text: str, context: Optional[Dict] = None) -> List[UIAction]:
|
||
"""
|
||
解析用户命令，返回操作序列
|
||
|
||
Args:
|
||
text: 用户的语音/文本命令
|
||
context: 当前上下文（当前APP、UI状态等）
|
||
|
||
Returns:
|
||
操作序列列表
|
||
"""
|
||
try:
|
||
# 构建消息
|
||
messages = [
|
||
{"role": "system", "content": self.system_prompt},
|
||
{"role": "user", "content": f"用户指令: {text}"}
|
||
]
|
||
|
||
# 如果有上下文，添加到消息中
|
||
if context:
|
||
context_str = json.dumps(context, ensure_ascii=False)
|
||
messages.append({"role": "user", "content": f"当前上下文: {context_str}"})
|
||
|
||
# 调用AI API
|
||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||
response = await client.post(
|
||
f"{self.api_url}/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": self.model,
|
||
"messages": messages,
|
||
"temperature": 0.3,
|
||
"max_tokens": 1000
|
||
}
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
logger.error(f"AI API错误: {response.status_code} - {response.text}")
|
||
return self._fallback_parse(text)
|
||
|
||
data = response.json()
|
||
content = data["choices"][0]["message"]["content"]
|
||
|
||
# 解析JSON
|
||
return self._parse_json_actions(content)
|
||
|
||
except Exception as e:
|
||
logger.error(f"AI解析失败: {e}")
|
||
return self._fallback_parse(text)
|
||
|
||
def _parse_json_actions(self, content: str) -> List[UIAction]:
|
||
"""解析JSON格式的操作序列"""
|
||
try:
|
||
# 尝试提取JSON
|
||
content = content.strip()
|
||
if content.startswith("```json"):
|
||
content = content[7:]
|
||
if content.startswith("```"):
|
||
content = content[3:]
|
||
if content.endswith("```"):
|
||
content = content[:-3]
|
||
content = content.strip()
|
||
|
||
actions_data = json.loads(content)
|
||
|
||
if not isinstance(actions_data, list):
|
||
actions_data = [actions_data]
|
||
|
||
actions = []
|
||
for item in actions_data:
|
||
action = UIAction(
|
||
action=item.get("action", ""),
|
||
params=item.get("params", {}),
|
||
description=item.get("description", "")
|
||
)
|
||
actions.append(action)
|
||
|
||
return actions
|
||
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"JSON解析失败: {e}, 内容: {content}")
|
||
return []
|
||
|
||
def _fallback_parse(self, text: str) -> List[UIAction]:
|
||
"""
|
||
降级解析 - 使用规则匹配
|
||
当AI不可用时使用
|
||
"""
|
||
text = text.lower()
|
||
actions = []
|
||
|
||
# 打开应用
|
||
app_keywords = {
|
||
"微信": "com.tencent.mm",
|
||
"抖音": "com.ss.android.ugc.aweme",
|
||
"支付宝": "com.eg.android.AlipayGphone",
|
||
"淘宝": "com.taobao.taobao",
|
||
"qq": "com.tencent.mobileqq",
|
||
"微博": "com.sina.weibo",
|
||
"设置": "com.android.settings",
|
||
"相机": "com.android.camera",
|
||
"浏览器": "com.android.chrome",
|
||
}
|
||
|
||
for keyword, package in app_keywords.items():
|
||
if f"打开{keyword}" in text or f"启动{keyword}" in text:
|
||
actions.append(UIAction(
|
||
action="open_app",
|
||
params={"package": package},
|
||
description=f"打开{keyword}"
|
||
))
|
||
return actions
|
||
|
||
# 返回/回去
|
||
if "返回" in text or "回去" in text or "退出" in text:
|
||
actions.append(UIAction(
|
||
action="back",
|
||
params={},
|
||
description="点击返回"
|
||
))
|
||
return actions
|
||
|
||
# 回到桌面
|
||
if "桌面" in text or "主页" in text or "home" in text:
|
||
actions.append(UIAction(
|
||
action="home",
|
||
params={},
|
||
description="回到桌面"
|
||
))
|
||
return actions
|
||
|
||
# 截图
|
||
if "截图" in text or "截屏" in text:
|
||
actions.append(UIAction(
|
||
action="screenshot",
|
||
params={"path": "/sdcard/screenshot.png"},
|
||
description="截图"
|
||
))
|
||
return actions
|
||
|
||
# 向上/下滑动
|
||
if "向上滑" in text or "上滑" in text:
|
||
actions.append(UIAction(
|
||
action="swipe",
|
||
params={"x1": 540, "y1": 1500, "x2": 540, "y2": 500, "duration": 300},
|
||
description="向上滑动"
|
||
))
|
||
return actions
|
||
|
||
if "向下滑" in text or "下滑" in text:
|
||
actions.append(UIAction(
|
||
action="swipe",
|
||
params={"x1": 540, "y1": 500, "x2": 540, "y2": 1500, "duration": 300},
|
||
description="向下滑动"
|
||
))
|
||
return actions
|
||
|
||
import re
|
||
m = re.search(r'(?:给|发给|跟|对|向)(.+?)(?:发消息|发送|说)[:：]?(.+)', text)
|
||
if not m:
|
||
m = re.search(r'(?:发消息给|发送给)(.+?)[:：](.+)', text)
|
||
if m:
|
||
to_id = m.group(1).strip()
|
||
content = m.group(2).strip()
|
||
actions.append(UIAction(
|
||
action="send_wechat_message",
|
||
params={"to_id": to_id, "content": content, "platform": "wechat"},
|
||
description=f"给{to_id}发微信消息: {content}"
|
||
))
|
||
return actions
|
||
|
||
m2 = re.search(r'(?:发消息给|发送消息给|发给)(.+)', text)
|
||
if m2:
|
||
to_id = m2.group(1).strip()
|
||
actions.append(UIAction(
|
||
action="send_wechat_message",
|
||
params={"to_id": to_id, "content": "", "platform": "wechat"},
|
||
description=f"给{to_id}发微信消息（需补充内容）"
|
||
))
|
||
return actions
|
||
|
||
logger.warning(f"无法解析命令: {text}")
|
||
return actions
|
||
|
||
|
||
# 全局实例
|
||
ai_agent = AIAgent()
|
||
|
||
|
||
async def process_voice_command(text: str, device_id: str = None) -> Dict:
|
||
"""
|
||
处理语音命令
|
||
|
||
Args:
|
||
text: 语音识别的文本
|
||
device_id: 设备ID
|
||
|
||
Returns:
|
||
处理结果
|
||
"""
|
||
logger.info(f"处理语音命令: {text} (设备: {device_id})")
|
||
|
||
# 解析命令
|
||
actions = await ai_agent.parse_command(text)
|
||
|
||
if not actions:
|
||
return {
|
||
"success": False,
|
||
"message": "无法理解您的指令",
|
||
"text": text,
|
||
"actions": []
|
||
}
|
||
|
||
# 转换为可序列化格式
|
||
actions_data = [
|
||
{
|
||
"action": a.action,
|
||
"params": a.params,
|
||
"description": a.description
|
||
}
|
||
for a in actions
|
||
]
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"已解析 {len(actions)} 个操作",
|
||
"text": text,
|
||
"actions": actions_data
|
||
}
|