项目概述

Oxylabs AI Studio Python SDK（oxylabs-ai-studio）是一个用于与 Oxylabs AI Studio API 服务无缝交互的 Python 客户端库。该 SDK 提供了对多种 AI 驱动数据提取工具的访问，包括 AI-Scraper、AI-Crawler、AI-Browser-Agent 等。 资料来源：readme.md:1-8

核心特性

• 多应用支持：集成爬虫、搜索、映射等多种数据提取功能
• 同步/异步接口：同时支持同步和异步编程模式
• 结构化输出：支持 JSON、Markdown、CSV、Screenshot 等多种输出格式
• Schema 生成：内置 AI 驱动的 Schema 自动生成功能
• 地理位置定位：支持全球代理位置定位
• JavaScript 渲染：可选的 JavaScript 渲染支持

系统架构

整体架构图

graph TD
    A[用户应用] --> B[oxylabs-ai-studio SDK]
    B --> C[AiCrawler]
    B --> D[AiScraper]
    B --> E[AiSearch]
    B --> F[AiMap]
    B --> G[BrowserAgent]
    C --> H[Oxylabs AI Studio API]
    D --> H
    E --> H
    F --> H
    G --> H
    H --> I[SERP / Web Content]

模块组成

安装与配置

环境要求

• Python 3.10 及以上版本
• 有效的 Oxylabs API 密钥

安装方式

pip install oxylabs-ai-studio

资料来源：readme.md:13

初始化配置

所有应用模块的初始化方式相同，需要提供 API 密钥：

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

应用模块详解

AI-Crawler（网站爬虫）

#### 功能说明

AiCrawler 模块用于深度爬取整个网站或指定路径下的所有相关页面，支持用户自定义的自然语言提示来指导提取过程。 资料来源：readme.md:27-28

#### 同步接口

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

url = "https://oxylabs.io"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="markdown",
    render_javascript=False,
    return_sources_limit=3,
    geo_location="United States",
)
print("Results:")
for item in result.data:
    print(item, "\n")

资料来源：readme.md:27-42

#### 参数说明

资料来源：readme.md:43-51

AI-Scraper（页面抓取）

#### 功能说明

AiScraper 专注于单页面内容抓取，可返回 Markdown 格式内容或根据用户提供的 JSON Schema 返回结构化数据。 资料来源：agentic_code_guide.md:62-66

#### 同步接口

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

schema = scraper.generate_schema(prompt="want to parse developer, platform, type, price game title, genre (array) and description")
print(f"Generated schema: {schema}")

url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema=schema,
    render_javascript=False,
)
print(result)

资料来源：readme.md:81-95

#### 异步接口

import asyncio
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

async def main():
    url = "https://sandbox.oxylabs.io/products/3"
    result = await scraper.scrape_async(
        url=url,
        output_format="json",
        schema={"type": "object", "properties": {"price": {"type": "string"}}, "required": []},
        render_javascript=False,
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

资料来源：agentic_code_guide.md:68-86

#### 参数说明

#### 返回数据结构

class AiScraperJob(BaseModel):
    run_id: str
    message: str | None = None
    data: str | dict | None

• output_format 为 "json" 时：data 为字典
• output_format 为 "markdown" 时：data 为字符串
• output_format 为 "csv" 时：data 为 CSV 格式字符串
• output_format 为 "screenshot" 时：data 为字符串（Base64 或 URL）

资料来源：agentic_code_guide.md:87-101

AI-Search（搜索引擎）

#### 功能说明

AiSearch 提供搜索引擎结果获取功能，支持标准搜索和即时搜索两种模式。即时搜索在 limit <= 10 且 return_content=False 时自动启用，提供更快的响应。 资料来源：readme.md:10-24

#### 标准搜索

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna recipe"
result = search.search(
    query=query,
    limit=5,
    render_javascript=False,
    return_content=True,
)
print(result.data)

资料来源：readme.md:10-24

#### 即时搜索

result = search.instant_search(
    query=query,
    limit=10,
)
print(result.data)

#### 参数说明

#### 即时搜索限制

• 最大 limit 为 10
• 仅支持 query、limit、geo_location 三个参数

资料来源：readme.md:17-24

AI-Map（网站映射）

#### 功能说明

AiMap 用于快速映射网站结构，通过关键词搜索定位相关页面。 资料来源：readme.md:54-70

#### 使用示例

from oxylabs_ai_studio.apps.ai_map import AiMap

ai_map = AiMap(api_key="<API_KEY>")
payload = {
    "url": "https://career.oxylabs.io",
    "search_keywords": ["career", "jobs", "vacancy"],
    "user_prompt": "job ad pages",
    "max_crawl_depth": 2,
    "limit": 10,
    "geo_location": "Germany",
    "render_javascript": False,
    "include_sitemap": True,
}
result = ai_map.map(**payload)
print(result.data)

#### 参数说明

资料来源：src/oxylabs_ai_studio/apps/ai_map.py:62-75

Browser-Agent（浏览器代理）

#### 功能说明

BrowserAgent 提供浏览器级别的自动化操作能力，可以执行复杂的用户交互任务，如点击、滚动、表单填写等，通过自然语言提示指导操作过程。 资料来源：agentic_code_guide.md:16-27

#### 同步接口

from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

result = browser_agent.run(
    url="https://sandbox.oxylabs.io/",
    user_prompt="Find if there is game 'super mario odyssey' in the store.",
    output_format="json",
    schema={"type": "object", "properties": {"page_url": {"type": "string"}}, "required": []},
)
print(result.data)

#### 异步接口

import asyncio
from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

async def main():
    prompt = "Find if there is game 'super mario odyssey' in the store."
    url = "https://sandbox.oxylabs.io/"
    result = await browser_agent.run_async(
        url=url,
        user_prompt=prompt,
        output_format="json",
        schema={"type": "object", "properties": {"page_url": {"type": "string"}}, "required": []},
    )
    print(result.data)

if __name__ == "__main__":
    asyncio.run(main())

#### 参数说明

#### 返回数据结构

class DataModel(BaseModel):
    type: Literal["json", "markdown", "html", "screenshot", "csv"]
    content: dict[str, Any] | str | None

class BrowserAgentJob(BaseModel):
    run_id: str
    message: str | None = None
    data: DataModel | None = None

资料来源：agentic_code_guide.md:28-48

Schema 生成功能

功能概述

所有主要模块都提供了 generate_schema() 方法，支持通过自然语言提示自动生成 OpenAPI JSON Schema，实现零配置的快速开发。 资料来源：examples/scrape_generated_schema.py:1-9

工作流程

graph LR
    A[输入自然语言提示] --> B[调用 API 生成 Schema]
    B --> C[获得 JSON Schema]
    C --> D[用于抓取/爬取任务]

使用示例

# 为爬虫生成 Schema
schema = crawler.generate_schema(
    prompt="proxy plans which have name, price, and features",
)
print("schema: ", schema)

# 为抓取生成 Schema
schema = scraper.generate_schema(
    prompt="want to parse developer, platform, type, price game title, genre (array) and description"
)
print(f"Generated schema: {schema}")

资料来源：examples/crawl_generated_schema.py:4-13

使用场景示例

电商产品数据提取

#### 场景描述

从电商网站提取产品分类页面的所有产品数据，并获取每个产品的详细信息。 资料来源：agentic_code_guide.md:103-125

#### 推荐工作流

graph TD
    A[识别分类页面] --> B[Browser-Agent 定位分类页 URL]
    B --> C[提取分页 URLs]
    C --> D[AI-Scraper 提取产品 URLs]
    D --> E[AI-Scraper 提取产品详情]

#### 实施步骤

1. 定位分类页面

• 使用 Browser-Agent 识别分类页面 URL 和所有分页 URL
• 定义返回分页 URL 的 JSON Schema

1. 提取产品链接

• 使用 AI-Scraper 从分页页面提取所有产品 URL

1. 提取产品详情

• 使用 AI-Scraper 和预定义的 JSON Schema 提取详细数据

#### 示例 Schema

{
  "type": "object",
  "properties": {
    "paginationUrls": {
      "type": "array",
      "description": "Return all URLs from first to last page in category pagination. If you noticed there are missing URLs, because category page does not list them all, create them to match existing ones.",
      "items": {
        "type": "string"
      }
    }
  },
  "required": []
}

资料来源：agentic_code_guide.md:115-127

地理位置定位

支持格式

SDK 支持多种地理位置指定方式：

资料来源：readme.md:46-48

使用示例

result = scraper.scrape(
    url="https://sandbox.oxylabs.io/products/1",
    output_format="markdown",
    render_javascript=False,
    geo_location="Germany",
)

错误处理与超时

超时处理

所有同步方法在超时时会抛出 TimeoutError 异常：

try:
    result = crawler.crawl(url="https://example.com", ...)
except TimeoutError as e:
    print(f"爬取超时: {e}")

用户取消

用户可通过键盘中断（Ctrl+C）取消正在执行的任务：

except KeyboardInterrupt:
    logger.info("[Cancelled] Request was cancelled by user.")
    raise KeyboardInterrupt from None

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:89-91

输出格式对比

资料来源：readme.md:43-44

依赖关系

核心依赖

• httpx: 异步 HTTP 客户端，用于 API 通信
• pydantic: 数据验证和模型定义

完整依赖列表

请参考项目根目录下的 pyproject.toml 文件获取完整的依赖声明。

概述

Oxylabs AI Studio Python SDK（oxylabs-ai-studio-py）是一个用于与 Oxylabs AI Studio API 服务进行交互的 Python SDK。该 SDK 封装了多个 AI 驱动的网页数据提取工具，为开发者提供了简洁的 Python 接口来调用 AI-Scraper、AI-Crawler、AI-Browser-Agent 等服务。资料来源：readme.md

系统要求

资料来源：readme.md

安装

通过 pip 安装 SDK：

pip install oxylabs-ai-studio

资料来源：readme.md

初始化配置

使用 SDK 前，需要实例化相应的应用类并传入 API 密钥：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

SDK 架构

graph TD
    A[用户代码] --> B[oxylabs-ai-studio-py SDK]
    B --> C[AiScraper]
    B --> D[AiCrawler]
    B --> E[AiSearch]
    B --> F[BrowserAgent]
    B --> G[AiMap]
    C --> H[Oxylabs AI Studio API]
    D --> H
    E --> H
    F --> H
    G --> H

核心应用模块

AI-Scraper（智能网页抓取）

用于抓取网页内容并返回 Markdown 格式或结构化 JSON 数据。当选择 JSON 输出时，用户需提供有效的 JSON Schema。资料来源：agentic_code_guide.md

#### 同步接口示例

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema={"type": "object", "properties": {"price": {"type": "string"}}, "required": []},
    render_javascript=False,
)
print(result)

