Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions examples/batch_results.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
Phone Agent Batch Task Execution Results
============================================================
Generated: 2025-12-14 12:05:05

Task 1: 打开小红书搜索美食攻略
Status: SUCCESS
Duration: 24.70 seconds
Result: 任务完成!小红书已成功打开并搜索"美食攻略"。页面已显示相关搜索结果,包括各种美食攻略内容,如桂林餐厅推荐、阳朔美食记录等。您可以浏览这些美食攻略内容了。
----------------------------------------
Task 2: 打开高德地图查看公交
Status: SUCCESS
Duration: 24.99 seconds
Result: 已成功打开高德地图并进入公交查看页面。现在可以看到:

1. **公交功能选项**:包括查公交、地铁图、实时公交、观光巴士、运动主页等功能
2. **周边的公交地铁**:显示了13路公交的路线信息,包括开往电子科大秀山校区和机电大楼的路线
3. **路线规划**:可以设置起点和终点来查询具体的公交线路
4. **公共交通模式**:可以切换到公共交通模式查看更多公交信息

您可以点击"查公交"来查询当前位置到目的地的公交线路,或者直接在"我的位置"和"输入终点"中设置起点终点来规划路线。
----------------------------------------
Task 3: 打开美团搜索附近的咖啡店
Status: SUCCESS
Duration: 135.38 seconds
Result: 任务完成!我已经成功在美团上搜索到附近的咖啡店。搜索结果显示了多家咖啡店,包括:

1. **库迪咖啡(桂电花江校区店)** - 评分4.3,距离1.2km
2. **瑞幸咖啡(桂林花江荟谷店)** - 评分4.2,距离1.9km
3. **瑞幸咖啡(桂林电子科技大学店)** - 评分4.2,人均¥25

每家咖啡店都显示了评分、距离、评价数以及各种优惠信息。您可以点击任意一家查看详细信息,选择"到店取"或查看外卖选项。
----------------------------------------
Task 4: 打开bilibili搜索Python教程
Status: SUCCESS
Duration: 60.61 seconds
Result: 任务完成!我已经成功打开bilibili并搜索了"Python教程"。搜索结果已经显示出来,包括:

1. UP主"Python教程_"(2.7万粉丝,129个视频)
2. 多个Python教程视频,如:
- 【白嫖】Python爬取腾讯视频付费视频
- 【激活码】PyCharm下载安装激活+Python
- 【2024版】Python下载安装教程
- 【直播特训】0基础入门Python数据分析
- 尚硅谷python教程、Python零基础入门

您可以点击任何感兴趣的视频开始观看学习。页面顶部还有"入门零基础"、"黑马程序员"、"入门到精通"等筛选选项,方便您进一步筛选教程内容。
----------------------------------------
Task 5: 打开微信,查看最近一条消息
Status: SUCCESS
Duration: 47.67 seconds
Result: 任务完成!我已经成功打开微信并查看了最近一条消息。

最近一条消息来自"文件传输助手",发送于昨天 15:15,内容是:
```
https://docs.google.com/sheets/d/1cchzP9dITZ3WXQTFYNhh3avx oLipqHN75v1Tb86uhHo/edit?pli=1&gid=0
```

这是一条分享 Google Docs 文档的链接消息。
----------------------------------------
248 changes: 248 additions & 0 deletions examples/batch_task_execution.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""
Batch Task Execution Example / 批量任务执行示例

