browser-use-rpa¶
第8章 · Agent 的自我进化 · 配套项目
chapter8/browser-use-rpa
项目说明¶
Browser-Use RPA with Learning Capability¶
能操作电脑，并且越做越熟练的 Agent
项目概述¶
本项目实现了一个具有学习能力的浏览器自动化Agent。该Agent能够：
- 学习阶段：通过多模态大模型（GPT-4o, Claude, Gemini等）完成新任务，并捕获成功的操作流程
- 应用阶段：识别相似任务，直接回放已学习的工作流，无需再次调用大模型
- 持续改进：记录执行指标，不断优化知识库
架构设计¶
browser-use-rpa/
├── browser-use/ # Browser-use 核心库（未修改）
├── learning_agent/ # 学习Agent封装层
│ ├── agent.py # 主Agent类，封装browser-use
│ ├── workflow.py # 工作流数据结构
│ ├── knowledge_base.py # 知识库管理
│ └── replay.py # 工作流回放器
├── demo_weather.py # 天气查询演示
├── demo_email.py # 邮件发送演示
└── knowledge_base/ # 存储学习到的工作流
核心组件¶
1. LearningAgent (agent.py)¶
- 封装browser-use的Agent类
- 拦截并记录每个操作步骤
- 提取稳定的XPath选择器
- 管理学习和回放模式
2. Workflow (workflow.py)¶
- 定义工作流数据结构
- 支持参数化（如不同的收件人、主题等）
- 记录元素选择器和操作参数
3. KnowledgeBase (knowledge_base.py)¶
- 持久化存储工作流
- 意图匹配算法
- 性能指标跟踪
4. WorkflowReplayer (replay.py)¶
- 使用Playwright直接控制浏览器
- 智能等待元素加载
- 错误恢复机制
安装¶
# 1. 安装依赖
pip install -r requirements.txt
# 2. 安装Playwright浏览器
playwright install chromium
# 3. 配置环境变量
cp env.example .env
# 编辑.env文件，添加你的API密钥
使用示例¶
基本用法¶
from learning_agent import LearningAgent
from llm_factory import make_llm # 包装层 LLM 工厂：OpenAI 直连，缺 Key 时 OpenRouter 兜底
# 创建学习Agent
agent = LearningAgent(
task="发送邮件给test@example.com，主题是'测试'，内容是'这是一封测试邮件'",
llm=make_llm(), # 默认 gpt-5.6-luna；无 OPENAI_API_KEY 时走 OpenRouter 兜底
knowledge_base_path="./knowledge_base",
headless=False # 显示浏览器界面
)
# 执行任务
result = agent.run_sync(max_steps=20)
print(f"任务完成: {'成功' if result['success'] else '失败'}")
print(f"执行时间: {result['execution_time']:.2f}秒")
print(f"是否使用已学习的工作流: {result['replay_used']}")
运行演示¶
demo_email.py 是本实验的主入口，演示「学习一次工作流 → 用不同参数高速回放」这一核心思想。
它提供了完整的中文命令行接口，运行 python demo_email.py --help 可查看所有参数：
# 运行完整的「学习 → 回放」对比演示（默认行为，会打开浏览器）
python demo_email.py
# 快速冒烟测试：只跑一次简单任务，不做学习/回放对比
python demo_email.py --quick
# 无界面模式 + 使用 Gemini 模型
python demo_email.py --model gemini-2.0-flash-exp --headless
# 自定义两个阶段的任务，并把指标对比写入 JSON 文件
python demo_email.py \
--task '给 a@b.com 发主题为"报告"的邮件' \
--replay-task '给 c@d.com 发主题为"周报"的邮件' \
--output results.json
# 天气查询演示（另一个更轻量的例子）
python demo_weather.py
命令行参数说明（demo_email.py）：
| 参数 | 说明 | 默认值 |
|---|---|---|
--task |
学习阶段的任务描述 | 向 test@example.com 发送测试邮件 |
--replay-task |
回放阶段的任务描述（参数不同、流程相同） | 向 another@example.com 发送邮件 |
--model |
大模型；gpt-* 走 OpenAI（缺 Key 时 OpenRouter 兜底），gemini-* 走 Google |
gpt-5.6-luna |
--headless |
以无界面模式运行浏览器 | 显示窗口 |
--knowledge-base |
工作流知识库存储目录 | ./email_knowledge |
--max-steps |
学习阶段最大操作步数 | 20 |
--output |
把学习/回放指标与知识库统计写入 JSON 文件 | 不写出 |
--quick |
快速冒烟测试模式 | 关闭 |
预期结果： 第一次运行时，Agent 处于学习阶段，会通过多模态大模型逐步探索并录制「发送邮件」工作流，
耗时较长且产生多次 LLM 调用；随后回放阶段用新的收件人/主题参数复用同一工作流，几乎不再调用大模型，
耗时显著下降。演示结束会打印两个阶段的耗时、LLM 调用次数与知识库统计；若指定了 --output，
上述对比会以结构化 JSON 保存，便于复盘“学习一次、复用多次”带来的成本收益。
注意：完整运行需要有效的模型 API Key（见
.env）以及本地可用的浏览器（playwright install chromium）。--help与参数解析本身无需上述依赖即可查看。
验收标准测试¶
1. 首次任务执行（学习阶段）¶
运行 demo_email.py，观察第一阶段：
- Agent通过"观察-思考-行动"循环完成任务
- 每步操作都需要调用大模型
- 成功后自动保存工作流到知识库
- 显示执行时间和步骤数
示例输出：
📚 PHASE 1: LEARNING - First Email Task
Task: Send email to test@example.com
🚀 Starting learning phase...
✅ Learning phase completed!
- Success: ✓
- Execution time: 35.2 seconds
- LLM calls made: 12
- Workflow captured: Yes
2. 重复任务执行（应用阶段）¶
继续观察第二阶段：
- Agent识别相似任务，匹配已学习的工作流
- 直接回放操作步骤，无需调用大模型
- 自动填充新的参数（收件人、主题等）
- 执行速度显著提升
示例输出：
🚀 PHASE 2: REPLAY - Second Email Task
Task: Send email to another@example.com
🔄 Starting replay phase...
✅ Replay phase completed!
- Success: ✓
- Execution time: 8.5 seconds
- Workflow reused: Yes
🎯 Performance Improvements:
- Speed: 4.1x faster
- LLM calls saved: 12
- Time saved: 26.7 seconds
技术特点¶
1. 稳定的元素定位¶
- XPath优先：捕获元素的完整XPath路径，对页面结构变化有较好的鲁棒性
- 多重回退：XPath失败时尝试CSS选择器、属性选择器等
- 智能等待：使用
wait_for(state='visible')确保元素加载完成
2. 工作流捕获机制¶
# 从browser-use的内部状态提取元素信息
element = selector_map[index]
workflow_step = WorkflowStep(
action_type=ActionType.CLICK,
xpath=element.xpath,
element_attributes={
'id': element.attributes.get('id'),
'class': element.attributes.get('class'),
...
}
)
3. 意图匹配算法¶
- 关键词匹配
- 动词识别（send, write, check等）
- 成功率加权
- 置信度评分
性能对比¶
| 指标 | 学习阶段 | 回放阶段 | 提升 |
|---|---|---|---|
| 执行时间 | 30-40秒 | 5-10秒 | 3-5倍 |
| LLM调用次数 | 10-15次 | 0次 | 100% |
| 成功率 | 85% | 95%+ | 10%+ |
知识库管理¶
查看知识库统计：
from learning_agent import KnowledgeBase
kb = KnowledgeBase("./knowledge_base")
stats = kb.get_statistics()
print(stats)
# {
# 'total_workflows': 5,
# 'total_executions': 23,
# 'success_rate': '91.3%',
# 'total_model_calls_saved': 156
# }
清空知识库：
限制和注意事项¶
- 动态内容：对于高度动态的页面，XPath可能会变化
- 认证状态：不会保存登录状态，每次都从头开始
- 复杂交互：暂不支持拖拽、右键菜单等复杂操作
- 多标签页：回放模式简化了标签页处理
扩展建议¶
- 更智能的参数提取：使用NLP模型提取任务参数
- 工作流组合：将多个小工作流组合成复杂任务
- 错误恢复：增强回放失败时的恢复策略
- 分布式知识库：支持团队共享学习成果
贡献¶
欢迎提交Issue和Pull Request来改进本项目。
许可¶
本项目基于browser-use开发，遵循其开源许可协议。
源代码¶
demo_email.py¶
"""
Demo: Email sending with learning capability
This demo shows how the agent learns to send emails and reuses the learned workflow.
Uses Ethereal Email (ethereal.email) as a test email service.
"""
import argparse
import asyncio
import json
import logging
import time
from dotenv import load_dotenv
from browser_use import ChatOpenAI, ChatGoogle
from learning_agent import LearningAgent
from llm_factory import make_llm, DEFAULT_MODEL
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
load_dotenv()
# Default two-phase tasks. Phase 1 teaches the workflow from scratch (expensive,
# multimodal LLM loop); Phase 2 replays it with different parameters (cheap, no LLM).
DEFAULT_LEARNING_TASK = """
Go to https://ethereal.email and send a test email with:
- To: test@example.com
- Subject: Hello from Learning Agent
- Message: This is a test email sent by the learning agent. The workflow will be captured for future reuse.
"""
DEFAULT_REPLAY_TASK = """
Send an email to another@example.com with subject "Workflow Test"
and message "This email is sent using a learned workflow. No LLM calls needed!"
"""
class EmailDemo:
"""Demo for email sending with learning capability."""
def __init__(self, llm_model=DEFAULT_MODEL, knowledge_base_path="./email_knowledge",
headless=False, max_steps=20, learning_task=None, replay_task=None,
output_path=None):
self.llm_model = llm_model
self.knowledge_base_path = knowledge_base_path
self.headless = headless
self.max_steps = max_steps
self.learning_task = learning_task or DEFAULT_LEARNING_TASK
self.replay_task = replay_task or DEFAULT_REPLAY_TASK
self.output_path = output_path
async def run_full_demo(self):
"""Run complete email demo with learning and replay phases."""
print("=" * 80)
print("EMAIL SENDING DEMO - LEARNING AGENT")
print("=" * 80)
print("\nThis demo uses Ethereal Email (ethereal.email) for testing.")
print("Note: This is a test service - emails won't actually be delivered.")
print("=" * 80)
# Phase 1: Learning
await self.phase1_learning()
# Wait before Phase 2
print("\n⏳ Waiting 5 seconds before replay phase...")
await asyncio.sleep(5)
# Phase 2: Replay
await self.phase2_replay()
# Phase 3: Statistics
self.show_statistics()
# Optionally persist the before/after comparison for later analysis
if self.output_path:
self.save_results()
async def phase1_learning(self):
"""Phase 1: Learn how to send an email."""
print("\n" + "📚 PHASE 1: LEARNING - First Email Task ".ljust(70, "="))
print("\nThe agent will learn how to send an email from scratch.")
print("This requires multiple LLM calls to explore and understand the interface.")
print("-" * 70)
# Task with specific details
task = self.learning_task
print(f"Task: {task.strip()}")
print("Expected behavior:")
print(" 1. Navigate to Ethereal Email")
print(" 2. Create a test account (if needed)")
print(" 3. Compose and send the email")
print(" 4. Capture workflow for future reuse")
# Create learning agent
agent = LearningAgent(
task=task,
llm=self._get_llm(),
knowledge_base_path=self.knowledge_base_path,
headless=self.headless
)
print("\n🚀 Starting learning phase...")
start_time = time.time()
try:
result = await agent.run(max_steps=self.max_steps)
elapsed = time.time() - start_time
print("\n✅ Learning phase completed!")
print(f" 📊 Results:")
print(f" - Success: {'✓' if result['success'] else '✗'}")
print(f" - Execution time: {elapsed:.2f} seconds")
print(f" - LLM calls made: {result['llm_calls']}")
print(f" - Workflow captured: {'Yes' if result['success'] else 'No'}")
if result['success']:
print("\n 💡 Workflow successfully learned and saved!")
print(" The agent can now repeat similar tasks without LLM calls.")
# Store metrics for comparison
self.learning_metrics = {
'time': elapsed,
'llm_calls': result['llm_calls'],
'success': result['success']
}
except Exception as e:
print(f"\n❌ Learning phase failed: {e}")
self.learning_metrics = {'time': 0, 'llm_calls': 0, 'success': False}
async def phase2_replay(self):
"""Phase 2: Replay learned workflow with different parameters."""
print("\n" + "🚀 PHASE 2: REPLAY - Second Email Task ".ljust(70, "="))
print("\nThe agent will reuse the learned workflow with different parameters.")
print("This should be much faster and require NO LLM calls.")
print("-" * 70)
# Different email parameters
task = self.replay_task
print(f"Task: {task.strip()}")
print("Expected behavior:")
print(" 1. Match task to learned workflow")
print(" 2. Extract new parameters (recipient, subject, message)")
print(" 3. Replay workflow with new parameters")
print(" 4. Complete task without any LLM calls")
# Create agent for replay
agent = LearningAgent(
task=task,
llm=self._get_llm(),
knowledge_base_path=self.knowledge_base_path,
headless=self.headless
)
print("\n🔄 Starting replay phase...")
start_time = time.time()
try:
result = await agent.run(max_steps=self.max_steps)
elapsed = time.time() - start_time
print("\n✅ Replay phase completed!")
print(f" 📊 Results:")
print(f" - Success: {'✓' if result['success'] else '✗'}")
print(f" - Execution time: {elapsed:.2f} seconds")
print(f" - Workflow reused: {'Yes' if result['replay_used'] else 'No'}")
if result['replay_used']:
# Calculate improvements
if hasattr(self, 'learning_metrics') and self.learning_metrics['success']:
speedup = self.learning_metrics['time'] / elapsed
calls_saved = self.learning_metrics['llm_calls']
print(f"\n 🎯 Performance Improvements:")
print(f" - Speed: {speedup:.1f}x faster")
print(f" - LLM calls saved: {calls_saved}")
print(f" - Time saved: {self.learning_metrics['time'] - elapsed:.1f} seconds")
else:
print(f" - LLM calls made: {result.get('llm_calls', 0)}")
print("\n ⚠️ Workflow was not reused. Task may be too different.")
# Store replay metrics
self.replay_metrics = {
'time': elapsed,
'replay_used': result['replay_used'],
'success': result['success']
}
except Exception as e:
print(f"\n❌ Replay phase failed: {e}")
self.replay_metrics = {'time': 0, 'replay_used': False, 'success': False}
def show_statistics(self):
"""Show knowledge base statistics."""
print("\n" + "📊 KNOWLEDGE BASE STATISTICS ".ljust(70, "="))
from learning_agent import KnowledgeBase
kb = KnowledgeBase(self.knowledge_base_path)
stats = kb.get_statistics()
print("\n Current Knowledge Base Status:")
for key, value in stats.items():
formatted_key = key.replace('_', ' ').title()
print(f" - {formatted_key}: {value}")
# Show comparison if both phases completed
if hasattr(self, 'learning_metrics') and hasattr(self, 'replay_metrics'):
if self.learning_metrics['success'] and self.replay_metrics['replay_used']:
print("\n 📈 Performance Comparison:")
print(f" Phase 1 (Learning):")
print(f" - Time: {self.learning_metrics['time']:.2f}s")
print(f" - LLM Calls: {self.learning_metrics['llm_calls']}")
print(f" Phase 2 (Replay):")
print(f" - Time: {self.replay_metrics['time']:.2f}s")
print(f" - LLM Calls: 0")
improvement = (1 - self.replay_metrics['time'] / self.learning_metrics['time']) * 100
print(f"\n 🚀 Overall Improvement: {improvement:.0f}% faster with replay!")
print("\n" + "=" * 70)
print("DEMO COMPLETED SUCCESSFULLY")
print("=" * 70)
def save_results(self):
"""Persist learning/replay metrics and KB stats to a JSON file."""
from learning_agent import KnowledgeBase
kb = KnowledgeBase(self.knowledge_base_path)
payload = {
'model': self.llm_model,
'knowledge_base_path': self.knowledge_base_path,
'headless': self.headless,
'max_steps': self.max_steps,
'learning_task': self.learning_task.strip(),
'replay_task': self.replay_task.strip(),
'learning_metrics': getattr(self, 'learning_metrics', None),
'replay_metrics': getattr(self, 'replay_metrics', None),
'knowledge_base_stats': kb.get_statistics(),
}
with open(self.output_path, 'w', encoding='utf-8') as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
print(f"\n💾 Results saved to: {self.output_path}")
def _get_llm(self):
"""Get LLM instance based on configuration (OpenAI 直连，缺 Key 时 OpenRouter 兜底)。"""
return make_llm(self.llm_model)
async def quick_test(model=DEFAULT_MODEL, headless=False, max_steps=15,
knowledge_base_path="./test_knowledge", task=None):
"""Quick test with a single simple task (no replay comparison)."""
print("\n🧪 QUICK TEST - Simple Email Task")
print("-" * 40)
task = task or "Go to ethereal.email and send a test email to demo@test.com"
llm = make_llm(model)
agent = LearningAgent(
task=task,
llm=llm,
knowledge_base_path=knowledge_base_path,
headless=headless
)
print(f"Task: {task}")
result = await agent.run(max_steps=max_steps)
print(f"\nResult: {'Success' if result['success'] else 'Failed'}")
print(f"Time: {result['execution_time']:.2f}s")
print(f"LLM calls: {result.get('llm_calls', 0)}")
def build_parser() -> argparse.ArgumentParser:
"""构建命令行参数解析器（RPA 邮件学习/回放演示）。"""
parser = argparse.ArgumentParser(
prog="demo_email.py",
description=(
"browser-use RPA 演示：学习一次「发送邮件」工作流，之后用不同参数高速回放。\n"
"第一阶段（学习）通过多模态大模型逐步探索并录制工作流；\n"
"第二阶段（回放）直接复用工作流、无需再调用大模型，对比耗时与调用次数。"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例：\n"
" python demo_email.py # 运行完整的「学习→回放」对比演示\n"
" python demo_email.py --quick # 只跑一次简单任务，快速冒烟测试\n"
" python demo_email.py --model gemini-2.0-flash-exp --headless\n"
" python demo_email.py --task '给 a@b.com 发主题为\"报告\"的邮件' \\\n"
" --replay-task '给 c@d.com 发主题为\"周报\"的邮件' \\\n"
" --output results.json\n"
),
)
parser.add_argument(
"--task", default=None,
help="学习阶段的任务描述（默认：向 test@example.com 发送测试邮件）",
)
parser.add_argument(
"--replay-task", default=None,
help="回放阶段的任务描述，参数不同但流程相同（默认：向 another@example.com 发送邮件）",
)
parser.add_argument(
"--model", default=DEFAULT_MODEL,
help="使用的大模型，gpt-* 走 OpenAI（缺 Key 时走 OpenRouter 兜底），"
"gemini-* 走 Google（默认：gpt-5.6-luna）",
)
parser.add_argument(
"--headless", action="store_true",
help="以无界面（headless）模式运行浏览器（默认：显示浏览器窗口）",
)
parser.add_argument(
"--knowledge-base", default="./email_knowledge", metavar="PATH",
help="工作流知识库的存储目录（默认：./email_knowledge）",
)
parser.add_argument(
"--max-steps", type=int, default=20, metavar="N",
help="学习阶段允许的最大操作步数（默认：20）",
)
parser.add_argument(
"--output", default=None, metavar="PATH",
help="将学习/回放的指标对比与知识库统计写入指定 JSON 文件",
)
parser.add_argument(
"--quick", action="store_true",
help="快速冒烟测试：只运行一次简单任务，不做学习/回放对比",
)
return parser
def main():
"""Main entry point."""
args = build_parser().parse_args()
if args.quick:
# Run quick test (single task, no replay comparison)
asyncio.run(quick_test(
model=args.model,
headless=args.headless,
max_steps=args.max_steps,
knowledge_base_path=args.knowledge_base,
task=args.task,
))
else:
# Run full learning + replay demo
demo = EmailDemo(
llm_model=args.model,
knowledge_base_path=args.knowledge_base,
headless=args.headless,
max_steps=args.max_steps,
learning_task=args.task,
replay_task=args.replay_task,
output_path=args.output,
)
asyncio.run(demo.run_full_demo())
if __name__ == "__main__":
main()
demo_weather.py¶
"""
Demo: Weather checking with learning capability
This demo shows how the agent learns to check weather and reuses the learned workflow.
"""
import asyncio
import logging
from dotenv import load_dotenv
from browser_use import ChatOpenAI
from learning_agent import LearningAgent
from llm_factory import make_llm
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
load_dotenv()
async def demo_weather_learning():
"""Demonstrate weather checking with learning."""
print("=" * 60)
print("WEATHER CHECKING DEMO - LEARNING AGENT")
print("=" * 60)
# First task - agent will learn
print("\n📚 PHASE 1: LEARNING - First weather check (Beijing)")
print("-" * 40)
task1 = "Check the weather in Beijing"
agent1 = LearningAgent(
task=task1,
llm=make_llm(),
knowledge_base_path="./weather_knowledge",
headless=False # Show browser for demo
)
print(f"Task: {task1}")
print("The agent will use browser-use to complete this task from scratch...")
result1 = await agent1.run(max_steps=10)
print(f"\n✅ Task completed!")
print(f" - Success: {result1['success']}")
print(f" - Execution time: {result1['execution_time']:.2f}s")
print(f" - LLM calls made: {result1['llm_calls']}")
print(f" - Workflow learned: {'Yes' if result1['success'] else 'No'}")
# Wait a bit before second task
await asyncio.sleep(3)
# Second task - agent should reuse learned workflow
print("\n🚀 PHASE 2: REPLAY - Second weather check (Shanghai)")
print("-" * 40)
task2 = "Check the weather in Shanghai"
agent2 = LearningAgent(
task=task2,
llm=make_llm(),
knowledge_base_path="./weather_knowledge",
headless=False
)
print(f"Task: {task2}")
print("The agent will try to reuse the learned workflow...")
result2 = await agent2.run(max_steps=10)
print(f"\n✅ Task completed!")
print(f" - Success: {result2['success']}")
print(f" - Execution time: {result2['execution_time']:.2f}s")
print(f" - Replay used: {result2['replay_used']}")
if result2['replay_used']:
print(f" - LLM calls saved: {result1['llm_calls']}")
speedup = result1['execution_time'] / result2['execution_time']
print(f" - Speed improvement: {speedup:.1f}x faster")
else:
print(f" - LLM calls made: {result2['llm_calls']}")
# Show knowledge base statistics
print("\n📊 KNOWLEDGE BASE STATISTICS")
print("-" * 40)
kb = agent2.knowledge_base
stats = kb.get_statistics()
for key, value in stats.items():
print(f" - {key.replace('_', ' ').title()}: {value}")
print("\n" + "=" * 60)
print("DEMO COMPLETED")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(demo_weather_learning())
learning_agent/__init__.py¶
"""
Learning Agent for Browser-Use RPA
A wrapper system that adds learning and experience-replay capabilities to browser-use.
This agent can learn from successful task executions and replay learned workflows efficiently.
"""
from .agent import LearningAgent
from .knowledge_base import KnowledgeBase
from .workflow import Workflow, WorkflowStep
__all__ = ['LearningAgent', 'KnowledgeBase', 'Workflow', 'WorkflowStep']
learning_agent/agent.py¶
"""
Learning Agent that extends browser-use with experience-based learning.
This agent wraps the browser-use Agent to capture successful workflows
and replay them efficiently without LLM calls.
"""
import asyncio
import logging
from typing import Any, Dict, List, Optional
import time
from pathlib import Path
import sys
import os
# Add parent directory to path to import browser-use
sys.path.append(str(Path(__file__).parent.parent))
from browser_use import Agent, Browser
from browser_use.agent.views import AgentOutput, ActionResult
from browser_use.browser.views import BrowserStateSummary
from .workflow import Workflow, WorkflowStep, ActionType
from .knowledge_base import KnowledgeBase
from .replay import WorkflowReplayer
logger = logging.getLogger(__name__)
class LearningAgent:
"""
An agent that learns from experience and can replay learned workflows.
This agent:
1. Attempts to match tasks to learned workflows
2. Falls back to browser-use Agent for new tasks
3. Captures successful executions as new workflows
4. Improves over time by building a knowledge base
"""
def __init__(self,
task: str,
llm: Any,
browser: Optional[Browser] = None,
knowledge_base_path: str = "./knowledge_base",
headless: bool = False,
**agent_kwargs):
"""
Initialize the learning agent.
Args:
task: The task to be performed
llm: Language model to use (browser-use compatible)
browser: Browser instance (optional)
knowledge_base_path: Path to store learned workflows
headless: Whether to run browser in headless mode for replay
**agent_kwargs: Additional arguments for browser-use Agent
"""
self.task = task
self.llm = llm
self.browser = browser
self.headless = headless
# Initialize knowledge base
self.knowledge_base = KnowledgeBase(knowledge_base_path)
# Initialize workflow replayer
self.replayer = WorkflowReplayer(headless=headless)
# Workflow capture state
self.current_workflow: Optional[Workflow] = None
self.is_learning = False
self.captured_steps: List[Dict[str, Any]] = []
# Browser-use agent (lazy initialization)
self._agent: Optional[Agent] = None
self._agent_kwargs = agent_kwargs
# Metrics
self.metrics = {
"llm_calls": 0,
"replay_used": False,
"execution_time": 0,
"success": False
}
@property
def agent(self) -> Agent:
"""Lazy initialization of browser-use agent."""
if self._agent is None:
# Create agent with step callback for capturing
self._agent = Agent(
task=self.task,
llm=self.llm,
browser=self.browser,
**self._agent_kwargs
)
# Store original step method
self._original_step = self._agent.step
# Monkey-patch the step method to capture actions
self._agent.step = self._wrapped_step
return self._agent
async def _wrapped_step(self, step_info=None):
"""Wrapped step method that captures workflow information."""
# Call original step
await self._original_step(step_info)
# Capture step information if learning
if self.is_learning:
await self._capture_step()
async def _capture_step(self):
"""Capture the current step for workflow learning."""
try:
# Get the last action and result from agent state
if self.agent.state.last_model_output and self.agent.state.last_result:
model_output = self.agent.state.last_model_output
results = self.agent.state.last_result
# Get browser state for element information
browser_state = await self.agent.browser_session.get_browser_state_summary()
# Process each action in the step
for i, (action, result) in enumerate(zip(model_output.action, results)):
if result and not result.error:
# Extract action details
action_data = self._extract_action_data(action, result, browser_state)
if action_data:
self.captured_steps.append(action_data)
logger.debug(f"Captured step: {action_data['type']}")
except Exception as e:
logger.error(f"Failed to capture step: {e}")
def _extract_action_data(self, action: Any, result: ActionResult, browser_state: BrowserStateSummary) -> Optional[Dict[str, Any]]:
"""
Extract action data for workflow capture.
Args:
action: The action object from browser-use
result: The result of the action
browser_state: Current browser state
Returns:
Dictionary containing action data, or None if extraction fails
"""
try:
action_dict = action.model_dump() if hasattr(action, 'model_dump') else {}
# Determine action type
action_type = None
parameters = {}
element_info = None
# Parse different action types
if 'go_to_url' in action_dict:
action_type = ActionType.NAVIGATE
parameters = {'url': action_dict['go_to_url'].get('url')}
elif 'click_element' in action_dict:
action_type = ActionType.CLICK
click_data = action_dict['click_element']
parameters = {
'while_holding_ctrl': click_data.get('while_holding_ctrl', False)
}
# Get element info from selector map
index = click_data.get('index')
if index and browser_state.dom_state.selector_map:
element_info = self._get_element_info(index, browser_state.dom_state.selector_map)
elif 'input_text' in action_dict:
action_type = ActionType.INPUT_TEXT
input_data = action_dict['input_text']
parameters = {
'text': input_data.get('text', ''),
'clear_existing': input_data.get('clear_existing', True)
}
# Get element info
index = input_data.get('index')
if index and browser_state.dom_state.selector_map:
element_info = self._get_element_info(index, browser_state.dom_state.selector_map)
elif 'select_dropdown_option' in action_dict:
action_type = ActionType.SELECT_OPTION
select_data = action_dict['select_dropdown_option']
parameters = {
'text': select_data.get('text', '')
}
index = select_data.get('index')
if index and browser_state.dom_state.selector_map:
element_info = self._get_element_info(index, browser_state.dom_state.selector_map)
elif 'scroll' in action_dict:
action_type = ActionType.SCROLL
scroll_data = action_dict['scroll']
parameters = {
'down': scroll_data.get('down', True),
'num_pages': scroll_data.get('num_pages', 1)
}
elif 'upload_file' in action_dict:
action_type = ActionType.UPLOAD_FILE
upload_data = action_dict['upload_file']
parameters = {
'path': upload_data.get('path', '')
}
index = upload_data.get('index')
if index and browser_state.dom_state.selector_map:
element_info = self._get_element_info(index, browser_state.dom_state.selector_map)
elif 'done' in action_dict:
# Skip done action for workflow
return None
if action_type:
return {
'type': action_type,
'parameters': parameters,
'element_info': element_info,
'url': browser_state.url
}
except Exception as e:
logger.error(f"Failed to extract action data: {e}")
return None
def _get_element_info(self, index: int, selector_map: Dict) -> Optional[Dict[str, Any]]:
"""
Get element information from selector map.
Args:
index: Element index
selector_map: Browser-use selector map
Returns:
Dictionary containing element selectors and attributes
"""
try:
if index in selector_map:
element = selector_map[index]
# Extract stable selectors
info = {
'xpath': getattr(element, 'xpath', None),
'attributes': {}
}
# Get important attributes
if hasattr(element, 'attributes') and element.attributes:
for attr in ['id', 'name', 'class', 'type', 'role', 'aria-label', 'data-testid']:
if attr in element.attributes:
info['attributes'][attr] = element.attributes[attr]
return info
except Exception as e:
logger.error(f"Failed to get element info: {e}")
return None
async def run(self, max_steps: int = 100) -> Dict[str, Any]:
"""
Run the learning agent to complete the task.
Args:
max_steps: Maximum steps for browser-use agent
Returns:
Dictionary containing execution results and metrics
"""
start_time = time.time()
try:
# Check if we have a learned workflow for this task
match = self.knowledge_base.find_workflow_for_task(self.task)
if match and match.confidence > 0.6:
# Use learned workflow
logger.info(f"Found matching workflow: '{match.workflow.intent}' "
f"(confidence: {match.confidence:.2f})")
logger.info(f"Match reason: {match.match_reason}")
result = await self._run_with_replay(match.workflow)
# Update metrics
self.knowledge_base.update_workflow_metrics(
match.workflow.workflow_id,
success=result['success'],
execution_time=result['execution_time'],
model_calls_saved=result['model_calls_saved']
)
self.metrics['replay_used'] = True
self.metrics['success'] = result['success']
# If replay failed, fall back to learning mode
if not result['success']:
logger.warning("Replay failed, falling back to learning mode")
result = await self._run_with_learning(max_steps)
else:
# No matching workflow, run in learning mode
logger.info("No matching workflow found, running in learning mode")
result = await self._run_with_learning(max_steps)
finally:
self.metrics['execution_time'] = time.time() - start_time
# Log performance comparison
if self.metrics['replay_used']:
logger.info(f"Task completed with replay in {self.metrics['execution_time']:.2f}s")
logger.info(f"Model calls saved: {result.get('model_calls_saved', 0)}")
else:
logger.info(f"Task completed with learning in {self.metrics['execution_time']:.2f}s")
logger.info(f"LLM calls made: {self.metrics['llm_calls']}")
return self.metrics
async def _run_with_replay(self, workflow: Workflow) -> Dict[str, Any]:
"""
Run task using a learned workflow.
Args:
workflow: The workflow to replay
Returns:
Execution results
"""
logger.info("Replaying learned workflow...")
# Extract parameters from task if needed
parameters = self._extract_task_parameters(self.task, workflow)
# Setup replayer
await self.replayer.setup()
try:
# Replay workflow
result = await self.replayer.replay_workflow(
workflow,
parameters=parameters
)
logger.info(f"Replay completed: {result['steps_completed']}/{result['total_steps']} steps")
return result
finally:
await self.replayer.cleanup()
async def _run_with_learning(self, max_steps: int) -> Dict[str, Any]:
"""
Run task with browser-use agent and capture workflow.
Args:
max_steps: Maximum steps for agent
Returns:
Execution results
"""
logger.info("Running with browser-use agent (learning mode)...")
# Enable learning mode
self.is_learning = True
self.captured_steps = []
# Track LLM calls
original_get_model_output = self.agent.get_model_output
async def tracked_get_model_output(*args, **kwargs):
self.metrics['llm_calls'] += 1
return await original_get_model_output(*args, **kwargs)
self.agent.get_model_output = tracked_get_model_output
try:
# Run the agent
await self.agent.run(max_steps=max_steps)
# Check if task was successful
success = False
if self.agent.state.last_result:
for result in self.agent.state.last_result:
if result and hasattr(result, 'success') and result.success:
success = True
break
self.metrics['success'] = success
# If successful, save the workflow
if success and self.captured_steps:
await self._save_learned_workflow()
return {
'success': success,
'steps_completed': len(self.captured_steps),
'total_steps': len(self.captured_steps),
'execution_time': self.metrics['execution_time'],
'model_calls_saved': 0
}
finally:
self.is_learning = False
async def _save_learned_workflow(self):
"""Save the captured workflow to knowledge base."""
try:
# Create workflow from captured steps
workflow = Workflow(
workflow_id="", # Will be generated
intent=self.task,
description=f"Learned workflow for: {self.task}",
initial_url=self.captured_steps[0].get('url') if self.captured_steps else None
)
# Convert captured steps to workflow steps
for step_data in self.captured_steps:
step = WorkflowStep(
action_type=step_data['type'],
parameters=step_data['parameters']
)
# Add element info if available
if step_data.get('element_info'):
element_info = step_data['element_info']
step.xpath = element_info.get('xpath')
step.element_attributes = element_info.get('attributes', {})
workflow.add_step(step)
# Save to knowledge base
self.knowledge_base.save_workflow(workflow)
logger.info(f"Saved learned workflow with {len(workflow.steps)} steps")
except Exception as e:
logger.error(f"Failed to save learned workflow: {e}")
def _extract_task_parameters(self, task: str, workflow: Workflow) -> Dict[str, Any]:
"""
Extract parameters from task description for workflow.
Args:
task: Task description
workflow: Workflow that needs parameters
Returns:
Dictionary of extracted parameters
"""
# This is a simplified parameter extraction
# In production, you might use NLP or regex patterns
parameters = {}
# Example: Extract email addresses
import re
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
emails = re.findall(email_pattern, task)
if emails:
parameters['recipient'] = emails[0]
# Extract quoted text as subject or content
quoted = re.findall(r'"([^"]*)"', task)
if quoted:
if 'subject' in task.lower() or '主题' in task.lower():
parameters['subject'] = quoted[0]
if len(quoted) > 1:
parameters['content'] = quoted[1]
else:
parameters['content'] = quoted[0]
return parameters
def run_sync(self, max_steps: int = 100) -> Dict[str, Any]:
"""
Synchronous wrapper for run method.
Args:
max_steps: Maximum steps for browser-use agent
Returns:
Execution results
"""
return asyncio.run(self.run(max_steps))
learning_agent/knowledge_base.py¶
"""
Knowledge Base for storing and retrieving learned workflows.
This module provides persistent storage and intelligent retrieval of workflows,
including intent matching and workflow selection.
"""
import json
import os
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from datetime import datetime
import logging
from dataclasses import dataclass
import uuid
from .workflow import Workflow, WorkflowStep
logger = logging.getLogger(__name__)
@dataclass
class IntentMatch:
"""Represents a match between a task intent and a stored workflow"""
workflow: Workflow
confidence: float # 0.0 to 1.0
match_reason: str
class KnowledgeBase:
"""
Manages storage and retrieval of learned workflows.
The knowledge base provides:
- Persistent storage of workflows
- Intent matching to find relevant workflows
- Performance tracking and optimization
"""
def __init__(self, storage_path: str = "./knowledge_base"):
"""
Initialize the knowledge base.
Args:
storage_path: Directory path for storing workflow data
"""
self.storage_path = Path(storage_path)
self.storage_path.mkdir(exist_ok=True)
# In-memory cache of workflows
self.workflows: Dict[str, Workflow] = {}
# Intent index for fast matching
self.intent_index: Dict[str, List[str]] = {} # intent -> [workflow_ids]
# Load existing workflows
self.load_all_workflows()
def save_workflow(self, workflow: Workflow) -> None:
"""
Save a workflow to persistent storage.
Args:
workflow: The workflow to save
"""
# Generate ID if not present
if not workflow.workflow_id:
workflow.workflow_id = str(uuid.uuid4())
# Save to file
workflow_file = self.storage_path / f"workflow_{workflow.workflow_id}.json"
with open(workflow_file, 'w', encoding='utf-8') as f:
f.write(workflow.to_json())
# Update in-memory cache
self.workflows[workflow.workflow_id] = workflow
# Update intent index
if workflow.intent not in self.intent_index:
self.intent_index[workflow.intent] = []
if workflow.workflow_id not in self.intent_index[workflow.intent]:
self.intent_index[workflow.intent].append(workflow.workflow_id)
logger.info(f"Saved workflow '{workflow.workflow_id}' for intent: {workflow.intent}")
def load_all_workflows(self) -> None:
"""Load all workflows from storage into memory."""
workflow_files = list(self.storage_path.glob("workflow_*.json"))
for workflow_file in workflow_files:
try:
with open(workflow_file, 'r', encoding='utf-8') as f:
workflow_data = json.load(f)
workflow = Workflow.from_dict(workflow_data)
# Add to cache
self.workflows[workflow.workflow_id] = workflow
# Update intent index
if workflow.intent not in self.intent_index:
self.intent_index[workflow.intent] = []
self.intent_index[workflow.intent].append(workflow.workflow_id)
except Exception as e:
logger.error(f"Failed to load workflow from {workflow_file}: {e}")
logger.info(f"Loaded {len(self.workflows)} workflows from storage")
def find_workflow_for_task(self, task_description: str) -> Optional[IntentMatch]:
"""
Find the best matching workflow for a given task.
Args:
task_description: Natural language description of the task
Returns:
The best matching workflow with confidence score, or None if no match
"""
matches = self.find_matching_workflows(task_description)
if matches:
# Return the highest confidence match
return max(matches, key=lambda m: m.confidence)
return None
def find_matching_workflows(self, task_description: str) -> List[IntentMatch]:
"""
Find all workflows that might match the given task.
Args:
task_description: Natural language description of the task
Returns:
List of matching workflows sorted by confidence
"""
matches = []
# Normalize task description for matching
task_lower = task_description.lower()
for workflow in self.workflows.values():
confidence, reason = self._calculate_match_confidence(task_lower, workflow)
if confidence > 0.3: # Minimum threshold
matches.append(IntentMatch(
workflow=workflow,
confidence=confidence,
match_reason=reason
))
# Sort by confidence (highest first)
matches.sort(key=lambda m: m.confidence, reverse=True)
return matches
def _calculate_match_confidence(self, task: str, workflow: Workflow) -> Tuple[float, str]:
"""
Calculate how well a workflow matches a task description.
Args:
task: Normalized task description
workflow: Workflow to match against
Returns:
Tuple of (confidence_score, match_reason)
"""
confidence = 0.0
reasons = []
# Check intent match
intent_lower = workflow.intent.lower()
# Exact intent match
if intent_lower in task:
confidence += 0.5
reasons.append("exact intent match")
# Keyword matching for common patterns
intent_keywords = set(intent_lower.split())
task_keywords = set(task.split())
# Calculate keyword overlap
common_keywords = intent_keywords & task_keywords
if common_keywords:
keyword_score = len(common_keywords) / len(intent_keywords)
confidence += keyword_score * 0.3
reasons.append(f"keyword match: {', '.join(common_keywords)}")
# Check for action verbs (send, write, compose, create, etc.)
action_verbs = {
'send': ['send', 'email', 'mail', 'message'],
'write': ['write', 'compose', 'draft', 'create'],
'search': ['search', 'find', 'look', 'query'],
'check': ['check', 'verify', 'view', 'see'],
'login': ['login', 'signin', 'authenticate', 'log in', 'sign in'],
'order': ['order', 'buy', 'purchase', 'checkout'],
'book': ['book', 'reserve', 'schedule']
}
for action_group, verbs in action_verbs.items():
if any(verb in intent_lower for verb in verbs) and any(verb in task for verb in verbs):
confidence += 0.2
reasons.append(f"action verb match: {action_group}")
break
# Boost confidence for recently successful workflows
if workflow.success_count > workflow.failure_count:
success_rate = workflow.success_count / (workflow.success_count + workflow.failure_count)
confidence *= (1 + success_rate * 0.2)
if success_rate > 0.8:
reasons.append(f"high success rate: {success_rate:.0%}")
# Compile reason string
reason = "; ".join(reasons) if reasons else "partial match"
return confidence, reason
def update_workflow_metrics(self,
workflow_id: str,
success: bool,
execution_time: float,
model_calls_saved: int = 0) -> None:
"""
Update performance metrics for a workflow after execution.
Args:
workflow_id: ID of the workflow that was executed
success: Whether the execution was successful
execution_time: Time taken to execute the workflow
model_calls_saved: Number of LLM calls saved by using this workflow
"""
if workflow_id in self.workflows:
workflow = self.workflows[workflow_id]
# Update counters
if success:
workflow.success_count += 1
else:
workflow.failure_count += 1
# Update timing
workflow.last_used_at = datetime.now()
# Update average execution time
total_executions = workflow.success_count + workflow.failure_count
workflow.average_execution_time = (
(workflow.average_execution_time * (total_executions - 1) + execution_time)
/ total_executions
)
# Track model calls saved
workflow.model_calls_saved += model_calls_saved
# Save updated workflow
self.save_workflow(workflow)
logger.info(f"Updated metrics for workflow {workflow_id}: "
f"success={success}, time={execution_time:.2f}s, "
f"total_saved_calls={workflow.model_calls_saved}")
def get_statistics(self) -> Dict[str, any]:
"""
Get statistics about the knowledge base.
Returns:
Dictionary containing knowledge base statistics
"""
total_workflows = len(self.workflows)
total_executions = sum(w.success_count + w.failure_count for w in self.workflows.values())
total_successes = sum(w.success_count for w in self.workflows.values())
total_model_calls_saved = sum(w.model_calls_saved for w in self.workflows.values())
success_rate = (total_successes / total_executions * 100) if total_executions > 0 else 0
return {
"total_workflows": total_workflows,
"total_executions": total_executions,
"total_successes": total_successes,
"success_rate": f"{success_rate:.1f}%",
"total_model_calls_saved": total_model_calls_saved,
"unique_intents": len(self.intent_index)
}
def clear_all(self) -> None:
"""Clear all workflows from the knowledge base (use with caution)."""
# Clear files
for workflow_file in self.storage_path.glob("workflow_*.json"):
workflow_file.unlink()
# Clear memory
self.workflows.clear()
self.intent_index.clear()
logger.info("Cleared all workflows from knowledge base")
learning_agent/replay.py¶
"""
Workflow replay functionality using Playwright.
This module provides reliable replay of learned workflows by directly
controlling the browser through Playwright, bypassing the need for LLM calls.
"""
import asyncio
import logging
from typing import Any, Dict, Optional
from playwright.async_api import Page, Browser, async_playwright, Locator
import time
from .workflow import Workflow, WorkflowStep, ActionType
logger = logging.getLogger(__name__)
class WorkflowReplayer:
"""
Replays learned workflows using Playwright for direct browser control.
This replayer:
- Executes workflows without LLM calls
- Handles dynamic page loading with smart waits
- Provides robust error recovery
- Tracks execution metrics
"""
def __init__(self, headless: bool = False):
"""
Initialize the workflow replayer.
Args:
headless: Whether to run browser in headless mode
"""
self.headless = headless
self.browser: Optional[Browser] = None
self.page: Optional[Page] = None
self.context = None
self.playwright = None
async def setup(self):
"""Initialize Playwright and browser."""
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(
headless=self.headless,
args=['--disable-blink-features=AutomationControlled']
)
self.context = await self.browser.new_context(
viewport={'width': 1280, 'height': 720},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
)
self.page = await self.context.new_page()
# Set default timeout
self.page.set_default_timeout(30000)
logger.info("Workflow replayer initialized")
async def cleanup(self):
"""Clean up browser resources."""
if self.page:
await self.page.close()
if self.context:
await self.context.close()
if self.browser:
await self.browser.close()
if self.playwright:
await self.playwright.stop()
async def replay_workflow(self,
workflow: Workflow,
parameters: Optional[Dict[str, Any]] = None,
initial_url: Optional[str] = None) -> Dict[str, Any]:
"""
Replay a workflow with the given parameters.
Args:
workflow: The workflow to replay
parameters: Parameters to apply to the workflow
initial_url: Starting URL (uses workflow's initial_url if not provided)
Returns:
Dictionary containing execution results and metrics
"""
start_time = time.time()
results = {
"success": False,
"steps_completed": 0,
"total_steps": len(workflow.steps),
"errors": [],
"execution_time": 0,
"model_calls_saved": len(workflow.steps) # Each step would require an LLM call
}
try:
# Apply parameters if provided
if parameters:
workflow = workflow.parameterize(parameters)
# Navigate to initial URL
start_url = initial_url or workflow.initial_url
if start_url:
logger.info(f"Navigating to initial URL: {start_url}")
await self.page.goto(start_url, wait_until='domcontentloaded')
await self.page.wait_for_load_state('networkidle', timeout=10000)
# Execute each step
for i, step in enumerate(workflow.steps):
logger.info(f"Executing step {i+1}/{len(workflow.steps)}: {step.action_type.value}")
try:
await self._execute_step(step)
results["steps_completed"] += 1
# Small delay between actions for stability
await asyncio.sleep(0.5)
except Exception as e:
error_msg = f"Step {i+1} failed: {str(e)}"
logger.error(error_msg)
results["errors"].append(error_msg)
# Try to recover if it's not a critical error
if step.action_type not in [ActionType.NAVIGATE, ActionType.CLICK]:
continue
else:
break
# Check if workflow completed successfully
results["success"] = (results["steps_completed"] == results["total_steps"])
except Exception as e:
logger.error(f"Workflow replay failed: {e}")
results["errors"].append(str(e))
finally:
results["execution_time"] = time.time() - start_time
logger.info(f"Workflow replay completed in {results['execution_time']:.2f}s")
return results
async def _execute_step(self, step: WorkflowStep) -> None:
"""
Execute a single workflow step.
Args:
step: The step to execute
"""
# Wait before action if specified
if step.wait_before > 0:
await asyncio.sleep(step.wait_before)
# Execute based on action type
if step.action_type == ActionType.NAVIGATE:
await self._execute_navigate(step)
elif step.action_type == ActionType.CLICK:
await self._execute_click(step)
elif step.action_type == ActionType.INPUT_TEXT:
await self._execute_input(step)
elif step.action_type == ActionType.SELECT_OPTION:
await self._execute_select(step)
elif step.action_type == ActionType.SCROLL:
await self._execute_scroll(step)
elif step.action_type == ActionType.WAIT:
await self._execute_wait(step)
elif step.action_type == ActionType.SWITCH_TAB:
await self._execute_switch_tab(step)
elif step.action_type == ActionType.UPLOAD_FILE:
await self._execute_upload(step)
else:
logger.warning(f"Unsupported action type: {step.action_type}")
async def _get_element(self, step: WorkflowStep) -> Locator:
"""
Get element using stable selectors with fallback.
Args:
step: The step containing selector information
Returns:
Playwright Locator for the element
"""
locator = None
# Try XPath first (most stable)
if step.xpath:
try:
locator = self.page.locator(f"xpath={step.xpath}")
# Check if element exists
if await locator.count() > 0:
# Wait for element to be ready
await locator.wait_for(state='visible', timeout=step.timeout * 1000)
return locator
except Exception as e:
logger.debug(f"XPath locator failed: {e}")
# Try CSS selector as fallback
if step.css_selector:
try:
locator = self.page.locator(step.css_selector)
if await locator.count() > 0:
await locator.wait_for(state='visible', timeout=step.timeout * 1000)
return locator
except Exception as e:
logger.debug(f"CSS selector failed: {e}")
# Try to build selector from attributes
if step.element_attributes:
selector_parts = []
# Use ID if available
if 'id' in step.element_attributes:
return self.page.locator(f"#{step.element_attributes['id']}")
# Build attribute selector
for attr, value in step.element_attributes.items():
if attr in ['name', 'type', 'role', 'aria-label', 'data-testid']:
selector_parts.append(f"[{attr}='{value}']")
if selector_parts:
selector = ''.join(selector_parts)
try:
locator = self.page.locator(selector)
if await locator.count() > 0:
await locator.wait_for(state='visible', timeout=step.timeout * 1000)
return locator
except Exception as e:
logger.debug(f"Attribute selector failed: {e}")
# Try text content as last resort
if 'text' in step.parameters:
text = step.parameters['text']
locator = self.page.get_by_text(text)
if await locator.count() > 0:
return locator
raise Exception(f"Could not find element for step: {step.description or step.action_type}")
async def _execute_navigate(self, step: WorkflowStep) -> None:
"""Execute navigation action."""
url = step.parameters.get('url')
if url:
logger.debug(f"Navigating to: {url}")
await self.page.goto(url, wait_until='domcontentloaded')
await self.page.wait_for_load_state('networkidle', timeout=10000)
async def _execute_click(self, step: WorkflowStep) -> None:
"""Execute click action with smart waiting."""
element = await self._get_element(step)
# Ensure element is clickable
await element.wait_for(state='visible', timeout=step.timeout * 1000)
await element.scroll_into_view_if_needed()
# Check for ctrl/cmd modifier
if step.parameters.get('while_holding_ctrl'):
await element.click(modifiers=['Control'])
else:
await element.click()
# Wait for potential navigation or dynamic updates
try:
await self.page.wait_for_load_state('networkidle', timeout=5000)
except:
pass # Page might not navigate
async def _execute_input(self, step: WorkflowStep) -> None:
"""Execute text input action."""
element = await self._get_element(step)
text = step.parameters.get('text', '')
clear_existing = step.parameters.get('clear_existing', True)
# Click to focus
await element.click()
# Clear existing text if needed
if clear_existing:
await element.fill('')
# Type the text
await element.type(text, delay=50) # Small delay for more human-like typing
async def _execute_select(self, step: WorkflowStep) -> None:
"""Execute select option action."""
element = await self._get_element(step)
option_text = step.parameters.get('text', '')
# Try to select by visible text
await element.select_option(label=option_text)
async def _execute_scroll(self, step: WorkflowStep) -> None:
"""Execute scroll action."""
direction = 'down' if step.parameters.get('down', True) else 'up'
pages = step.parameters.get('num_pages', 1)
# Calculate scroll amount
viewport_height = await self.page.evaluate('window.innerHeight')
scroll_amount = viewport_height * pages
if direction == 'down':
await self.page.evaluate(f'window.scrollBy(0, {scroll_amount})')
else:
await self.page.evaluate(f'window.scrollBy(0, -{scroll_amount})')
# Wait for any lazy-loaded content
await asyncio.sleep(0.5)
async def _execute_wait(self, step: WorkflowStep) -> None:
"""Execute wait action."""
wait_time = step.parameters.get('seconds', 1)
await asyncio.sleep(wait_time)
async def _execute_switch_tab(self, step: WorkflowStep) -> None:
"""Execute tab switching (simplified for single tab replay)."""
# In replay mode, we typically work with a single tab
# This is a placeholder for multi-tab support
logger.debug("Tab switching in replay mode - continuing in current tab")
async def _execute_upload(self, step: WorkflowStep) -> None:
"""Execute file upload action."""
element = await self._get_element(step)
file_path = step.parameters.get('path', '')
# Set the file(s) to upload
await element.set_input_files(file_path)
learning_agent/workflow.py¶
"""
Workflow data structures for capturing and storing browser action sequences.
This module defines the structures used to represent learned workflows,
including individual steps and complete action sequences.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
from enum import Enum
import json
class ActionType(Enum):
"""Types of actions that can be recorded in a workflow"""
NAVIGATE = "navigate"
CLICK = "click"
INPUT_TEXT = "input_text"
SELECT_OPTION = "select_option"
SCROLL = "scroll"
WAIT = "wait"
SWITCH_TAB = "switch_tab"
CLOSE_TAB = "close_tab"
UPLOAD_FILE = "upload_file"
@dataclass
class WorkflowStep:
"""Represents a single step in a workflow"""
action_type: ActionType
# Stable selectors for element identification
xpath: Optional[str] = None
css_selector: Optional[str] = None
# Action parameters
parameters: Dict[str, Any] = field(default_factory=dict)
# Additional context
element_attributes: Dict[str, str] = field(default_factory=dict)
description: str = ""
# Timing information
wait_before: float = 0.0 # Seconds to wait before executing this step
timeout: float = 15.0 # Maximum time to wait for element to be ready
# Validation
expected_outcome: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert step to dictionary for serialization"""
return {
"action_type": self.action_type.value,
"xpath": self.xpath,
"css_selector": self.css_selector,
"parameters": self.parameters,
"element_attributes": self.element_attributes,
"description": self.description,
"wait_before": self.wait_before,
"timeout": self.timeout,
"expected_outcome": self.expected_outcome
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'WorkflowStep':
"""Create step from dictionary"""
data = data.copy()
data['action_type'] = ActionType(data['action_type'])
return cls(**data)
@classmethod
def from_browser_action(cls, action_type: str, element: Optional[Any] = None, **params) -> 'WorkflowStep':
"""Create a workflow step from browser-use action and element info"""
step = cls(
action_type=ActionType(action_type.lower()),
parameters=params
)
if element:
# Extract stable selectors from DOMInteractedElement
if hasattr(element, 'x_path'):
step.xpath = element.x_path
# Store relevant attributes for fallback identification
if hasattr(element, 'attributes') and element.attributes:
step.element_attributes = {
k: v for k, v in element.attributes.items()
if k in ['id', 'name', 'class', 'type', 'role', 'aria-label', 'data-testid']
}
return step
@dataclass
class Workflow:
"""Represents a complete workflow that can be learned and replayed"""
# Identification
workflow_id: str
intent: str # The task intent this workflow accomplishes
# Steps
steps: List[WorkflowStep] = field(default_factory=list)
# Metadata
created_at: datetime = field(default_factory=datetime.now)
last_used_at: Optional[datetime] = None
success_count: int = 0
failure_count: int = 0
# Learning context
initial_url: Optional[str] = None
example_parameters: Dict[str, Any] = field(default_factory=dict)
description: str = ""
# Performance metrics
average_execution_time: float = 0.0
model_calls_saved: int = 0
def add_step(self, step: WorkflowStep) -> None:
"""Add a step to the workflow"""
self.steps.append(step)
def to_dict(self) -> Dict[str, Any]:
"""Convert workflow to dictionary for serialization"""
return {
"workflow_id": self.workflow_id,
"intent": self.intent,
"steps": [step.to_dict() for step in self.steps],
"created_at": self.created_at.isoformat(),
"last_used_at": self.last_used_at.isoformat() if self.last_used_at else None,
"success_count": self.success_count,
"failure_count": self.failure_count,
"initial_url": self.initial_url,
"example_parameters": self.example_parameters,
"description": self.description,
"average_execution_time": self.average_execution_time,
"model_calls_saved": self.model_calls_saved
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Workflow':
"""Create workflow from dictionary"""
data = data.copy()
data['steps'] = [WorkflowStep.from_dict(s) for s in data.get('steps', [])]
data['created_at'] = datetime.fromisoformat(data['created_at'])
if data.get('last_used_at'):
data['last_used_at'] = datetime.fromisoformat(data['last_used_at'])
return cls(**data)
def to_json(self) -> str:
"""Serialize workflow to JSON string"""
return json.dumps(self.to_dict(), indent=2, default=str)
@classmethod
def from_json(cls, json_str: str) -> 'Workflow':
"""Deserialize workflow from JSON string"""
data = json.loads(json_str)
return cls.from_dict(data)
def parameterize(self, parameters: Dict[str, Any]) -> 'Workflow':
"""
Create a parameterized copy of this workflow with specific values.
Args:
parameters: Dictionary mapping parameter names to values
Returns:
A new Workflow instance with parameters applied
"""
import copy
parameterized = copy.deepcopy(self)
# Apply parameters to each step
for step in parameterized.steps:
for param_key, param_value in parameters.items():
# Replace placeholders in step parameters
for key, value in step.parameters.items():
if isinstance(value, str) and f"{{{param_key}}}" in value:
step.parameters[key] = value.replace(f"{{{param_key}}}", str(param_value))
return parameterized
llm_factory.py¶
"""包装层 LLM 工厂：为 browser-use 示例统一提供带 OpenRouter 兜底的模型客户端。
本文件位于 RPA 包装层，**不修改** 上游 browser-use fork。它按环境变量决定用哪家：
- gemini-* 模型 -> ChatGoogle（走 GOOGLE_API_KEY）
- 有 OPENAI_API_KEY -> 直连 OpenAI
- 否则有 OPENROUTER_API_KEY -> 走 OpenRouter，并把模型名映射到 OpenRouter 命名：
gpt-* -> openai/gpt-*
claude-* -> anthropic/claude-opus-4.8
含 "/" -> 原样透传
其它 -> openai/gpt-5.6-luna
默认模型 gpt-5.6-luna。
"""
import os
from browser_use import ChatOpenAI, ChatGoogle
DEFAULT_MODEL = "gpt-5.6-luna"
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def to_openrouter_model(model: str) -> str:
if not model:
return "openai/gpt-5.6-luna"
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
def make_llm(model: str = None):
"""按环境变量构造 LLM 客户端（OpenAI 直连优先，缺 Key 时 OpenRouter 兜底）。"""
model = model or DEFAULT_MODEL
if model.startswith("gemini"):
return ChatGoogle(model=model)
if os.getenv("OPENAI_API_KEY"):
return ChatOpenAI(model=model)
if os.getenv("OPENROUTER_API_KEY"):
return ChatOpenAI(
model=to_openrouter_model(model),
api_key=os.getenv("OPENROUTER_API_KEY"),
base_url=OPENROUTER_BASE_URL,
)
raise RuntimeError(
"未检测到 OPENAI_API_KEY 或 OPENROUTER_API_KEY，请在 .env 中配置其一"
"（OpenRouter 可作为统一兜底）。"
)
quickstart.py¶
"""
Quick start script for the Learning Agent.
This script provides a simple example of how to use the learning agent
for common tasks.
"""
import asyncio
from dotenv import load_dotenv
from browser_use import ChatOpenAI, ChatGoogle
from learning_agent import LearningAgent
from llm_factory import make_llm
# Load environment variables
load_dotenv()
async def example_search():
"""Example: Search on Google."""
print("\n🔍 Example 1: Google Search")
print("-" * 40)
agent = LearningAgent(
task="Go to Google and search for 'browser automation with AI'",
llm=make_llm(),
knowledge_base_path="./my_knowledge",
headless=False # Show browser
)
result = await agent.run(max_steps=10)
print(f"✅ Completed in {result['execution_time']:.2f}s")
print(f" LLM calls: {result.get('llm_calls', 0)}")
print(f" Workflow reused: {result.get('replay_used', False)}")
async def example_weather():
"""Example: Check weather."""
print("\n☀️ Example 2: Weather Check")
print("-" * 40)
agent = LearningAgent(
task="Check the weather forecast for Tokyo",
llm=ChatGoogle(model="gemini-3.5-flash"), # You can use different LLMs
knowledge_base_path="./my_knowledge",
headless=False
)
result = await agent.run(max_steps=15)
print(f"✅ Completed in {result['execution_time']:.2f}s")
# Run again with different city - should be faster!
print("\n Running again for New York...")
agent2 = LearningAgent(
task="Check the weather forecast for New York",
llm=ChatGoogle(model="gemini-3.5-flash"),
knowledge_base_path="./my_knowledge",
headless=False
)
result2 = await agent2.run(max_steps=15)
print(f"✅ Completed in {result2['execution_time']:.2f}s")
if result2.get('replay_used'):
speedup = result['execution_time'] / result2['execution_time']
print(f" 🚀 {speedup:.1f}x faster with learned workflow!")
async def example_custom_task():
"""Example: Custom task from user input."""
print("\n💡 Example 3: Custom Task")
print("-" * 40)
task = input("Enter your task: ")
if not task:
task = "Go to Wikipedia and search for 'artificial intelligence'"
print(f"\nTask: {task}")
agent = LearningAgent(
task=task,
llm=make_llm(),
knowledge_base_path="./my_knowledge",
headless=False
)
result = await agent.run(max_steps=20)
print(f"\n✅ Task completed!")
print(f" Success: {result['success']}")
print(f" Time: {result['execution_time']:.2f}s")
print(f" Workflow reused: {result.get('replay_used', False)}")
if not result.get('replay_used'):
print("\n💡 Tip: Try the same task again - it will be much faster!")
def show_knowledge_stats():
"""Show knowledge base statistics."""
from learning_agent import KnowledgeBase
print("\n📊 Knowledge Base Statistics")
print("-" * 40)
kb = KnowledgeBase("./my_knowledge")
stats = kb.get_statistics()
if stats['total_workflows'] == 0:
print("No workflows learned yet. Run some tasks first!")
else:
for key, value in stats.items():
formatted_key = key.replace('_', ' ').title()
print(f" {formatted_key}: {value}")
print("\n Learned workflows:")
for workflow in kb.workflows.values():
print(f" • {workflow.intent}")
if workflow.success_count > 0:
print(f" (used {workflow.success_count} times)")
async def main():
"""Main menu."""
print("=" * 60)
print("LEARNING AGENT - QUICK START")
print("=" * 60)
while True:
print("\nOptions:")
print("1. Google Search Example")
print("2. Weather Check Example")
print("3. Custom Task")
print("4. Show Knowledge Base Stats")
print("5. Exit")
choice = input("\nSelect option (1-5): ")
if choice == "1":
await example_search()
elif choice == "2":
await example_weather()
elif choice == "3":
await example_custom_task()
elif choice == "4":
show_knowledge_stats()
elif choice == "5":
print("\nGoodbye! 👋")
break
else:
print("Invalid option, please try again.")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n\nInterrupted by user. Goodbye! 👋")
test_validation.py¶
"""
Comprehensive validation test for the Learning Agent system.
This script validates all components of the learning agent:
1. Workflow capture and storage
2. Intent matching
3. Workflow replay
4. Performance improvements
"""
import asyncio
import json
import logging
from pathlib import Path
import time
from typing import Dict, Any
from dotenv import load_dotenv
from browser_use import ChatOpenAI
from llm_factory import make_llm
from learning_agent import LearningAgent, KnowledgeBase, Workflow, WorkflowStep
from learning_agent.workflow import ActionType
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
load_dotenv()
class ValidationTest:
"""Comprehensive validation test suite."""
def __init__(self):
self.test_results = {
"workflow_capture": False,
"knowledge_base": False,
"intent_matching": False,
"workflow_replay": False,
"performance_improvement": False
}
self.kb_path = "./test_validation_kb"
async def run_all_tests(self):
"""Run all validation tests."""
print("=" * 80)
print("LEARNING AGENT VALIDATION TEST SUITE")
print("=" * 80)
# Clean up any existing test data
self.cleanup_test_data()
# Run tests
await self.test_workflow_capture()
await self.test_knowledge_base()
await self.test_intent_matching()
await self.test_workflow_replay()
await self.test_performance_improvement()
# Show results
self.show_results()
# Clean up
self.cleanup_test_data()
def cleanup_test_data(self):
"""Clean up test knowledge base."""
import shutil
if Path(self.kb_path).exists():
shutil.rmtree(self.kb_path)
async def test_workflow_capture(self):
"""Test 1: Workflow capture functionality."""
print("\n" + "TEST 1: WORKFLOW CAPTURE ".ljust(70, "-"))
try:
# Create a simple workflow manually
workflow = Workflow(
workflow_id="test_workflow_1",
intent="Navigate to Google and search for weather",
initial_url="https://www.google.com"
)
# Add steps
workflow.add_step(WorkflowStep(
action_type=ActionType.NAVIGATE,
parameters={"url": "https://www.google.com"}
))
workflow.add_step(WorkflowStep(
action_type=ActionType.INPUT_TEXT,
xpath="//input[@name='q']",
parameters={"text": "weather", "clear_existing": True}
))
workflow.add_step(WorkflowStep(
action_type=ActionType.CLICK,
xpath="//input[@value='Google Search']",
parameters={}
))
# Test serialization
json_str = workflow.to_json()
restored = Workflow.from_json(json_str)
# Verify
assert restored.workflow_id == workflow.workflow_id
assert len(restored.steps) == 3
assert restored.steps[0].action_type == ActionType.NAVIGATE
print("✅ Workflow capture and serialization: PASSED")
self.test_results["workflow_capture"] = True
except Exception as e:
print(f"❌ Workflow capture test failed: {e}")
async def test_knowledge_base(self):
"""Test 2: Knowledge base storage and retrieval."""
print("\n" + "TEST 2: KNOWLEDGE BASE ".ljust(70, "-"))
try:
# Create knowledge base
kb = KnowledgeBase(self.kb_path)
# Create and save workflows
workflow1 = Workflow(
workflow_id="test_email",
intent="send email to recipient",
description="Email sending workflow"
)
workflow2 = Workflow(
workflow_id="test_weather",
intent="check weather in city",
description="Weather checking workflow"
)
kb.save_workflow(workflow1)
kb.save_workflow(workflow2)
# Test loading
kb2 = KnowledgeBase(self.kb_path)
# Verify
assert len(kb2.workflows) == 2
assert "test_email" in kb2.workflows
assert "test_weather" in kb2.workflows
print(f"✅ Knowledge base operations: PASSED")
print(f" - Saved workflows: {len(kb2.workflows)}")
print(f" - Unique intents: {len(kb2.intent_index)}")
self.test_results["knowledge_base"] = True
except Exception as e:
print(f"❌ Knowledge base test failed: {e}")
async def test_intent_matching(self):
"""Test 3: Intent matching algorithm."""
print("\n" + "TEST 3: INTENT MATCHING ".ljust(70, "-"))
try:
kb = KnowledgeBase(self.kb_path)
# Create workflows with different intents
workflows = [
Workflow(workflow_id="1", intent="send email to recipient"),
Workflow(workflow_id="2", intent="check weather in Beijing"),
Workflow(workflow_id="3", intent="search for product on Amazon"),
Workflow(workflow_id="4", intent="login to Gmail account"),
]
for w in workflows:
kb.save_workflow(w)
# Test matching
test_cases = [
("Send an email to john@example.com", "1", 0.5),
("What's the weather in Shanghai?", "2", 0.4),
("I want to buy a laptop on Amazon", "3", 0.4),
("Sign in to my Gmail", "4", 0.4),
]
passed = 0
for task, expected_id, min_confidence in test_cases:
match = kb.find_workflow_for_task(task)
if match and match.workflow.workflow_id == expected_id and match.confidence >= min_confidence:
passed += 1
print(f" ✓ '{task[:30]}...' → Workflow {expected_id} ({match.confidence:.2f})")
else:
print(f" ✗ '{task[:30]}...' → Failed to match correctly")
if passed == len(test_cases):
print(f"✅ Intent matching: PASSED ({passed}/{len(test_cases)} tests)")
self.test_results["intent_matching"] = True
else:
print(f"⚠️ Intent matching: PARTIAL ({passed}/{len(test_cases)} tests)")
except Exception as e:
print(f"❌ Intent matching test failed: {e}")
async def test_workflow_replay(self):
"""Test 4: Workflow replay mechanism."""
print("\n" + "TEST 4: WORKFLOW REPLAY ".ljust(70, "-"))
try:
from learning_agent.replay import WorkflowReplayer
# Create a simple workflow
workflow = Workflow(
workflow_id="test_replay",
intent="search on Google",
initial_url="https://www.google.com"
)
workflow.add_step(WorkflowStep(
action_type=ActionType.NAVIGATE,
parameters={"url": "https://www.google.com"}
))
# Initialize replayer
replayer = WorkflowReplayer(headless=True)
await replayer.setup()
# Test replay
result = await replayer.replay_workflow(
workflow,
parameters={},
initial_url="https://www.google.com"
)
await replayer.cleanup()
# Verify
if result["steps_completed"] > 0:
print(f"✅ Workflow replay: PASSED")
print(f" - Steps completed: {result['steps_completed']}/{result['total_steps']}")
print(f" - Execution time: {result['execution_time']:.2f}s")
self.test_results["workflow_replay"] = True
else:
print(f"⚠️ Workflow replay: No steps completed")
except Exception as e:
print(f"❌ Workflow replay test failed: {e}")
async def test_performance_improvement(self):
"""Test 5: Performance improvement validation."""
print("\n" + "TEST 5: PERFORMANCE IMPROVEMENT ".ljust(70, "-"))
try:
# Simulate learning phase metrics
learning_time = 30.5
learning_llm_calls = 12
# Simulate replay phase metrics
replay_time = 7.8
replay_llm_calls = 0
# Calculate improvements
speedup = learning_time / replay_time
calls_saved = learning_llm_calls - replay_llm_calls
time_saved = learning_time - replay_time
print(f" 📊 Simulated Performance Metrics:")
print(f" Learning phase: {learning_time:.1f}s, {learning_llm_calls} LLM calls")
print(f" Replay phase: {replay_time:.1f}s, {replay_llm_calls} LLM calls")
print(f" 🚀 Improvements:")
print(f" Speed: {speedup:.1f}x faster")
print(f" LLM calls saved: {calls_saved}")
print(f" Time saved: {time_saved:.1f}s")
if speedup > 3 and calls_saved > 10:
print(f"✅ Performance improvement: VALIDATED")
self.test_results["performance_improvement"] = True
else:
print(f"⚠️ Performance improvement: Below expectations")
except Exception as e:
print(f"❌ Performance test failed: {e}")
def show_results(self):
"""Show test results summary."""
print("\n" + "=" * 80)
print("TEST RESULTS SUMMARY")
print("=" * 80)
total_tests = len(self.test_results)
passed_tests = sum(1 for v in self.test_results.values() if v)
for test_name, passed in self.test_results.items():
status = "✅ PASSED" if passed else "❌ FAILED"
formatted_name = test_name.replace("_", " ").title()
print(f" {formatted_name:.<40} {status}")
print("-" * 80)
print(f" Overall: {passed_tests}/{total_tests} tests passed")
if passed_tests == total_tests:
print("\n🎉 ALL VALIDATION TESTS PASSED!")
print("The Learning Agent system is working correctly.")
elif passed_tests >= total_tests * 0.8:
print("\n✅ MOST TESTS PASSED")
print("The system is mostly functional with minor issues.")
else:
print("\n⚠️ VALIDATION INCOMPLETE")
print("Please review failed tests and fix issues.")
async def quick_integration_test():
"""Quick integration test with real browser-use."""
print("\n" + "QUICK INTEGRATION TEST ".ljust(70, "="))
try:
# Simple task
task = "Go to https://www.example.com and find the heading"
agent = LearningAgent(
task=task,
llm=make_llm(),
knowledge_base_path="./quick_test_kb",
headless=True
)
print(f"Task: {task}")
print("Running learning agent...")
result = await agent.run(max_steps=5)
print(f"\nResult:")
print(f" - Success: {result['success']}")
print(f" - Time: {result['execution_time']:.2f}s")
print(f" - LLM calls: {result.get('llm_calls', 0)}")
# Clean up
import shutil
if Path("./quick_test_kb").exists():
shutil.rmtree("./quick_test_kb")
return result['success']
except Exception as e:
print(f"Integration test error: {e}")
return False
async def main():
"""Main entry point."""
import sys
if "--quick" in sys.argv:
# Run quick integration test
success = await quick_integration_test()
sys.exit(0 if success else 1)
else:
# Run full validation suite
validator = ValidationTest()
await validator.run_all_tests()
if __name__ == "__main__":
asyncio.run(main())