资料来源：examples/scrape_generated_schema.py

#### 异步接口示例

import asyncio
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

async def main():
    url = "https://sandbox.oxylabs.io/products/3"
    result = await scraper.scrape_async(
        url=url,
        output_format="json",
        schema={"type": "object", "properties": {"price": {"type": "string"}}, "required": []},
        render_javascript=False,
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

#### 参数说明

#### 返回数据模型

class AiScraperJob(BaseModel):
    run_id: str
    message: str | None = None
    data: str | dict | None

• 当 output_format 为 "json" 时，data 为字典
• 当 output_format 为 "markdown" 时，data 为字符串
• 当 output_format 为 "csv" 时，data 为 CSV 格式字符串
• 当 output_format 为 "screenshot" 时，data 为字符串

资料来源：agentic_code_guide.md

AI-Crawler（智能网站爬取）

用于从起始 URL 开始爬取整个网站，根据自然语言提示引导提取过程。资料来源：readme.md

#### 基本示例

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

url = "https://oxylabs.io"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="markdown",
    render_javascript=False,
    return_sources_limit=3,
    geo_location="France",
)
print("Results:")
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_markdown.py

#### 使用 Pydantic Schema

from pydantic import BaseModel, Field
from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

class ProxyPlan(BaseModel):
    name: str = Field(description="The name of the proxy plan")
    price: str = Field(description="The price of the proxy plan")
    features: list[str] = Field(description="The features of the proxy plan")

class ProxyPlans(BaseModel):
    proxy_plans: list[ProxyPlan] = Field(description="The proxy plans")

url = "https://oxylabs.io/"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=ProxyPlans.model_json_schema(),
    render_javascript=False,
)

资料来源：examples/crawl_pydantic_schema.py

#### 参数说明

AI-Search（SERP 搜索）

用于执行搜索引擎结果页面（SERP）搜索。资料来源：readme.md

#### 搜索并返回内容

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna recipe"
result = search.search(
    query=query,
    limit=5,
    render_javascript=False,
    return_content=True,
)
print(result.data)

资料来源：examples/search_with_content.py

#### 即时搜索（快速）

当 limit <= 10 且 return_content=False 时，搜索自动使用即时端点（/search/instant），返回结果更快。资料来源：readme.md

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna recipes"
result = search.instant_search(
    query=query,
    limit=5,
    geo_location="United States",
)
print(result.data)

资料来源：examples/search_instant.py

#### 参数说明

Browser-Agent（浏览器自动化）

用于控制浏览器执行点击、滚动和导航等操作，根据文本提示执行动作。资料来源：agentic_code_guide.md

#### 同步接口

from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

prompt = "Find if there is game 'super mario odyssey' in the store."
url = "https://sandbox.oxylabs.io/"
result = browser_agent.run(
    url=url,
    user_prompt=prompt,
    output_format="json",
    schema={"type": "object", "properties": {"page_url": {"type": "string"}}, "required": []},
)
print(result.data)

#### 异步接口

import asyncio
from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

async def main():
    prompt = "Find if there is game 'super mario odyssey' in the store."
    url = "https://sandbox.oxylabs.io/"
    result = await browser_agent.run_async(
        url=url,
        user_prompt=prompt,
        output_format="json",
        schema={"type": "object", "properties": {"page_url": {"type": "string"}}, "required": []},
    )
    print(result.data)

if __name__ == "__main__":
    asyncio.run(main())

Schema 生成

SDK 提供了 generate_schema 方法，可以根据自然语言描述自动生成 JSON Schema，无需手动编写。资料来源：examples/scrape_generated_schema.py

schema = scraper.generate_schema(
    prompt="want to parse developer, platform, type, price game title, genre (array) and description"
)
print(f"Generated schema: {schema}")

实现最佳实践

根据官方指南，使用 SDK 时应遵循以下最佳实践：资料来源：agentic_code_guide.md

1. 安装最新版本：确保使用最新版本的 oxylabs-ai-studio
2. 限流控制：实现限流机制，遵守所购买套餐的速率限制
3. 重试机制：为处理失败的请求引入重试逻辑，但需设置重试次数上限以避免无限循环

工作流程图

graph TD
    A[安装 SDK] --> B[获取 API Key]
    B --> C[实例化应用类]
    C --> D{选择功能}
    D --> E[AiScraper]
    D --> F[AiCrawler]
    D --> G[AiSearch]
    D --> H[BrowserAgent]
    E --> I[调用 scrape 方法]
    F --> J[调用 crawl 方法]
    G --> K[调用 search/instant_search]
    H --> L[调用 run 方法]
    I --> M[处理返回结果]
    J --> M
    K --> M
    L --> M

常见输出格式对比

异步编程支持

SDK 同时支持同步和异步接口，便于集成到异步应用中：

import asyncio
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

async def main():
    scraper = AiScraper(api_key="<API_KEY>")
    result = await scraper.scrape_async(url="https://example.com")
    print(result)

asyncio.run(main())

概述

AI-Scraper 是 Oxylabs AI Studio Python SDK 中的核心模块之一，专门用于从目标网页提取结构化或非结构化内容。该工具通过自然语言提示和 JSON Schema 定义，实现智能化的网页数据抓取功能，支持将网页内容转换为 Markdown、JSON、CSV 或截图格式。 资料来源：readme.md

AI-Scraper 的主要应用场景包括：

• 电商产品数据提取：从商品页面抓取价格、描述、规格等信息
• 内容聚合：批量提取新闻、文章、评论等内容
• 数据监控：定期抓取网站内容进行市场分析
• 结构化数据导出：将网页数据转换为可读的 JSON 或 CSV 格式

工作流程

graph TD
    A[初始化 AiScraper] --> B[设置目标 URL]
    B --> C[定义输出格式]
    C --> D{output_format == json?}
    D -->|是| E[提供 JSON Schema]
    D -->|否| F[直接抓取]
    E --> G[调用 scrape 方法]
    F --> G
    G --> H[AI 引擎解析内容]
    H --> I[返回 AiScraperJob 结果]
    I --> J{data 类型}
    J -->|JSON| K[dict 对象]
    J -->|Markdown| L[字符串]
    J -->|CSV| M[CSV 格式字符串]
    J -->|Screenshot| N[图片字符串]

核心类定义

AiScraper

AiScraper 是 AI-Scraper 模块的主类，负责处理网页抓取请求。 资料来源：agentic_code_guide.md

#### 初始化

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

构造函数参数：

AiScraperJob

抓取任务完成后返回的数据模型类。

class AiScraperJob(BaseModel):
    run_id: str
    message: str | None = None
    data: str | dict | None

资料来源：agentic_code_guide.md

API 方法

同步接口