Demonstrates how to use Phone Agent to execute multiple tasks in batch, including:
- Task list definition
- Sequential execution
- Result tracking and reporting
- Error handling and retry mechanism
- History management for batch tasks
演示如何使用 Phone Agent 批量执行多个任务,包括:
- 任务列表定义
- 顺序执行
- 结果跟踪和报告
- 错误处理和重试机制
"""

import os
import sys
import time
from typing import Dict, List, Tuple

# Add the project root to the Python path
# 将项目根目录添加到 Python 路径中
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from phone_agent import PhoneAgent
from phone_agent.agent import AgentConfig
from phone_agent.config import get_messages
from phone_agent.model import ModelConfig


class BatchTaskResult:
"""Batch task result structure / 批量任务结果结构"""

def __init__(self, task: str):
self.task = task
self.success = False
self.result = ""
self.start_time = 0
self.end_time = 0
self.duration = 0
self.error = None
self.retry_count = 0

def start(self):
"""Mark task as started / 标记任务开始"""
self.start_time = time.time()

def complete(self, result: str):
"""Mark task as completed / 标记任务完成"""
self.end_time = time.time()
self.duration = self.end_time - self.start_time
self.success = True
self.result = result

def fail(self, error: Exception):
"""Mark task as failed / 标记任务失败"""
self.end_time = time.time()
self.duration = self.end_time - self.start_time
self.success = False
self.error = str(error)

def increment_retry(self):
"""Increment retry count / 增加重试计数"""
self.retry_count += 1

def __str__(self):
status = "✅" if self.success else "❌"
duration = f"{self.duration:.2f}s"
if self.success:
return f"{status} {self.task} ({duration})"
else:
return f"{status} {self.task} ({duration}) - Error: {self.error}"


def example_batch_task_execution(
lang: str = "cn",
max_retries: int = 2,
use_history: bool = True
):
"""Batch task execution example / 批量任务执行示例"""
print("=" * 60)
print("Phone Agent - Batch Task Execution Example")
print("=" * 60)

# Configure model endpoint
model_config = ModelConfig(
base_url="https://open.bigmodel.cn/api/paas/v4",
api_key="",
model_name="autoglm-phone",
temperature=0.1,
)

# Configure Agent behavior
agent_config = AgentConfig(
max_steps=50,
verbose=True,
lang=lang,
)

# Create Agent with history support
agent = PhoneAgent(
model_config=model_config,
agent_config=agent_config,
)

# Define batch tasks
tasks = [
"打开小红书搜索美食攻略",
"打开高德地图查看公交",
"打开美团搜索附近的咖啡店",
"打开bilibili搜索Python教程",
"打开微信,查看最近一条消息",
]

# Initialize results storage
results: List[BatchTaskResult] = []

# Start batch execution
print(f"\n📋 任务列表 ({len(tasks)} tasks):")
for i, task in enumerate(tasks, 1):
print(f" {i}. {task}")

print(f"\n🚀 开始批量执行...")
total_start_time = time.time()

# Execute tasks sequentially
for i, task in enumerate(tasks, 1):
result = BatchTaskResult(task)
results.append(result)

print(f"\n{'=' * 50}")
print(f"📱 Task {i}/{len(tasks)}: {task}")
print(f"{'-' * 50}")

# Execute with retry mechanism
for attempt in range(max_retries + 1):
result.start()

try:
if attempt > 0:
print(f"🔄 Retry {attempt}/{max_retries}...")

# Execute task
task_result = agent.run(task)

# Task completed successfully
result.complete(task_result)
print(f"✅ 任务完成:")
print(f" 结果: {task_result}")
break

except Exception as e:
result.fail(e)
result.increment_retry()
print(f"❌ 任务失败:")
print(f" Error: {e}")

if attempt < max_retries:
print(f"⏱️ 3秒后重试...")
time.sleep(3)
else:
print(f"💥 达到最大重试次数")

# Reset agent state for next task (optional, depends on use case)
if not use_history or "老样子" not in task:
agent.reset()

# Calculate total statistics
total_end_time = time.time()
total_duration = total_end_time - total_start_time
success_count = sum(1 for r in results if r.success)
failure_count = len(results) - success_count

# Generate summary report
print(f"\n{'=' * 60}")
print(f"📊 批量执行总结")
print(f"{'=' * 60}")
print(f"📋 总任务数: {len(tasks)}")
print(f"✅ 成功任务数: {success_count}")
print(f"❌ 失败任务数: {failure_count}")
print(f"⏱️ 总时长: {total_duration:.2f} 秒")
print(f"📈 成功率: {success_count / len(tasks) * 100:.1f}%")

print(f"\n{'-' * 60}")
print(f"📋 任务详情:")
print(f"{'-' * 60}")

for i, result in enumerate(results, 1):
status = "✅" if result.success else "❌"
duration = f"{result.duration:.2f}s"
retry_info = f" (重试: {result.retry_count})" if result.retry_count > 0 else ""

print(f" {i}. {status} {result.task} {duration}{retry_info}")
if not result.success and result.error:
print(f" Error: {result.error[:100]}...")

print(f"\n{'=' * 60}")
print(f"🎉 批量执行完成")
print(f"{'=' * 60}")

# Save results to file (optional)
save_results_to_file(results, "batch_results.txt")
print(f"📁 结果已保存到: batch_results.txt")


def save_results_to_file(results: List[BatchTaskResult], filename: str):
"""Save batch results to file / 将批量结果保存到文件"""
with open(filename, "w", encoding="utf-8") as f:
f.write("Phone Agent Batch Task Execution Results\n")
f.write("=" * 60 + "\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("\n")

for i, result in enumerate(results, 1):
f.write(f"Task {i}: {result.task}\n")
f.write(f"Status: {'SUCCESS' if result.success else 'FAILED'}\n")
f.write(f"Duration: {result.duration:.2f} seconds\n")
if result.retry_count > 0:
f.write(f"Retries: {result.retry_count}\n")
if result.success:
f.write(f"Result: {result.result}\n")
else:
f.write(f"Error: {result.error}\n")
f.write("-" * 40 + "\n")


def main(lang: str = "cn"):
"""Main function / 主函数"""
# Run batch task execution example
example_batch_task_execution(lang)


if __name__ == "__main__":
import argparse

parser = argparse.ArgumentParser(description="Phone Agent Batch Task Execution Example")
parser.add_argument(
"--lang",
type=str,
default="cn",
choices=["cn", "en"],
help="Language for UI messages (cn=Chinese, en=English) / UI消息语言(中文/英文)",
)
args = parser.parse_args()

main(lang=args.lang)