别再手动爬数据了！Skyvern用AI操控浏览器，完成你想都不敢想的自动化任务
为什么值得关注 / 项目简介
传统浏览器自动化的困境
在 Skyvern 出现之前，如果你想实现浏览器自动化，通常会面临以下挑战：
- Selenium/Playwright 需要编写大量 xpath 选择器，一旦网站改版就全部失效
- 反爬机制越来越严格，验证码、IP限制让数据采集难上加贵
- 复杂的多步骤表单操作（比如需要先登录、再查询、最后导出）需要手动维护大量状态
- 无法处理动态加载的内容，页面交互逻辑需要写大量等待代码
Skyvern带来了什么改变
Skyvern 是一个基于 AI 的浏览器自动化框架，它的核心思想是：让大语言模型直接理解和操控网页。你只需要用自然语言描述你想做什么，Skyvern 就会：
- 自动分析网页结构，理解各个元素的作用
- 规划完成任务的步骤序列
- 像真人一样与网页交互——点击、输入、滚动、等待
- 智能处理验证码、弹窗、加载等待等边界情况
- 自动提取结构化的目标数据
想象一下，你告诉 Skyvern：「帮我登录这个网站，搜索关键词’人工智能’，然后把所有搜索结果的标题和链接整理成一个表格」，它就能自动完成这一切——不需要写任何 xpath，不需要分析网站结构，只需要描述你的需求。
这个项目目前在 GitHub 上已经获得了大量关注，它重新定义了浏览器自动化的可能性边界。
环境搭建 / Getting Started
系统要求
在开始之前，确保你的系统满足以下要求：
- Python 3.10 或更高版本
- 至少 4GB 可用内存
- 稳定的网络连接（需要访问外部网站）
- 推荐使用 Linux 或 macOS 系统（Windows 也支持但可能需要额外配置）
使用Docker快速启动（推荐）
Docker 是最简单的方式，可以避免依赖冲突问题：
# 拉取最新的 Skyvern 镜像
docker pull skyvern/skyvern:latest
# 运行容器
docker run -p 8080:8080 \
-e OPENAI_API_KEY="your-api-key-here" \
-e SKYVERNET_DB_URL="postgresql://user:pass@localhost:5432/skyvern" \
skyvern/skyvern:latest
服务启动后，可以通过 http://localhost:8080 访问 Web 界面。
本地安装
如果你更喜欢本地开发环境，可以按以下步骤操作：
# 创建虚拟环境（推荐使用 uv 或 conda）
python -m venv skyvern-env
# 激活虚拟环境
source skyvern-env/bin/activate # Linux/macOS
# skyvern-env\Scripts\activate # Windows
# 克隆仓库
git clone https://github.com/Skyvern-AI/skyvern.git
cd skyvern
# 安装依赖
pip install -e .
# 或者使用 uv 安装（更快）
uv pip install -e .
必需的环境变量
Skyvern 需要配置一些环境变量才能正常工作：
# 创建环境变量配置文件
cat > .env << 'EOF'
# LLM API 配置（必需）
OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxx"
# 数据库配置（使用 SQLite 用于开发）
SKYVERNET_DB_URL="sqlite:///./skyvern.db"
# 可选配置
SKYVERNET_TEMP_DIR="/tmp/skyvern"
LOG_LEVEL="INFO"
# 如果使用 Playwright 作为浏览器后端
SKYVERNET_BROWSER_TYPE="playwright"
EOF
验证安装
安装完成后，运行以下命令验证配置是否正确：
# 验证脚本 - 保存为 verify_installation.py
"""
Skyvern 安装验证脚本
用于检查所有依赖和配置是否正确
"""
import sys
def verify_python_version():
"""检查 Python 版本"""
version = sys.version_info
if version.major < 3 or (version.major == 3 and version.minor < 10):
print(f"✗ Python 版本过低: {version.major}.{version.minor}")
print(" Skyvern 需要 Python 3.10 或更高版本")
return False
print(f"✓ Python 版本: {version.major}.{version.minor}.{version.micro}")
return True
def verify_dependencies():
"""检查关键依赖包"""
required_packages = [
'fastapi',
'playwright',
'openai',
'httpx',
'sqlalchemy',
'pydantic'
]
missing = []
for package in required_packages:
try:
__import__(package)
print(f"✓ {package} 已安装")
except ImportError:
print(f"✗ {package} 缺失")
missing.append(package)
return len(missing) == 0
def verify_api_key():
"""检查 API Key 配置"""
import os
api_key = os.environ.get('OPENAI_API_KEY', '')
if not api_key:
print("✗ OPENAI_API_KEY 未设置")
print(" 请设置环境变量: export OPENAI_API_KEY='your-key'")
return False
if not api_key.startswith('sk-'):
print("✗ API Key 格式可能不正确（应以 sk- 开头）")
return False
print("✓ OPENAI_API_KEY 已配置")
return True
if __name__ == "__main__":
print("=" * 50)
print("Skyvern 安装验证")
print("=" * 50)
results = [
verify_python_version(),
verify_dependencies(),
verify_api_key()
]
print("=" * 50)
if all(results):
print("✓ 所有检查通过！可以开始使用 Skyvern")
else:
print("✗ 部分检查失败，请修复上述问题")
print("=" * 50)
运行验证脚本：
python verify_installation.py
安装 Playwright 浏览器
Skyvern 使用 Playwright 进行浏览器控制，需要安装浏览器：
# 安装 Playwright 浏览器（使用 uvx 或 npx）
uvx playwright install chromium
# 或者使用 npm
npx playwright install chromium
# 如果需要其他浏览器
npx playwright install firefox
npx playwright install webkit
核心概念与架构
Skyvern 的工作原理
Skyvern 的架构设计非常精妙，理解它的工作流程能帮助你更好地使用这个工具：
用户需求（自然语言）
↓
┌─────────────────────────────────────┐
│ LLM 分析层 │
│ 理解任务意图 → 规划执行步骤 │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ 浏览器控制层 │
│ Playwright/Selenium 操作浏览器 │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ 元素识别层 │
│ OCR + 视觉模型 定位交互元素 │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ 数据提取层 │
│ 结构化输出 → JSON/CSV/数据库 │
└─────────────────────────────────────┘
关键组件详解
1. Task（任务）
Task 是 Skyvern 的核心执行单元，每个自动化任务对应一个 Task 实例：
# Task 的基本结构
task_config = {
"url": "https://example.com/search", # 目标网页
"instruction": "搜索'Python教程'，提取前10个结果的标题和链接",
"webhook_url": "https://your-server.com/callback", # 完成回调（可选）
"max_steps": 20, # 最大执行步数
"提取字段": ["title", "url", "description"] # 期望提取的字段
}
2. Agent（代理）
Agent 负责理解任务并规划执行策略：
"""
Agent 的决策流程
当遇到需要决策的页面时，Agent 会分析当前状态并决定下一步操作
"""
class AgentDecisionExample:
"""
模拟 Agent 的决策过程
"""
def analyze_page_state(self, page_content):
"""分析页面当前状态"""
# Agent 会识别：
# - 当前在哪个页面
# - 有哪些可交互元素
# - 是否出现了预期外的元素（弹窗、验证码等）
pass
def plan_next_action(self, task_goal, current_state):
"""规划下一步动作"""
# Agent 可能的决策：
# - 点击某个按钮
# - 填写表单
# - 滚动页面
# - 等待加载
# - 提取数据
pass
3. Browser（浏览器）
Browser 组件封装了实际的浏览器操作：
"""
Browser 组件支持的浏览器操作
"""
# 可用的浏览器操作示例
browser_operations = {
# 基础操作
"navigate": "打开指定URL",
"click": "点击元素",
"type": "输入文本",
"hover": "鼠标悬停",
"scroll": "滚动页面",
# 等待操作
"wait_for_selector": "等待元素出现",
"wait_for_navigation": "等待页面导航完成",
"wait_for_timeout": "等待指定时间",
# 特殊操作
"screenshot": "截图",
"extract_content": "提取页面内容",
"handle_popup": "处理弹窗"
}
实战教程：从入门到精通
第一个示例：自动搜索并提取数据
让我们从最简单的例子开始——使用 Skyvern 自动访问 Google 搜索某个关键词，并提取搜索结果：
"""
第一个 Skyvern 示例：自动化网页搜索与数据提取
保存为: example_01_basic_search.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def main():
"""主函数：执行自动化搜索任务"""
# 初始化 Forge 实例
# Forge 是 Skyvern 的核心编排器
forge = Forge()
# 创建任务配置
search_task = Task(
url="https://www.google.com",
instruction="""
1. 在搜索框中输入 "Python 自动化测试"
2. 点击搜索按钮或按回车键
3. 等待搜索结果加载完成
4. 提取前5个搜索结果的：
- 标题（title）
- 链接（url）
- 简要描述（snippet）
5. 将结果以 JSON 格式返回
""",
max_steps=15, # 最多执行15步，避免无限循环
extraction_fields=[
{"name": "title", "type": "text"},
{"name": "url", "type": "link"},
{"name": "snippet", "type": "text"}
]
)
# 提交任务并等待完成
task_result = await forge.run_task(search_task)
# 输出结果
print("任务执行完成！")
print(f"任务状态: {task_result.status}")
print(f"提取的数据: {task_result.data}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
第二个示例：登录并填写表单
这个例子展示如何处理需要登录的复杂流程：
"""
第二个示例：登录网站并填写表单
保存为: example_02_login_form.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def login_and_submit_form():
"""演示登录和表单填写流程"""
forge = Forge()
# 定义一个需要登录的表单填写任务
form_task = Task(
url="https://example.com/login",
instruction="""
这个任务演示完整的登录和数据提交流程：
第一步：登录
1. 找到用户名输入框，输入 "test_user"
2. 找到密码输入框，输入 "test_password"
3. 点击登录按钮
第二步：导航到表单页面
4. 登录成功后，点击"新建表单"按钮
5. 等待表单页面加载完成
第三步：填写表单
6. 在"姓名"字段输入 "张三"
7. 在"邮箱"字段输入 "zhangsan@example.com"
8. 在"备注"字段输入 "这是测试备注"
9. 点击提交按钮
第四步：验证提交结果
10. 确认页面显示提交成功的提示
11. 提取提交后的确认信息
""",
max_steps=25,
# 如果目标网站需要登录，可以提供凭据
credentials={
"username": "test_user",
"password": "test_password"
}
)
result = await forge.run_task(form_task)
return result
# 运行示例
if __name__ == "__main__":
import asyncio
result = asyncio.run(login_and_submit_form())
print(result)
第三个示例：批量数据采集
实际工作中经常需要采集多个页面的数据：
"""
第三个示例：批量数据采集任务
保存为: example_03_batch_scraping.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
import asyncio
async def batch_scraping_example():
"""批量采集多个页面的数据"""
forge = Forge()
# 要采集的页面列表
target_urls = [
"https://news.ycombinator.com/news",
"https://news.ycombinator.com/news?p=2",
"https://news.ycombinator.com/news?p=3",
]
# 采集任务配置
extraction_instruction = """
1. 等待页面完全加载
2. 滚动页面加载更多内容（如果需要）
3. 提取页面上所有新闻条目的：
- 标题（title）
- 链接（url）
- 点数（score）
- 评论数（comments）
4. 返回结构化的 JSON 数据
"""
all_results = []
# 遍历每个 URL 进行采集
for page_num, url in enumerate(target_urls, 1):
print(f"正在采集第 {page_num} 页: {url}")
task = Task(
url=url,
instruction=extraction_instruction,
max_steps=10,
extraction_fields=[
{"name": "title", "type": "text"},
{"name": "url", "type": "link"},
{"name": "score", "type": "number"},
{"name": "comments", "type": "number"}
]
)
result = await forge.run_task(task)
if result.status == "completed":
all_results.extend(result.data)
print(f"✓ 第 {page_num} 页采集完成，获取 {len(result.data)} 条数据")
else:
print(f"✗ 第 {page_num} 页采集失败: {result.error}")
# 合并所有结果
print(f"\n总计采集到 {len(all_results)} 条数据")
return all_results
if __name__ == "__main__":
results = asyncio.run(batch_scraping_example())
# 可以将结果保存为 JSON 文件
import json
with open("scraped_data.json", "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print("数据已保存到 scraped_data.json")
第四个示例：处理验证码和弹窗
实际场景中经常会遇到各种障碍，Skyvern 内置了智能处理机制：
"""
第四个示例：处理验证码和其他障碍
保存为: example_04_handling_obstacles.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def handle_obstacles():
"""演示如何处理各种网页障碍"""
forge = Forge()
task = Task(
url="https://example-site-with-captcha.com",
instruction="""
任务要求：完成整个数据查询流程
正常流程：
1. 导航到目标页面
2. 如果出现 Cookie 同意弹窗，点击"接受"或"同意全部"
3. 如果出现登录弹窗，说明这是预期的登录流程
4. 如果出现验证码：
- 尝试识别并填写验证码
- 如果无法自动处理，记录下来并尝试刷新重试
5. 完成主要的业务操作
6. 如果遇到任何无法处理的弹窗，尝试点击关闭或取消按钮
异常处理策略：
- 超时重试：最多重试3次
- 元素找不到：尝试备用选择器或滚动到可见区域
- 页面卡住：截图记录当前状态并终止任务
""",
max_steps=30,
# 配置处理策略
obstacle_handling={
"captcha": "auto", # 自动处理验证码
"popup": "auto", # 自动关闭弹窗
"cookie_banner": "accept" # 自动接受 Cookie
}
)
return await forge.run_task(task)
if __name__ == "__main__":
import asyncio
result = asyncio.run(handle_obstacles())
print(f"任务状态: {result.status}")
第五个示例：与数据库集成
Skyvern 可以直接将采集的数据存入数据库：
"""
第五个示例：数据采集并存入数据库
保存为: example_05_database_integration.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import json
# 数据库配置
DATABASE_URL = "sqlite:///./skyvern_data.db"
engine = create_engine(DATABASE_URL)
# 定义数据库表结构
from sqlalchemy import Column, Integer, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class ScrapedArticle(Base):
"""存储采集的文章数据"""
__tablename__ = "articles"
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(500), nullable=False)
url = Column(String(1000), unique=True)
snippet = Column(Text)
source = Column(String(100))
scraped_at = Column(DateTime, default=datetime.utcnow)
def __repr__(self):
return f"<Article(title='{self.title}', url='{self.url}')>"
# 创建表
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
async def scrape_and_store():
"""采集数据并存储到数据库"""
forge = Forge()
session = Session()
task = Task(
url="https://example-news-site.com/tech",
instruction="""
1. 提取页面上所有科技新闻的标题、链接和摘要
2. 返回 JSON 格式的数据列表
""",
max_steps=15
)
result = await forge.run_task(task)
if result.status == "completed" and result.data:
# 将结果写入数据库
for item in result.data:
article = ScrapedArticle(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("snippet", ""),
source="example-news-site.com"
)
session.add(article)
session.commit()
print(f"成功存储 {len(result.data)} 条记录到数据库")
session.close()
return result
if __name__ == "__main__":
import asyncio
asyncio.run(scrape_and_store())
# 验证数据
session = Session()
articles = session.query(ScrapedArticle).limit(5).all()
for article in articles:
print(f"- {article.title}")
session.close()
常见使用场景
场景一：竞品价格监控
"""
使用场景示例：电商竞品价格监控
保存为: use_case_price_monitoring.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
from datetime import datetime
import json
async def monitor_competitor_prices():
"""监控竞品价格变动"""
forge = Forge()
# 要监控的商品列表
products = [
{
"name": "iPhone 15",
"urls": [
"https://www.jd.com/product/iphone15.html",
"https://www.amazon.cn/dp/iphone15",
"https://www.taobao.com/search?q=iphone15"
]
},
{
"name": "MacBook Pro",
"urls": [
"https://www.jd.com/product/macbookpro.html",
"https://www.amazon.cn/dp/macbookpro"
]
}
]
results = []
for product in products:
product_name = product["name"]
for url in product["urls"]:
task = Task(
url=url,
instruction=f"""
提取 {product_name} 的以下信息：
1. 当前价格（注意单位：元/美元）
2. 是否在促销中（discount）
3. 促销信息（如有）
4. 库存状态
5. 评价评分
返回 JSON 格式结果，包含价格历史记录
""",
max_steps=10
)
result = await forge.run_task(task)
if result.status == "completed":
results.append({
"product": product_name,
"source": url,
"price_data": result.data,
"scraped_at": datetime.now().isoformat()
})
# 保存监控结果
with open(f"price_monitor_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
return results
if __name__ == "__main__":
import asyncio
asyncio.run(monitor_competitor_prices())
场景二：社交媒体数据采集
"""
使用场景示例：社交媒体数据分析
保存为: use_case_social_media.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def analyze_social_media():
"""分析社交媒体上的热门话题"""
forge = Forge()
task = Task(
url="https://twitter.com/explore/tabs/trending",
instruction="""
1. 等待页面完全加载
2. 提取当前热门话题列表
3. 对每个话题，获取：
- 话题标题
- 讨论数量
- 相关的标签
4. 按讨论数量排序
5. 返回前10个热门话题
""",
max_steps=15,
# 如果需要登录，配置凭据
# credentials={"username": "...", "password": "..."}
)
result = await forge.run_task(task)
return result
# 类似的，可以采集：
# - 微博热搜榜
# - 知乎热榜
# - Reddit 热门帖子
# - LinkedIn 职位信息
场景三：自动化测试
"""
使用场景示例：Web 应用自动化测试
保存为: use_case_automated_testing.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
import asyncio
async def automated_web_testing():
"""使用 Skyvern 进行 Web 应用端到端测试"""
forge = Forge()
test_cases = [
{
"name": "用户注册流程测试",
"url": "https://app.example.com/register",
"steps": [
"填写用户名",
"填写邮箱",
"填写密码",
"确认密码",
"点击注册按钮",
"验证注册成功提示"
]
},
{
"name": "用户登录测试",
"url": "https://app.example.com/login",
"steps": [
"填写用户名",
"填写密码",
"点击登录按钮",
"验证登录后跳转"
]
},
{
"name": "商品搜索测试",
"url": "https://shop.example.com",
"steps": [
"在搜索框输入商品名称",
"点击搜索",
"验证搜索结果"
]
}
]
test_results = []
for test_case in test_cases:
print(f"执行测试: {test_case['name']}")
task = Task(
url=test_case["url"],
instruction=f"""
执行以下测试步骤：
{' '.join([f'{i+1}. {step}' for i, step in enumerate(test_case['steps'])])}
验证每个步骤是否正常执行
如果发现问题，详细记录错误信息
""",
max_steps=20
)
result = await forge.run_task(task)
test_results.append({
"test_name": test_case["name"],
"status": result.status,
"passed": result.status == "completed",
"error": result.error if result.status != "completed" else None
})
# 输出测试报告
print("\n" + "=" * 50)
print("测试报告")
print("=" * 50)
passed = sum(1 for r in test_results if r["passed"])
failed = len(test_results) - passed
for result in test_results:
status_icon = "✓" if result["passed"] else "✗"
print(f"{status_icon} {result['test_name']}")
if result["error"]:
print(f" 错误: {result['error']}")
print(f"\n总计: {passed} 通过, {failed} 失败")
return test_results
if __name__ == "__main__":
asyncio.run(automated_web_testing())
场景四：学术文献调研
"""
使用场景示例：学术文献自动化调研
保存为: use_case_academic_research.py
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def academic_paper_research():
"""自动化学术文献调研"""
forge = Forge()
# 在 Google Scholar 上搜索相关论文
task = Task(
url="https://scholar.google.com/scholar?q=machine+learning+optimization+2024",
instruction="""
1. 等待搜索结果加载完成
2. 提取前20篇论文的信息：
- 论文标题
- 作者列表
- 发表期刊/会议
- 发表年份
- 引用次数
- 摘要（如果可见）
3. 按引用次数降序排列
4. 对于每篇论文，检查是否能访问 PDF 全文
""",
max_steps=15
)
result = await forge.run_task(task)
return result
# 同样可以用于：
# - arXiv 论文搜索
# - PubMed 医学文献检索
# - IEEE Xplore 技术文献
# - CNKI 中国知网
高级技巧与最佳实践
技巧一：优化任务指令
好的指令能显著提高任务成功率：
"""
最佳实践：如何写好任务指令
"""
# 不好的指令示例（太模糊）
BAD_INSTRUCTION = """
去那个网站看看有没有什么有用的信息
"""
# 好的指令示例（清晰、具体）
GOOD_INSTRUCTION = """
执行以下操作序列：
第一步：导航和搜索
1. 打开 https://example-ecommerce.com
2. 在顶部搜索框中输入 "无线蓝牙耳机"
3. 点击搜索按钮或按回车键
第二步：筛选和导航
4. 等待搜索结果页面加载完成
5. 点击"价格从低到高"排序
6. 点击"有货"筛选条件
第三步：数据提取
7. 提取前10个商品的以下信息：
- 商品名称
- 当前价格
- 原价（如有）
- 评分（5星制）
- 评论数量
- 商品链接
第四步：输出格式
8. 将结果整理成 JSON 格式，每个商品包含上述所有字段
9. 确保返回的是有效的数据，不要返回空值或占位符
"""
# 更多优化建议：
OPTIMIZATION_TIPS = """
1. 分解大任务为小步骤
- AI 更擅长完成一系列简单的步骤
- 而不是一个模糊的大型目标
2. 明确输出格式
- 指定返回 JSON、CSV 还是其他格式
- 列出所有需要的字段名称
3. 设置边界条件
- 最多处理多少条数据
- 遇到问题时如何处理
4. 提供上下文信息
- 如果需要登录，提供凭据
- 如果有特殊要求，提前说明
"""
技巧二：提高稳定性
"""
高级技巧：提高任务执行的稳定性
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def stable_task_execution():
"""稳定执行任务的配置建议"""
forge = Forge()
task = Task(
url="https://example-dynamic-site.com",
instruction="""
执行数据采集任务
""",
max_steps=15,
# 重试配置
retry={
"max_attempts": 3, # 最多重试3次
"retry_on_timeout": True, # 超时后重试
"retry_on_error": True # 出错后重试
},
# 等待策略
wait_strategy={
"page_load": "network_idle", # 等待网络空闲
"element_appear": 5, # 元素出现最多等5秒
"navigation": 10 # 导航最多等10秒
},
# 浏览器配置
browser_config={
"viewport": {"width": 1920, "height": 1080},
"user_agent": "Mozilla/5.0 ...",
"headless": False, # 开发时可设为 False 便于调试
"slow_mo": 100 # 操作延迟（毫秒），便于观察
}
)
return await forge.run_task(task)
技巧三：处理复杂页面
"""
高级技巧：处理复杂动态页面
"""
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def handle_complex_pages():
"""处理需要滚动、加载更多的页面"""
forge = Forge()
task = Task(
url="https://example-infinite-scroll.com/feed",
instruction="""
这个页面采用无限滚动加载机制，需要特殊处理：
1. 初始加载
- 打开页面，等待初始内容加载
- 提取当前可见的前20条内容
2. 滚动加载（重复5次）
- 滚动到页面底部
- 等待3秒让新内容加载
- 提取新加载的20条内容
- 记录已提取的总数
3. 数据提取
- 内容类型：帖子
- 提取字段：
* 帖子标题/内容
* 作者
* 发布时间
* 点赞数
* 评论数
4. 停止条件
- 已提取100条数据后停止
- 或滚动5次后仍未加载新内容时停止
5. 输出
- 返回所有提取的数据
- 标注每条数据的加载轮次
""",
max_steps=40 # 需要更多步骤来处理滚动
)
return await forge.run_task(task)
技巧四：并发执行多个任务
"""
高级技巧：并发执行提高效率
"""
import asyncio
from skyvern.forge import Forge
from skyvern.forge.sdk.schemas.task import Task
async def concurrent_task_execution():
"""同时执行多个任务"""
# 创建多个 Forge 实例（每个实例管理独立的浏览器）
forge_1 = Forge()
forge_2 = Forge()
forge_3 = Forge()
# 定义不同的任务
tasks = [
Task(url="https://site1.com/data", instruction="提取数据1"),
Task(url="https://site2.com/data", instruction="提取数据2"),
Task(url="https://site3.com/data", instruction="提取数据3"),
]
# 为每个 Forge 实例分配一个任务
forge_task_pairs = [
(forge_1, tasks[0]),
(forge_2, tasks[1]),
(forge_3, tasks[2]),
]
# 并发执行
async def run_single_task(forge, task):
return await forge.run_task(task)
# 使用 asyncio.gather 并发运行所有任务
results = await asyncio.gather(
*[run_single_task(forge, task) for forge, task in forge_task_pairs]
)
for i, result in enumerate(results):
print(f"任务 {i+1} 完成: {result.status}")
return results
# 运行并发任务
if __name__ == "__main__":
results = asyncio.run(concurrent_task_execution())
Web API 使用指南
Skyvern 提供了完整的 REST API，可以通过 HTTP 请求调用：
启动 API 服务
# 启动 API 服务
uvicorn skyvern.forge.api.app:app --host 0.0.0.0 --port 8080
API 端点
1. 创建任务
# 创建新任务
curl -X POST http://localhost:8080/api/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"url": "https://example.com",
"instruction": "提取页面标题和所有链接",
"max_steps": 10
}'
2. 获取任务状态
# 查看任务状态和结果
curl http://localhost:8080/api/v1/tasks/{task_id}
3. 列出所有任务
# 获取任务列表
curl "http://localhost:8080/api/v1/tasks?status=completed&limit=10"
Python API 客户端
"""
使用 Python 客户端调用 Skyvern API
"""
import httpx
import asyncio
class SkyvernAPIClient:
"""Skyvern API 客户端封装"""
def __init__(self, base_url: str = "http://localhost:8080"):
self.base_url = base_url
self.client = httpx.AsyncClient(timeout=60.0)
async def create_task(
self,
url: str,
instruction: str,
max_steps: int = 15
) -> dict:
"""创建新任务"""
response = await self.client.post(
f"{self.base_url}/api/v1/tasks",
json={
"url": url,
"instruction": instruction,
"max_steps": max_steps
}
)
response.raise_for_status()
return response.json()
async def get_task(self, task_id: str) -> dict:
"""获取任务详情"""
response = await self.client.get(
f"{self.base_url}/api/v1/tasks/{task_id}"
)
response.raise_for_status()
return response.json()
async def list_tasks(
self,
status: str = None,
limit: int = 10
) -> dict:
"""列出任务"""
params = {"limit": limit}
if status:
params["status"] = status
response = await self.client.get(
f"{self.base_url}/api/v1/tasks",
params=params
)
response.raise_for_status()
return response.json()
async def wait_for_completion(self, task_id: str, poll_interval: int = 2) -> dict:
"""等待任务完成"""
import time
while True:
task = await self.get_task(task_id)
status = task.get("status")
if status in ["completed", "failed", "cancelled"]:
return task
print(f"任务进行中... 当前状态: {status}")
time.sleep(poll_interval)
async def close(self):
"""关闭客户端"""
await self.client.aclose()
# 使用示例
async def api_usage_example():
"""API 使用示例"""
client = SkyvernAPIClient()
try:
# 创建任务
task = await client.create_task(
url="https://news.ycombinator.com",
instruction="提取首页前10条新闻的标题和链接"
)
task_id = task["task_id"]
print(f"任务已创建: {task_id}")
# 等待完成
result = await client.wait_for_completion(task_id)
# 处理结果
if result["status"] == "completed":
print(f"任务完成！")
print(f"提取的数据: {result['data']}")
else:
print(f"任务失败: {result.get('error')}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(api_usage_example())
常见问题与解决方案
Q1: 任务执行超时怎么办？
"""
问题：任务执行时间过长导致超时
解决方案：
"""
# 方案1：增加超时时间
task = Task(
url="https://slow-loading-site.com",
instruction="采集数据",
max_steps=30 # 增加步数限制
)
# 方案2：优化指令，减少不必要的步骤
optimized_instruction = """
优化后的指令：
1. 直接导航到目标数据页面（而不是通过首页逐层点击）
2. 只提取必要的字段，减少处理步骤
3. 设置合理的超时时间
"""
# 方案3：分批处理
async def batch_processing():
"""将大任务拆分为多个小任务"""
# 不是一次性提取1000条数据
# 而是分10次，每次提取100条
pass
Q2: 如何处理需要登录的网站？
"""
问题：目标网站需要登录才能访问
解决方案：
"""
# 方案1：提供登录凭据
task = Task(
url="https://member-only-site.com/data",
instruction="采集数据",
credentials={
"username": "your_username",
"password": "your_password"
}
)
# 方案2：预先处理登录流程
async def login_first():
"""分步骤处理登录"""
forge = Forge()
# 第一步：登录
login_task = Task(
url="https://site.com/login",
instruction="""
1. 输入用户名和密码
2. 点击登录
3. 等待登录成功
4. 截图确认当前状态
""",
max_steps=10
)
login_result = await forge.run_task(login_task)
# 获取登录后的 cookies
cookies = login_result.cookies
# 第二步：使用 cookies 访问目标页面
data_task = Task(
url="https://site.com/protected/data",
instruction="采集数据",
cookies=cookies # 复用登录 cookies
)
return await forge.run_task(data_task)
# 方案3：使用 session 管理
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch()
context = browser.new_context()
page = context.new_page()
# 手动完成登录
page.goto("https://site.com/login")
page.fill("#username", "user")
page.fill("#password", "pass")
page.click("#login-btn")
page.wait_for_url("**/dashboard")
# 保存 cookies
cookies = context.cookies()
browser.close()
# 后续任务使用保存的 cookies
task = Task(
url="https://site.com/protected/data",
instruction="采集数据",
cookies=cookies
)
Q3: 遇到验证码怎么处理？
"""
问题：页面包含验证码阻止操作
解决方案：
"""
# Skyvern 内置了一些验证码处理策略
task = Task(
url="https://site-with-captcha.com",
instruction="完成任务",
# 验证码处理策略
captcha_handling={
# 策略1：自动识别（需要额外的验证码识别服务）
"strategy": "auto",
"ocr_provider": "your-ocr-api", # 如 2Captcha, Anti-Captcha
# 策略2：人工介入
"strategy": "manual",
"notify_webhook": "https://your-server.com/captcha-alert",
# 策略3：跳过（如果验证码是关键步骤，任务会失败）
"strategy": "skip"
}
)
# 对于简单的图片验证码，Skyvern 会尝试 OCR 识别
# 对于复杂的验证码，建议使用策略2：人工介入模式
Q4: 如何处理动态加载的内容？
"""
问题：页面内容通过 JavaScript 动态加载
解决方案：
"""
task = Task(
url="https://spa-example.com",
instruction="""
这个页面使用 JavaScript 动态加载内容，需要特殊处理：
1. 等待初始内容渲染（等待3秒）
2. 检查内容是否已加载
3. 如果没有，继续等待或尝试滚动触发加载
4. 滚动到页面底部（可能触发懒加载）
5. 等待新内容加载完成
6. 重复步骤4-5，直到获取足够数据或页面不再有新内容
""",
max_steps=30 # 动态页面需要更多步骤
)
Q5: 数据提取不完整怎么办？
"""
问题：提取的数据不完整或格式不对
解决方案：
"""
# 优化指令，明确输出格式
task = Task(
url="https://example.com",
instruction="""
严格按照以下格式提取数据：
输出必须是一个 JSON 数组，每个元素包含：
{
"title": "文章标题（字符串，不能为空）",
"url": "文章链接（必须是完整的 URL，包含 https://）",
"date": "发布日期（格式：YYYY-MM-DD）",
"views": "阅读量（整数，如：1234）"
}
注意事项：
- 确保每个字段都有值，不要返回 null 或空字符串
- 如果某个字段在页面上找不到，在该字段填入 "N/A"
- 返回至少10条数据
""",
max_steps=20
)
# 额外技巧：添加验证步骤
instruction_with_validation = """
1. 提取数据
2. 对提取的数据进行验证：
- 检查每个字段是否有值
- 检查 URL 格式是否正确
- 检查日期格式是否符合要求
3. 如果验证失败，尝试重新提取
4. 返回通过验证的数据
"""
性能优化与生产部署
生产环境配置
"""
生产环境部署配置示例
保存为: production_config.py
"""
import os
# 生产环境环境变量配置
PRODUCTION_CONFIG = """
# 数据库配置（使用 PostgreSQL）
SKYVERNET_DB_URL="postgresql://user:password@host:5432/skyvern_prod"
# Redis 配置（用于任务队列）
SKYVERNET_REDIS_URL="redis://localhost:6379/0"
# LLM 配置
OPENAI_API_KEY="${OPENAI_API_KEY}"
OPENAI_API_BASE="https://api.openai.com/v1"
# 并发配置
SKYVERNET_MAX_CONCURRENT_TASKS=10
SKYVERNET_TASK_TIMEOUT=600
# 浏览器配置
SKYVERNET_BROWSER_HEADLESS=true
SKYVERNET_BROWSER_MAX_POOL_SIZE=20
# 日志配置
LOG_LEVEL="WARNING"
LOG_FORMAT="json"
# 安全配置
SKYVERNET_SECRET_KEY="your-secret-key-here"
"""
# Docker Compose 配置
DOCKER_COMPOSE_YML = """
version: '3.8'
services:
skyvern:
image: skyvern/skyvern:latest
ports:
- "8080:8080"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- SKYVERNET_DB_URL=postgresql://postgres:postgres@db:5432/skyvern
- SKYVERNET_REDIS_URL=redis://redis:6379/0
depends_on:
- db
- redis
restart: unless-stopped
db:
image: postgres:15
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_DB=skyvern
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
restart: unless-stopped
volumes:
postgres_data:
redis_data:
"""
监控与日志
"""
生产环境监控配置
"""
from skyvern.forge import Forge
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("skyvern")
# 创建 Forge 实例时启用监控
forge = Forge(
enable_metrics=True,
metrics_port=9090, # Prometheus 指标端口
# Webhook 通知
webhook_config={
"on_start": "https://your-server.com/webhook/start",
"on_progress": "https://your-server.com/webhook/progress",
"on_complete": "https://your-server.com/webhook/complete",
"on_error": "https://your-server.com/webhook/error"
}
)
# 监控指标
METRICS_TO_TRACK = """
Prometheus 指标：
- skyvern_tasks_total：任务总数
- skyvern_tasks_completed：完成的任务数
- skyvern_tasks_failed：失败的任务数
- skyvern_task_duration_seconds：任务执行时长
- skyvern_browser_actions_total：浏览器操作总数
- skyvern_llm_calls_total：LLM 调用次数
"""
总结与资源链接
Skyvern 的优势总结
| 特性 | 传统方案（Selenium） | Skyvern |
|---|---|---|
| 配置复杂度 | 高（需要写大量代码） | 低（自然语言描述） |
| 维护成本 | 高（网站改版即失效） | 低（AI 智能适配） |
| 错误处理 | 需要手动编写 | 内置自动重试 |
| 学习曲线 | 陡峭 | 平缓 |
| 适用场景 | 结构固定的场景 | 任意网页场景 |
相关项目推荐
如果你对 Skyvern 感兴趣，以下项目也值得关注：
- Playwright：强大的浏览器自动化工具，Skyvern 的底层依赖
- Puppeteer：Node.js 生态的浏览器自动化框架
- Selenium：老牌浏览器自动化工具
- LangChain：构建 LLM 应用的框架，可与 Skyvern 结合
- AutoGPT：自主 AI Agent 项目，与 Skyvern 有相似的理念
- Browser-use：另一个基于 AI 的浏览器自动化库
- Crawl4AI：专为 LLM 设计的网页爬虫
继续学习资源
- 官方文档：https://docs.skyvern.ai
- GitHub 仓库：https://github.com/Skyvern-AI/skyvern
- 示例代码：https://github.com/Skyvern-AI/skyvern/tree/main/examples
- Discord 社区：加入讨论获取帮助
- OpenAI 文档：了解更多关于 GPT-4 Vision 的能力
动手实践
学习 Skyvern 最好的方式就是动手实践。建议按照以下顺序进行练习：
- 第一阶段：运行官方示例，理解基本用法
- 第二阶段：修改示例代码，尝试不同的任务
- 第三阶段：解决实际问题，如数据采集、自动化测试等
- 第四阶段：深入源码，理解内部机制
- 第五阶段：贡献代码，参与开源社区
无论你是开发者、数据科学家，还是对自动化感兴趣的普通用户，Skyvern 都能帮助你以前所未有的效率完成网页自动化任务。现在就开始你的 Skyvern 之旅吧！
本文档会持续更新，欢迎关注 GitHub 仓库获取最新内容。如果有任何问题或建议，请在评论区留言！
评论区