scrape 方法用于执行同步网页抓取操作。

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema={"type": "object", "properties": {"price": {"type": "string"}}, "required": []},
    render_javascript=False,
)
print(result)

资料来源：agentic_code_guide.md

异步接口

scrape_async 方法用于执行异步网页抓取操作，适用于需要并发处理多个请求的场景。

import asyncio
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

async def main():
    url = "https://sandbox.oxylabs.io/products/3"
    result = await scraper.scrape_async(
        url=url,
        output_format="json",
        schema={"type": "object", "properties": {"price": {"type": "string"}}, "required": []},
        render_javascript=False,
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

资料来源：agentic_code_guide.md

Schema 生成方法

generate_schema 方法可以根据自然语言描述自动生成 JSON Schema，无需手动编写复杂的 Schema 定义。

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

schema = scraper.generate_schema(
    prompt="want to parse developer, platform, type, price game title, genre (array) and description"
)
print(f"Generated schema: {schema}")

资料来源：examples/scrape_generated_schema.py

参数说明

scrape / scrape_async 方法参数

output_format 可选值

资料来源：readme.md

使用示例

基础 Markdown 抓取

抓取网页并以 Markdown 格式返回内容：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

url = "https://sandbox.oxylabs.io/products/1"
result = scraper.scrape(
    url=url,
    output_format="markdown",
    render_javascript=False,
    geo_location="Germany",
)
print(result)

资料来源：examples/scrape_markdown.py

使用 JSON Schema 抓取结构化数据

定义明确的 JSON Schema 来提取特定字段：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema={
        "type": "object",
        "properties": {
            "price": {"type": "string"},
            "title": {"type": "string"}
        },
        "required": []
    },
    render_javascript=False,
)
print(result.data)

资料来源：agentic_code_guide.md

使用 Pydantic 模型

通过 Pydantic 模型定义数据结构，利用 model_json_schema() 方法自动生成 Schema：

from pydantic import BaseModel
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

class Game(BaseModel):
    title: str
    genre: list[str]
    developer: str
    platform: str
    game_type: str
    description: str
    price: str
    availability: str

url = "https://sandbox.oxylabs.io/products/1"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema=Game.model_json_schema(),
    render_javascript=False,
)
print(result)

资料来源：examples/scrape_pydantic_schema.py

使用自动生成的 Schema

让 AI 自动根据描述生成 Schema，简化开发流程：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

# 自动生成 Schema
schema = scraper.generate_schema(
    prompt="want to parse developer, platform, type, price game title, genre (array) and description"
)
print(f"Generated schema: {schema}")

# 使用生成的 Schema 进行抓取
url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema=schema,
    render_javascript=False,
)
print(result)

资料来源：examples/scrape_generated_schema.py

JavaScript 渲染配置

render_javascript 参数控制是否执行页面中的 JavaScript 代码：

当设置为 "auto" 时，服务会自动判断目标网页是否需要 JavaScript 渲染才能完整显示内容。 资料来源：readme.md

地理位置定位

通过 geo_location 参数可以指定代理服务器的位置，以获取特定地区的内容版本：

result = scraper.scrape(
    url=url,
    output_format="markdown",
    geo_location="Germany",  # 使用德国代理
)

支持的格式：

• ISO 2 字母代码（如 "DE"、"US"）
• 国家标准名称（如 "Germany"、"United States"）

资料来源：readme.md

最佳实践

速率限制

确保实现遵守与购买计划关联的速率限制，以防止服务中断或过度使用。

重试机制

引入重试逻辑以处理失败的请求，但应设置重试次数上限，避免无限循环或过度 API 调用。

Schema 设计建议

• 使用 Pydantic 模型定义复杂数据结构，便于类型检查和维护
• 对于简单提取任务，使用 generate_schema 方法自动生成 Schema
• 在 Schema 的 required 数组中仅包含必需字段，减少解析失败的概率

JavaScript 渲染策略

• 仅在必要时启用 JavaScript 渲染，以优化性能和成本
• 使用 "auto" 模式让系统自动决策
• 对于纯静态页面，禁用渲染以提高响应速度

返回值处理

根据不同的 output_format，result.data 的类型会有所不同：

访问返回数据：

result = scraper.scrape(url=url, output_format="json", schema=schema)
if isinstance(result.data, dict):
    price = result.data.get("price")
    print(f"价格: {price}")

资料来源：agentic_code_guide.md

错误处理

建议实现完整的错误处理机制：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper
from oxylabs_ai_studio.exceptions import (
    AuthenticationError,
    RateLimitError,
    ScrapingError
)

scraper = AiScraper(api_key="<API_KEY>")

try:
    result = scraper.scrape(
        url=url,
        output_format="markdown",
    )
    if result.message:
        print(f"警告: {result.message}")
    print(result.data)
except AuthenticationError:
    print("API 密钥无效")
except RateLimitError:
    print("速率限制已达到，请稍后重试")
except ScrapingError as e:
    print(f"抓取失败: {e}")

概述

AI-Crawler 是 Oxylabs AI Studio Python SDK 中的核心模块之一，专门用于自动化网页爬取任务。该模块基于自然语言提示词驱动，能够智能识别和提取目标网页中的相关内容，支持多种输出格式和结构化数据提取。

AI-Crawler 继承自 OxyStudioAIClient 基类，通过异步轮询机制与 Oxylabs 后端 API 进行交互，实现高效的网页内容抓取。资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:35

核心组件

数据模型

AI-Crawler 使用 Pydantic BaseModel 定义返回数据结构，确保类型安全和数据验证。

class AiCrawlerJob(BaseModel):
    run_id: str
    message: str | None = None
    data: list[dict[str, Any]] | list[str] | None = None

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:22-25

配置常量

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:9-11

主要功能

1. 网页爬取 (crawl)

crawl() 方法是 AiCrawler 的核心功能，用于执行网页爬取任务。

def crawl(
    self,
    url: str,
    user_prompt: str,
    output_format: Literal["json", "markdown", "csv", "toon"] = "markdown",
    schema: dict[str, Any] | None = None,
    render_javascript: bool = False,
    return_sources_limit: int = 25,
    geo_location: str | None = None,
    max_credits: int | None = None,
) -> AiCrawlerJob

#### 参数说明

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:32-41

#### 业务逻辑验证

当 output_format 设置为 json、csv 或 toon 时，必须提供 schema 参数，否则将抛出 ValueError：

if output_format in ["json", "csv", "toon"] and schema is None:
    raise ValueError(
        "openapi_schema is required when output_format is json, csv or toon.",
    )

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:44-48

2. Schema 生成 (generate_schema)

generate_schema() 方法允许用户通过自然语言描述自动生成 JSON Schema，无需手动编写复杂的 Schema 定义。

def generate_schema(self, prompt: str) -> dict[str, Any] | None:
    logger.info("Generating schema")
    body = {"user_prompt": prompt}
    response = self.call_api(
        client=self.get_client(),
        url="/crawl/generate-params",
        method="POST",
        body=body,
    )
    if response.status_code != 200:
        raise Exception(f"Failed to generate schema: {response.text}")
    json_response: SchemaResponse = response.json()
    return json_response.get("data", {}).get("schema")

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:68-80

执行流程

AI-Crawler 采用同步轮询机制获取爬取结果，其工作流程如下：

graph TD
    A[开始 crawl 调用] --> B[调用 /crawl/run 接口提交任务]
    B --> C{响应状态码}
    C -->|202| D[提取 run_id]
    C -->|其他| E[抛出异常]
    D --> F[轮询 /crawl/run/data 接口]
    F --> G{获取状态}
    G -->|202 processing| F
    G -->|200 completed| H[返回 AiCrawlerJob]
    G -->|200 failed| I[返回 AiCrawlerJob with error]
    G -->|其他| F
    H --> J[结束]
    I --> J
    E --> J

超时处理

当轮询次数超过 POLL_MAX_ATTEMPTS（120次）或总耗时超过 CRAWLER_TIMEOUT_SECONDS（600秒）时，任务将抛出 TimeoutError：

raise TimeoutError(f"Failed to crawl {url}: timeout.")

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:61

用户中断处理

支持 KeyboardInterrupt 异常捕获，允许用户通过中断信号取消爬取任务：

except KeyboardInterrupt:
    logger.info("[Cancelled] Crawling was cancelled by user.")
    raise KeyboardInterrupt from None

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:57-59

使用示例

示例一：Markdown 格式输出

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

url = "https://oxylabs.io"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="markdown",
    render_javascript=False,
    return_sources_limit=3,
    geo_location="France",
)
print("Results:")
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_markdown.py 和 readme.md

示例二：使用 Pydantic Schema

from pydantic import BaseModel, Field
from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

class ProxyPlan(BaseModel):
    name: str = Field(description="The name of the proxy plan")
    price: str = Field(description="The price of the proxy plan")
    features: list[str] = Field(description="The features of the proxy plan")

class ProxyPlans(BaseModel):
    proxy_plans: list[ProxyPlan] = Field(description="The proxy plans")

url = "https://oxylabs.io/"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=ProxyPlans.model_json_schema(),
    render_javascript=False,
)
print("Results:\n")
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_pydantic_schema.py

示例三：自动生成 Schema

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

schema = crawler.generate_schema(
    prompt="proxy plans which have name, price, and features",
)
print("schema: ", schema)

url = "https://oxylabs.io"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=schema,
    render_javascript=False,
)
print("Results:")
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_generated_schema.py

输出格式

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:37

地理位置支持

geo_location 参数支持多种格式：

• ISO 2 字母国家代码（如 FR、DE、US）
• 国家正式名称（如 France、Germany、United States）

支持的地理位置列表可参考 官方文档。

资料来源：readme.md

错误处理

常见错误场景

任务状态流转

stateDiagram-v2
    [*] --> Processing: 任务提交
    Processing --> Processing: 状态码 202
    Processing --> Completed: 状态码 200 + status="completed"
    Processing --> Failed: 状态码 200 + status="failed"
    Completed --> [*]
    Failed --> [*]

最佳实践

1. 合理设置超时时间

默认超时时间为 10 分钟，适用于大多数爬取任务。对于大型网站或网络环境较差的情况，可考虑：

• 使用 render_javascript=True 时预留更多时间
• 分批次爬取以降低单次任务复杂度

2. Schema 设计建议

• 使用清晰的字段描述（description）
• 对于数组类型，指定 items 的具体结构
• 使用 Pydantic 模型时，通过 model_json_schema() 方法自动转换

3. 积分控制

通过 max_credits 参数限制单次任务的最大积分消耗，避免意外超额：

result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    max_credits=100,
)

4. 地理位置选择

选择与目标网站服务器相近的地理位置可提高响应速度和稳定性：

geo_location="Germany"  # 爬取德国服务器网站时推荐

类层次结构

OxyStudioAIClient (基类)
    └── AiCrawler (子类)

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:35

依赖模块

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:1-8

总结

AI-Crawler 模块提供了一套完整的智能网页爬取解决方案，其核心优势包括：

• 自然语言驱动：通过 user_prompt 描述提取需求，无需编写复杂的 CSS/XPath 选择器
• 多格式支持：原生支持 Markdown、JSON、CSV 等多种输出格式
• Schema 自动生成：降低结构化数据提取的学习成本
• 完善的错误处理：统一的异常处理机制和任务状态追踪
• 灵活的地理定位：支持全球多地区的代理爬取

开发者可通过本模块快速实现复杂网页内容的结构化提取，适用于价格监控、竞品分析、内容聚合等多种应用场景。

概述

AI-Search 是 Oxylabs AI Studio Python SDK 中的搜索引擎模块，提供基于自然语言处理的智能搜索功能。该模块封装了与 Oxylabs AI Search API 的交互逻辑，支持同步和异步两种调用方式，能够执行标准搜索和即时搜索两种模式。

主要功能包括：

• 执行 SERP（搜索引擎结果页面）搜索
• 支持返回完整的 Markdown 内容
• 提供即时搜索模式（适用于少量结果场景）
• 支持地理位置定位
• 支持 JavaScript 渲染

核心类与数据模型

AiSearchJob 类

AiSearchJob 是搜索任务的返回结果模型，继承自 Pydantic 的 BaseModel：

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:21-23

SearchResult 类

SearchResult 定义了单条搜索结果的数据结构：

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:14-18

AiSearch 类

AiSearch 类是搜索引擎模块的主入口类，继承自 OxyStudioAIClient 基类：

class AiSearch(OxyStudioAIClient):
    """AI Search app."""

    def __init__(self, api_key: str | None = None):
        super().__init__(api_key=api_key)

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:26-29

架构与工作流程

搜索流程图

graph TD
    A[开始搜索] --> B{limit <= 10 且 return_content = False?}
    B -->|是| C[调用 instant_search 即时搜索]
    B -->|否| D[调用 /search/run 创建任务]
    C --> E[直接返回结果]
    D --> F{轮询状态}
    F -->|202 状态码| G[等待 POLL_INTERVAL_SECONDS]
    G --> F
    F -->|200 完成| H[返回 AiSearchJob]
    F -->|失败| I[返回失败状态]
    E --> J[结束]
    H --> J
    I --> J

轮询机制参数

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:9-11

同步接口

标准搜索 (search 方法)

标准搜索采用轮询机制，适用于需要获取完整内容的场景：

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna recipe"
result = search.search(
    query=query,
    limit=5,
    render_javascript=False,
    return_content=True,
)
print(result.data)

资料来源：examples/search_with_content.py:1-13

搜索不带内容 (return_content=False)

当不需要返回完整内容时，可以设置 return_content=False：

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna"
result = search.search(
    query=query,
    limit=5,
    render_javascript=False,
    return_content=False,
    geo_location="Italy",
)
print(result.data)

资料来源：examples/search_no_content.py:1-15

即时搜索 (instant_search 方法)

即时搜索是一种快速搜索模式，适用于以下条件：

• limit <= 10
• return_content = False

即时搜索直接调用 /search/instant 端点，无需轮询，立即返回结果。

from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

query = "lasagna recipes"
result = search.instant_search(
    query=query,
    limit=5,
    geo_location="United States",
)
print(result.data)

资料来源：examples/search_instant.py:1-13

异步接口

异步搜索 (search_async 方法)

异步版本的搜索方法，适用于异步应用程序：

import asyncio
from oxylabs_ai_studio.apps.ai_search import AiSearch

search = AiSearch(api_key="<API_KEY>")

async def main():
    result = await search.search_async(
        query="lasagna recipe",
        limit=5,
        render_javascript=False,
        return_content=True,
    )
    print(result.data)

if __name__ == "__main__":
    asyncio.run(main())

异步接口的内部逻辑与同步版本相同，首先判断是否满足即时搜索条件，满足则调用 instant_search_async 方法。

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:57-106

异步即时搜索 (instant_search_async 方法)

异步即时搜索方法：

async def instant_search_async(
    self,
    query: str,
    limit: int = 10,
    geo_location: str | None = None,
) -> AiSearchJob:
    """Async instant SERP search."""
    if not query:
        raise ValueError("query is required")
    body = {
        "query": query,
        "limit": limit,
        "geo_location": geo_location,
    }
    async with self.async_client() as client:
        response = await self.call_api_async(
            client=client, url="/search/instant", method="POST", body=body
        )
        status_code = response.status_code
        if status_code != 200:
            raise Exception(f"Failed to perform instant search: `{response.text}`")
        resp_body = response.json()
        return AiSearchJob(
            run_id=resp_body.get("run_id", ""),
            message=resp_body.get("status", None),
            data=resp_body.get("data", None),
        )

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:153-175

参数说明

search / search_async 方法参数

instant_search / instant_search_async 方法参数

使用场景对比

地理位置支持

geo_location 参数支持多种格式：

• ISO 2-letter 格式（如 "US", "DE"）
• 国家规范名称（如 "United States", "Germany"）
• 坐标格式（高级定位场景）

地理位置功能允许搜索特定区域的内容，返回针对该地区的搜索引擎结果。

错误处理

常见错误

超时与重试机制

graph LR
    A[发起请求] --> B{响应状态}
    B -->|200| C[返回成功结果]
    B -->|202| D[等待 5 秒]
    D --> B
    B -->|其他| E[等待 5 秒重试]
    E --> B
    C --> F[结束]
    G[超时 3 分钟] --> H[抛出 TimeoutError]

返回值示例

成功响应 (AiSearchJob)

AiSearchJob(
    run_id="abc123-def456",
    message=None,
    data=[
        SearchResult(
            url="https://example.com/recipe/lasagna",
            title="Classic Italian Lasagna Recipe",
            description="A delicious traditional lasagna...",
            content="# Lasagna Recipe\n\n## Ingredients\n..."
        ),
        SearchResult(
            url="https://example.com/lasagna-tips",
            title="10 Tips for Perfect Lasagna",
            description="Learn how to make...",
            content=None
        )
    ]
)

即时搜索响应

即时搜索返回的数据结构相同，但不包含 content 字段：

AiSearchJob(
    run_id="xyz789",
    message="completed",
    data=[
        SearchResult(
            url="https://example.com/recipe",
            title="Recipe",
            description="Description...",
            content=None
        )
    ]
)

最佳实践

1. 选择合适的搜索方法

• 需要内容时使用 search() 并设置 return_content=True
• 快速预览使用 instant_search()

1. 合理设置 limit

• 即时搜索最大 10 条
• 标准搜索最大 50 条

1. 地理位置优化

• 根据目标受众设置正确的地理位置
• 支持多语言内容定向

1. 错误处理

• 实现重试机制应对临时网络问题
• 设置合理的超时时间

1. 异步应用

• 在异步环境中优先使用异步方法
• 配合 asyncio 事件循环使用

概述

AI-Map 是 Oxylabs AI Studio SDK 中的站点地图映射模块，通过 AI 技术自动发现并映射网站结构。该模块能够根据指定的关键词或自然语言提示，在给定域名的不同路径层级中查找相关 URL，帮助用户快速建立目标站点的链接图谱。

AI-Map 的核心价值在于将传统的被动 sitemap 解析与主动的智能发现相结合。它不仅能够读取标准的站点地图文件，还可以通过关键词匹配和 AI 驱动的路径预测，发现那些在 sitemap 中未被收录但实际存在的页面。

核心功能

AI-Map 模块提供以下主要能力：

快速开始

基础调用示例

from oxylabs_ai_studio.apps.ai_map import AiMap

ai_map = AiMap(api_key="<API_KEY>")
payload = {
    "url": "https://career.oxylabs.io",
    "search_keywords": ["career", "jobs", "vacancy"],
    "user_prompt": "job ad pages",
    "max_crawl_depth": 2,
    "limit": 10,
    "geo_location": "Germany",
    "render_javascript": False,
    "include_sitemap": True,
    "max_credits": None,
    "allow_subdomains": False,
    "allow_external_domains": False,
}
result = ai_map.map(**payload)
print(result.data)

资料来源：readme.md

工作流程

graph TD
    A[输入 URL 和配置] --> B[初始化爬取任务]
    B --> C{include_sitemap 配置}
    C -->|是| D[解析并加载 sitemap]
    C -->|否| E[跳过 sitemap]
    D --> F[生成初始 URL 队列]
    E --> F
    F --> G[按 max_crawl_depth 层级爬取]
    G --> H{search_keywords 过滤}
    H -->|匹配| I[URL 进入结果集]
    H -->|不匹配| J[丢弃]
    I --> K{limit 检查}
    K -->|未达上限| G
    K -->|已达上限| L[返回映射结果]
    J --> G

参数说明

必需参数

可选参数

配置详解

深度爬取策略

max_crawl_depth 参数控制爬取遍历的层级深度。合理的深度设置对结果质量和资源消耗有显著影响：

关键词过滤机制

search_keywords 参数支持多关键词匹配，URL 路径中包含任一关键词的页面将被保留：

payload = {
    "url": "https://example.com",
    "search_keywords": ["product", "category", "search"],
    # ...
}

资料来源：readme.md

域名访问控制

通过 allow_subdomains 和 allow_external_domains 两个布尔参数，可以精确控制爬取范围：

Sitemap 集成

include_sitemap 参数默认为 True，这意味着模块会自动尝试从 域名/sitemap.xml 获取已知的 URL 作为种子。如果网站没有 sitemap 文件或文件不完整，建议同时配置 search_keywords 以确保发现更多相关页面。

返回结果

AI-Map 的 map() 方法返回 AiMapJob 对象，包含以下属性：

访问结果数据的方式：

result = ai_map.map(**payload)
print(result.data)

使用场景

场景一：职位页面发现

当需要在招聘网站中查找所有职位相关页面时：

payload = {
    "url": "https://career.oxylabs.io",
    "search_keywords": ["career", "jobs", "vacancy"],
    "user_prompt": "job ad pages",
    "max_crawl_depth": 2,
    "limit": 10,
    "geo_location": "Germany",
    "render_javascript": False,
    "include_sitemap": True,
}

资料来源：readme.md

场景二：电商产品分类映射

对于电商站点，需要发现所有产品分类页面：

payload = {
    "url": "https://shop.example.com",
    "search_keywords": ["category", "products", "collections"],
    "user_prompt": "product category pages",
    "max_crawl_depth": 3,
    "limit": 50,
    "include_sitemap": True,
    "allow_subdomains": False,
}

与其他模块的协作

AI-Map 通常作为数据采集流程的起始环节：

graph LR
    A[AI-Map] --> B[发现目标 URL 列表]
    B --> C[AI-Scraper]
    C --> D[提取页面内容]
    D --> E[AI-Crawler]
    E --> F[深度数据采集]

典型工作流：

1. 使用 AI-Map 发现相关页面 - 通过关键词定位目标 URL
2. 使用 AI-Scraper 提取结构化数据 - 对单个或少量页面进行精确提取
3. 使用 AI-Crawler 进行大规模采集 - 对发现的所有相关页面进行批量内容提取

错误处理

AI-Map 抛出的主要异常类型：

最佳实践

1. 渐进式探索：从 max_crawl_depth=1 开始，确认结果符合预期后再逐步增加深度
2. 精准关键词：使用精确的路径关键词可大幅减少无关页面的发现
3. 积分控制：通过 max_credits 设置消费上限，避免意外超支
4. sitemap 优先：大多数站点 sitemap 包含完整结构，建议保持 include_sitemap=True
5. 地理定位：根据目标用户群体设置 geo_location，获取区域性内容

概述

Browser Agent 是 Oxylabs AI Studio Python SDK 中的一个核心模块，提供基于 AI 的浏览器自动化能力。该模块能够控制浏览器执行点击、滚动、导航等操作，并根据自然语言提示完成复杂的数据提取任务。

Browser Agent 的主要特点包括：

• 支持同步和异步两种调用方式
• 内置自动模式识别功能，可根据输出格式自动生成 JSON Schema
• 支持多种输出格式（JSON、Markdown、HTML、Screenshot、CSV、Toon）
• 提供地理位置模拟能力，可设置代理位置
• 支持 JavaScript 渲染
• 可自定义 User-Agent 请求头

资料来源：readme.md

架构设计

组件关系图

graph TD
    A[BrowserAgent] --> B[API 调用层]
    B --> C[轮询机制]
    C --> D{状态检查}
    D -->|processing| E[等待继续轮询]
    D -->|completed| F[返回 BrowserAgentJob]
    D -->|failed| G[返回失败状态]
    E --> D
    B --> H[Async Client]
    H --> I[异步轮询机制]
    I --> J{异步状态检查}
    J -->|processing| K[异步等待]
    J -->|completed| L[返回 BrowserAgentJob]
    J -->|failed| M[返回失败状态]

核心数据模型

Browser Agent 使用 Pydantic 模型定义返回数据结构：

资料来源：agentic_code_guide.md

安装与初始化

安装 SDK

pip install oxylabs-ai-studio

初始化 Browser Agent

from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

API 参数说明

run 方法参数

资料来源：readme.md

generate_schema 方法参数

使用示例

同步调用示例

from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

# 生成 Schema
schema = browser_agent.generate_schema(
    prompt="game name, platform, review stars and price"
)
print("schema: ", schema)

# 执行浏览器任务
prompt = "Find if there is game 'super mario odyssey' in the store. If there is, find the price. Use search bar to find the game."
url = "https://sandbox.oxylabs.io/"
result = browser_agent.run(
    url=url,
    user_prompt=prompt,
    output_format="json",
    schema=schema,
    geo_location="Spain",
)
print(result.data)

资料来源：examples/browser_agent.py

异步调用示例

import asyncio
from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

async def main():
    prompt = "Find if there is game 'super mario odyssey' in the store."
    url = "https://sandbox.oxylabs.io/"
    result = await browser_agent.run_async(
        url=url,
        user_prompt=prompt,
        output_format="json",
        schema={"type": "object", "properties": {"page_url": {"type": "string"}}, "required": []},
    )
    print(result.data)

if __name__ == "__main__":
    asyncio.run(main())

资料来源：agentic_code_guide.md

异步 Schema 生成

import asyncio
from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

async def main():
    schema = await browser_agent.generate_schema_async(
        prompt="game name, platform, review stars and price"
    )
    print("Generated schema:", schema)

if __name__ == "__main__":
    asyncio.run(main())

执行流程

sequenceDiagram
    participant Client
    participant BrowserAgent
    participant API
    
    Client->>BrowserAgent: run(url, user_prompt, output_format, schema)
    BrowserAgent->>API: POST /browser-agent/run
    API-->>BrowserAgent: {run_id, status}
    BrowserAgent->>API: GET /browser-agent/run/data?run_id={run_id}
    API-->>BrowserAgent: status: processing
    Note over BrowserAgent: 轮询等待
    BrowserAgent->>API: GET /browser-agent/run/data?run_id={run_id}
    API-->>BrowserAgent: status: completed
    BrowserAgent-->>Client: BrowserAgentJob(data)

错误处理与超时

超时机制

Browser Agent 内置超时检测，当请求超时时抛出 TimeoutError：

try:
    result = browser_agent.run(
        url=url,
        user_prompt=prompt,
        output_format="json",
        schema=schema,
    )
except TimeoutError as e:
    print(f"请求超时: {e}")

用户取消

用户可通过键盘中断（KeyboardInterrupt）取消正在执行的爬取任务：

try:
    result = browser_agent.run(
        url=url,
        user_prompt=prompt,
        output_format="json",
        schema=schema,
    )
except KeyboardInterrupt:
    logger.info("[Cancelled] Browser agent was cancelled by user.")

资料来源：src/oxylabs_ai_studio/apps/browser_agent.py

返回值结构

成功响应

BrowserAgentJob(
    run_id="xxx-xxx-xxx",
    message=None,
    data=DataModel(
        type="json",
        content={"game_name": "Super Mario Odyssey", "price": "$59.99"}
    )
)

失败响应

BrowserAgentJob(
    run_id="xxx-xxx-xxx",
    message="error_code",
    data=None
)

与其他模块的对比

最佳实践

1. 合理设置 user_prompt：提示词应描述要执行的操作，而非单纯说明要提取的内容
2. 使用 Schema 限制输出：当需要结构化数据时，预先定义 JSON Schema
3. 处理异步操作：对于大量任务，优先使用 run_async 方法提高效率
4. 设置合理的超时时间：根据目标网站响应时间调整超时配置
5. 地理位置选择：根据目标网站的地区限制选择合适的 geo_location 参数

限制与注意事项

• Python 版本要求：3.10 及以上
• 需要有效的 API Key
• 部分功能可能产生额外的 API 调用费用
• 地理位置参数需使用支持的 ISO2 格式或国家规范名称

概述

Oxylabs AI Studio Python SDK 的核心架构采用面向对象继承模式，所有应用模块均继承自统一的基类 OxyStudioAIClient。该架构设计遵循单一职责原则，将 HTTP 通信、日志记录、配置管理和业务逻辑分离到不同的模块中，实现了高度的内聚性和低耦合性。

客户端架构支持以下核心功能：

• 同步与异步双接口设计
• 自动重试与超时处理
• 结构化数据提取
• 代理地理位置定位
• JavaScript 渲染支持

架构分层

graph TD
    subgraph "表现层 (Presentation Layer)"
        A[应用模块: AiCrawler<br>AiScraper<br>AiSearch<br>BrowserAgent<br>AiMap]
    end
    
    subgraph "核心客户端层 (Core Client Layer)"
        B[OxyStudioAIClient]
    end
    
    subgraph "工具层 (Utility Layer)"
        C[logger.py]
        D[settings.py]
        E[utils.py]
    end
    
    subgraph "数据模型层 (Data Model Layer)"
        F[models.py<br>SchemaResponse<br>AiScraperJob<br>BrowserAgentJob]
    end
    
    subgraph "外部服务层"
        G[Oxylabs AI Studio API]
    end
    
    A --> B
    B --> C
    B --> D
    B --> E
    B --> F
    B --> G

核心组件详解

1. OxyStudioAIClient 基类

OxyStudioAIClient 是所有应用模块的父类，位于 src/oxylabs_ai_studio/client.py。该类封装了所有与 API 通信的通用逻辑。

#### 核心职责

#### 继承结构

class OxyStudioAIClient:
    def __init__(self, api_key: str | None = None):
        """初始化客户端实例"""
        pass
    
    def get_client(self) -> httpx.Client:
        """获取 HTTP 客户端实例"""
        pass
    
    def call_api(self, client, url, method, **kwargs) -> httpx.Response:
        """调用 API 的通用方法"""
        pass

所有应用模块继承此基类：

class BrowserAgent(OxyStudioAIClient):
    def __init__(self, api_key: str | None = None):
        super().__init__(api_key=api_key)

class AiScraper(OxyStudioAIClient):
    # 继承相同模式
    pass

class AiCrawler(OxyStudioAIClient):
    # 继承相同模式
    pass

资料来源：src/oxylabs_ai_studio/client.py

2. 应用模块

SDK 提供五个主要应用模块，每个模块对应特定的数据采集场景：

#### 同步与异步接口模式

每个应用模块同时提供同步 (run/scrape/search) 和异步 (run_async/scrape_async/search_async) 接口：

# 同步接口
result = scraper.scrape(url=url, output_format="markdown")

# 异步接口
result = await scraper.scrape_async(url=url, output_format="markdown")

3. 日志系统

日志模块位于 src/oxylabs_ai_studio/logger.py，采用 Python 标准库 logging 实现。

#### 日志级别配置

logger = get_logger(__name__)

logger.info("正在生成 schema")
logger.error(f"API 调用失败: {response.text}")

#### 日志输出目标

• 控制台 (Console)
• 可配置的日志文件

4. 配置管理

src/oxylabs_ai_studio/settings.py 负责管理 SDK 全局配置：

5. 工具函数

src/oxylabs_ai_studio/utils.py 提供通用辅助函数：

• 数据验证
• 响应格式化
• 错误码映射

请求流程

标准请求流程

sequenceDiagram
    participant User as 用户代码
    participant App as 应用模块
    participant Client as OxyStudioAIClient
    participant API as Oxylabs API
    participant Response as 响应数据

    User->>App: 调用方法 (scrape/crawl/search)
    App->>App: 参数验证与格式化
    App->>Client: call_api(client, url, method, body)
    Client->>API: HTTP POST/GET 请求
    API-->>Client: 202 (处理中) 或 200 (完成)
    
    alt 长任务 (状态码 202)
        Client->>Client: 轮询直到完成
        loop 轮询
            Client->>API: GET /status
            API-->>Client: status: processing
        end
    end
    
    Client-->>App: 返回响应数据
    App-->>User: 返回 Job 对象

长任务轮询机制

对于需要较长时间处理的任务（如爬取大量页面），SDK 采用轮询机制：

POLL_INTERVAL_SECONDS = 5
POLL_MAX_ATTEMPTS = 120  # 默认 10 分钟超时

状态转换：

stateDiagram-v2
    [*] --> Submitted: POST /run
    Submitted --> Processing: 202 响应
    Processing --> Processing: 继续轮询
    Processing --> Completed: 200 + status=completed
    Processing --> Failed: status=failed
    Completed --> [*]
    Failed --> [*]

资料来源：src/oxylabs_ai_studio/apps/browser_agent.py:6-7

数据模型

通用响应模型

所有应用模块返回统一的 Job 对象结构：

class AiScraperJob(BaseModel):
    run_id: str
    message: str | None = None
    data: str | dict | None

class BrowserAgentJob(BaseModel):
    run_id: str
    message: str | None = None
    data: DataModel | None = None

class DataModel(BaseModel):
    type: Literal["json", "markdown", "html", "screenshot", "csv"]
    content: dict[str, Any] | str | None

Schema 响应模型

class SchemaResponse(BaseModel):
    schema: dict[str, Any]
    user_prompt: str

认证机制

API 认证采用 API Key 方式，通过 HTTP Header 传递：

headers = {
    "Authorization": f"Bearer {self.api_key}",
    "Content-Type": "application/json"
}

初始化方式：

# 方式一：直接传入
scraper = AiScraper(api_key="your-api-key")

# 方式二：环境变量
# 设置环境变量 OXyLABS_API_KEY
scraper = AiScraper()  # 自动从环境变量读取

错误处理

错误类型

参数验证示例

def scrape(self, ..., output_format: str, schema: dict | None = None):
    if output_format in ["json", "csv", "toon"] and schema is None:
        raise ValueError(
            "schema is required when output_format is json, csv or toon."
        )

高级配置

地理位置定位

通过 geo_location 参数指定代理位置：

result = scraper.scrape(
    url="https://example.com",
    geo_location="Germany",  # 或 "DE" ISO2 格式
)

支持的格式：

• ISO 2-letter 代码："DE", "US", "IT"
• 国家名称："Germany", "United States"

JavaScript 渲染

# 禁用 JS 渲染 (默认)
render_javascript=False

# 启用 JS 渲染
render_javascript=True

# 自动检测
render_javascript="auto"

结构化提取

使用 JSON Schema 定义输出结构：

schema = {
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "price": {"type": "string"},
        "features": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["title", "price"]
}

或使用 Pydantic 模型自动生成 Schema：

from pydantic import BaseModel, Field

class Product(BaseModel):
    title: str = Field(description="产品名称")
    price: str = Field(description="产品价格")

schema = Product.model_json_schema()

使用示例

基础用法

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")
result = scraper.scrape(
    url="https://example.com/products/1",
    output_format="markdown",
)
print(result.data)

异步用法

import asyncio
from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

async def main():
    result = await scraper.scrape_async(
        url="https://example.com/products/1",
        output_format="json",
        schema={"type": "object", "properties": {"title": {"type": "string"}}},
    )
    print(result.data)

asyncio.run(main())

最佳实践

1. API Key 管理：使用环境变量存储 API Key，避免硬编码
2. 错误处理：捕获特定异常类型，进行针对性处理
3. 重试机制：实现指数退避策略处理临时性失败
4. 超时配置：根据任务复杂度合理设置超时时间
5. 异步批量处理：大量请求时使用异步接口提高吞吐量

概述

Oxylabs AI Studio Python SDK 的数据模型是整个SDK架构的核心组成部分，定义了各应用模块返回数据结构的一致性模式。SDK采用Pydantic作为数据验证库，为所有API响应提供类型安全的数据结构。

数据模型系统包含两类核心模型：作业模型（Job Models）和数据模型（Data Models）。作业模型封装了API调用的完整响应信息，包括任务标识符、状态消息和实际数据内容。数据模型则定义了具体输出的格式和结构，支持多种输出类型如JSON、Markdown、HTML、CSV和Screenshot。

核心数据模型架构

模型继承关系

graph TD
    A[BaseModel - Pydantic] --> B[AiSearchJob]
    A --> C[AiScraperJob]
    A --> D[AiCrawlerJob]
    A --> E[AiMapJob]
    A --> F[BrowserAgentJob]
    
    G[DataModel] --> C
    G --> F
    
    H[SchemaResponse] --> I[内部响应模型]

作业模型详解

AiSearchJob 搜索作业模型

AiSearchJob 是搜索功能的返回模型，用于封装SERP（搜索引擎结果页面）搜索操作的结果。该模型继承自Pydantic的BaseModel，确保数据类型的安全性。

class AiSearchJob(BaseModel):
    run_id: str
    message: str | None = None
    data: Any | None = None

资料来源：src/oxylabs_ai_studio/apps/ai_search.py:1-100

搜索功能支持两种模式：标准搜索和即时搜索。即时搜索在满足limit <= 10且return_content=False条件时自动启用，直接返回结果而无需轮询状态。

AiScraperJob 网页抓取作业模型

AiScraperJob 是网页抓取功能的核心返回模型，与搜索模型类似但增加了对多种输出格式的支持。

class AiScraperJob(BaseModel):
    run_id: str
    message: str | None = None
    data: str | dict | None

资料来源：agentic_code_guide.md

AiScraperJob的data字段根据output_format参数呈现不同类型：JSON格式返回字典对象，Markdown格式返回字符串，CSV格式返回逗号分隔的字符串，Screenshot格式返回Base64编码的字符串。

AiCrawlerJob 网站爬取作业模型

AiCrawlerJob 用于封装网站爬取功能的结果，支持批量URL的数据提取。该模型返回的data字段为列表类型，每个元素代表一个页面的提取结果。

class AiCrawlerJob(BaseModel):
    run_id: str
    message: str | None = None
    data: list[Any] | None = None

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:1-100

爬取功能支持自定义Pydantic Schema进行结构化数据提取，可以定义复杂的嵌套数据类型：

class ProxyPlan(BaseModel):
    name: str = Field(description="The name of the proxy plan")
    price: str = Field(description="The price of the proxy plan")
    features: list[str] = Field(description="The features of the proxy plan")

class ProxyPlans(BaseModel):
    proxy_plans: list[ProxyPlan] = Field(description="The proxy plans")

资料来源：examples/crawl_pydantic_schema.py:1-25

AiMapJob 站点映射作业模型

AiMapJob 是站点映射功能的返回模型，用于生成网站结构地图。该模型返回的数据为字典类型，包含URL与关键词的映射关系。

class AiMapJob(BaseModel):
    run_id: str
    message: str | None = None
    data: dict | None = None

资料来源：src/oxylabs_ai_studio/apps/ai_map.py:1-100

BrowserAgentJob 浏览器代理作业模型

BrowserAgentJob 是浏览器代理功能的核心模型，包含DataModel嵌套结构用于封装结构化的页面数据。

class DataModel(BaseModel):
    type: Literal["json", "markdown", "html", "screenshot", "csv"]
    content: dict[str, Any] | str | None

class BrowserAgentJob(BaseModel):
    run_id: str
    message: str | None = None
    data: DataModel | None = None

资料来源：agentic_code_guide.md

Schema生成与响应模型

Schema生成机制

SDK提供智能Schema生成功能，允许用户通过自然语言描述期望的数据结构，系统自动生成对应的JSON Schema。

graph LR
    A[自然语言提示] --> B[generate_schema方法]
    B --> C[POST /scrape/schema]
    C --> D[API响应]
    D --> E[openapi_schema]
    E --> F[用于scrape/crawl]

def generate_schema(self, prompt: str) -> dict[str, Any] | None:
    logger.info("Generating schema")
    body = {"user_prompt": prompt}
    response = self.call_api(
        client=self.get_client(), url="/scrape/schema", method="POST", body=body
    )
    if response.status_code != 200:
        raise Exception(f"Failed to generate schema: {response.text}")
    json_response: SchemaResponse = response.json()
    return json_response.get("openapi_schema", None)

资料来源：src/oxylabs_ai_studio/apps/ai_scraper.py:1-100

SchemaResponse 响应模型

class SchemaResponse(BaseModel):
    openapi_schema: dict[str, Any] | None

Schema生成功能同样支持异步调用，通过generate_schema_async方法实现：

async def generate_schema_async(self, prompt: str) -> dict[str, Any] | None:
    logger.info("Generating schema (async)")
    body = {"user_prompt": prompt}
    async with self.async_client() as client:
        response = await self.call_api_async(
            client=client, url="/crawl/generate-params", method="POST", body=body
        )
        # ...

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py:1-100

输出格式与数据类型映射

支持的输出格式

格式选择指南

graph TD
    A[需要结构化数据?] --> |是| B[output_format=json]
    A --> |否| C[需要渲染后内容?]
    C --> |是| D[output_format=screenshot]
    C --> |否| E[需要原始HTML?]
    E --> |是| F[output_format=html]
    E --> |否| G[output_format=markdown]
    
    B --> H[需要schema参数]
    G --> I[可选schema]

选择输出格式时需考虑以下因素：

1. JSON格式：需要提供schema参数定义数据结构，适合程序化处理
2. Markdown格式：适合内容提取和文本分析，无需schema
3. Screenshot格式：适合需要渲染JavaScript的动态内容
4. CSV格式：适合表格数据导出，需要schema定义列结构

资料来源：readme.md

状态轮询与响应处理

异步任务状态流程

stateDiagram-v2
    [*] --> 提交任务
    提交任务 --> processing: 状态检查
    processing --> processing: 轮询中
    processing --> completed: status=="completed"
    processing --> failed: status=="failed"
    completed --> [*]
    failed --> [*]

所有作业模型都采用轮询机制获取最终结果：

def scrape(self, url: str, ...) -> AiScraperJob:
    # 提交抓取任务
    response = self.call_api(...)
    run_id = response.json()["run_id"]
    
    # 轮询状态直到完成
    while True:
        get_response = self.call_api(...)
        resp_body = get_response.json()
        
        if resp_body["status"] == "completed":
            return AiScraperJob(
                run_id=run_id,
                message=resp_body.get("error_code", None),
                data=resp_body.get("data", None),
            )
        
        if resp_body["status"] == "failed":
            return AiScraperJob(
                run_id=run_id,
                message=resp_body.get("error_code", None),
                data=None,
            )
        
        time.sleep(POLL_INTERVAL_SECONDS)

资料来源：src/oxylabs_ai_studio/apps/ai_scraper.py:1-100

响应状态码处理

错误处理与数据验证

数据验证机制

所有数据模型使用Pydantic进行运行时类型验证：

from pydantic import BaseModel, Field

class ProxyPlan(BaseModel):
    name: str = Field(description="The name of the proxy plan")
    price: str = Field(description="The price of the proxy plan")
    features: list[str] = Field(description="The features of the proxy plan")

资料来源：examples/crawl_pydantic_schema.py:1-25

异常处理策略

最佳实践

Schema设计建议

1. 使用明确的Field描述帮助AI理解字段含义
2. 嵌套模型不要超过3层深度
3. 数组字段使用有意义的复数名词命名
4. 始终指定required字段列表

数据模型使用示例

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

# 使用Pydantic模型定义Schema
schema = ProxyPlans.model_json_schema()

result = crawler.crawl(
    url="https://oxylabs.io",
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=schema,
)

# 安全访问结果
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_pydantic_schema.py:1-30

相关文档

• AI-Scraper使用指南
• 示例代码集合
• API参考文档

概述

Schema 生成功能是 oxylabs AI Studio SDK 中的核心特性之一，它允许用户通过自然语言描述自动生成符合 OpenAPI 规范的 JSON Schema，从而实现结构化数据提取。该功能简化了传统手动编写复杂 Schema 的流程，使开发者能够快速定义期望提取的数据结构。

Schema 生成功能在以下三个主要应用模块中可用：

• AiScraper - 单页面数据抓取
• AiCrawler - 整站爬取
• BrowserAgent - 浏览器代理操作

工作原理

Schema 生成采用自然语言处理技术，将用户的描述性提示转换为标准化的 JSON Schema。生成的 Schema 遵循 OpenAPI 规范，可直接用于指定输出格式为 json、csv 或 toon 时的结构化数据提取。

graph LR
    A[自然语言提示] --> B[Schema 生成 API]
    B --> C[JSON Schema]
    C --> D[结构化数据提取]
    
    style A fill:#e1f5fe
    style B fill:#b3e5fc
    style C fill:#81d4fa
    style D fill:#4fc3f7

使用方法

AiScraper 中的 Schema 生成

AiScraper 模块提供了 generate_schema() 方法，用于为单页面抓取生成 Schema。

同步调用示例：

from oxylabs_ai_studio.apps.ai_scraper import AiScraper

scraper = AiScraper(api_key="<API_KEY>")

# 生成 Schema
schema = scraper.generate_schema(
    prompt="want to parse developer, platform, type, price game title, genre (array) and description"
)
print(f"Generated schema: {schema}")

# 使用生成的 Schema 进行抓取
url = "https://sandbox.oxylabs.io/products/3"
result = scraper.scrape(
    url=url,
    output_format="json",
    schema=schema,
    render_javascript=False,
)
print(result)

参数说明：

生成的 Schema 示例：

{
  "type": "object",
  "properties": {
    "developer": {"type": "string"},
    "platform": {"type": "string"},
    "type": {"type": "string"},
    "price": {"type": "string"},
    "game_title": {"type": "string"},
    "genre": {"type": "array", "items": {"type": "string"}},
    "description": {"type": "string"}
  },
  "required": []
}

资料来源：examples/scrape_generated_schema.py

BrowserAgent 中的 Schema 生成

BrowserAgent 模块同样支持 Schema 生成，适用于需要浏览器交互的场景。

使用示例：

from oxylabs_ai_studio.apps.browser_agent import BrowserAgent

browser_agent = BrowserAgent(api_key="<API_KEY>")

# 生成 Schema
schema = browser_agent.generate_schema(
    prompt="game name, platform, review stars and price"
)
print("schema: ", schema)

# 使用生成的 Schema 执行浏览器操作
prompt = "Find if there is game 'super mario odyssey' in the store. If there is, find the price. Use search bar to find the game."
url = "https://sandbox.oxylabs.io/"
result = browser_agent.run(
    url=url,
    user_prompt=prompt,
    output_format="json",
    schema=schema,
)
print(result.data)

参数说明：

资料来源：examples/browser_agent.py

AiCrawler 中的 Schema 生成

AiCrawler 模块支持为整站爬取任务生成 Schema。

使用示例：

from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

# 生成 Schema
schema = crawler.generate_schema(
    prompt="proxy plans which have name, price, and features",
)
print("schema: ", schema)

# 使用生成的 Schema 进行爬取
url = "https://oxylabs.io"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=schema,
    render_javascript=False,
)
print("Results:")
for item in result.data:
    print(item, "\n")

资料来源：examples/crawl_generated_schema.py

配合 Pydantic 模型使用

生成的 Schema 可以与 Pydantic 模型无缝配合使用，实现类型安全的数据验证。

示例：

from pydantic import BaseModel, Field
from oxylabs_ai_studio.apps.ai_crawler import AiCrawler

crawler = AiCrawler(api_key="<API_KEY>")

class ProxyPlan(BaseModel):
    name: str = Field(description="The name of the proxy plan")
    price: str = Field(description="The price of the proxy plan")
    features: list[str] = Field(description="The features of the proxy plan")

class ProxyPlans(BaseModel):
    proxy_plans: list[ProxyPlan] = Field(description="The proxy plans")

url = "https://oxylabs.io/"
result = crawler.crawl(
    url=url,
    user_prompt="Find all pages with proxy products pricing",
    output_format="json",
    schema=ProxyPlans.model_json_schema(),
    render_javascript=False,
)

资料来源：examples/crawl_pydantic_schema.py

输出格式与 Schema 的关系

Schema 生成功能与输出格式紧密相关。不同的输出格式对 Schema 有不同的要求：

当输出格式为 json、csv 或 toon 时，必须提供有效的 JSON Schema。Schema 生成功能确保用户能够快速获得符合规范的 Schema 定义。

资料来源：agentic_code_guide.md

最佳实践

1. 字段描述清晰化

在编写 prompt 时，应明确指定每个字段的名称和数据类型：

# 推荐：明确指定字段
schema = scraper.generate_schema(
    prompt="extract product name (string), price (number), tags (array of strings), and description (string)"
)

# 不推荐：描述模糊
schema = scraper.generate_schema(
    prompt="extract product info"
)

2. 数组字段明确声明

当需要提取数组类型数据时，应在 prompt 中明确说明：

schema = scraper.generate_schema(
    prompt="game title, genre (array) and description"
)

3. 嵌套对象结构

对于复杂的嵌套数据结构，可以详细描述层级关系：

schema = crawler.generate_schema(
    prompt="proxy plans which have name, price, and features",
)

实现机制

Schema 生成通过调用后端 API /crawl/generate-params 或类似的端点实现：

def generate_schema(self, prompt: str) -> dict[str, Any] | None:
    logger.info("Generating schema")
    body = {"user_prompt": prompt}
    response = self.call_api(
        client=self.get_client(),
        url="/crawl/generate-params",
        method="POST",
        body=body,
    )
    if response.status_code != 200:
        raise Exception(f"Failed to generate schema: {response.text}")
    json_response: SchemaResponse = ...

资料来源：src/oxylabs_ai_studio/apps/ai_crawler.py

错误处理

Schema 生成过程中可能遇到的错误及处理方式：

总结

Schema 生成功能是 oxylabs AI Studio SDK 的一项强大特性，它通过自然语言处理技术自动将用户的数据提取需求转换为标准化的 JSON Schema。该功能显著降低了结构化数据提取的门槛，使开发者无需深入了解 JSON Schema 规范即可实现复杂的数据提取任务。

通过配合使用 generate_schema() 方法和 SDK 的抓取/爬取功能，用户可以快速构建端到端的数据提取工作流，实现从网页内容到结构化数据的自动化转换。

Pitfall Log / 踩坑日志

项目：oxylabs/oxylabs-ai-studio-py

摘要：发现 9 个潜在踩坑项，其中 0 个为 high/blocking；最高优先级：身份坑 - 仓库名和安装名不一致。

1. 身份坑 · 仓库名和安装名不一致

• 严重度：medium
• 证据强度：runtime_trace
• 发现：仓库名 oxylabs-ai-studio-py 与安装入口 oxylabs-ai-studio 不完全一致。
• 对用户的影响：用户照着仓库名搜索包或照着包名找仓库时容易走错入口。
• 建议检查：在 npm/PyPI/GitHub 上确认包名映射和官方 README 说明。
• 复现命令：pip install oxylabs-ai-studio
• 防护动作：页面必须同时展示 repo 名和真实安装入口，避免用户搜索错包。
• 证据：identity.distribution | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | repo=oxylabs-ai-studio-py; install=oxylabs-ai-studio

2. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | README/documentation is current enough for a first validation pass.

3. 维护坑 · 来源证据：v.0.2.19

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题：v.0.2.19
• 对用户的影响：可能影响升级、迁移或版本选择。
• 建议检查：来源显示可能已有修复、规避或版本变化，说明书中必须标注适用版本。
• 防护动作：不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_96a2e22cbea94aac8c2788f763ce7c04 | https://github.com/oxylabs/oxylabs-ai-studio-py/releases/tag/v0.2.19 | 来源类型 github_release 暴露的待验证使用条件。

4. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | last_activity_observed missing

5. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作：下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | no_demo; severity=medium

6. 安全/权限坑 · 存在安全注意事项

• 严重度：medium
• 证据强度：source_linked
• 发现：No sandbox install has been executed yet; downstream must verify before user use.
• 对用户的影响：用户安装前需要知道权限边界和敏感操作。
• 建议检查：转成明确权限清单和安全审查提示。
• 防护动作：安全注意事项必须面向用户前置展示。
• 证据：risks.safety_notes | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | No sandbox install has been executed yet; downstream must verify before user use.

7. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | no_demo; severity=medium

8. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作：issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | issue_or_pr_quality=unknown

9. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | github_repo:1003630893 | https://github.com/oxylabs/oxylabs-ai-studio-py | release_recency=unknown