以编程方式使用文档

Browserbase 是一个开发者平台,用于可靠地运行、管理和监控无头浏览器。

为您的 AI 数据检索提供动力: - 无服务器基础设施 提供可靠的浏览器从复杂 UI 中提取数据 - 隐身模式 包含指纹识别策略和自动验证码解决 - 会话调试器 用于检查浏览器会话的网络时间线和日志 - 实时调试 快速调试您的自动化

安装和设置

  • - 从以下位置获取 API 密钥和项目 ID browserbase.com 并将其设置为环境变量 (BROWSERBASE_API_KEY, BROWSERBASE_PROJECT_ID).
  • - 安装 Browserbase SDK:
pip install browserbase

文档加载器

查看 使用示例.

from langchain_community.document_loaders import BrowserbaseLoader

Deep Agents 集成

Deep Agents 通过将浏览器功能作为 Python 工具公开而不是通过 CLI 路由,与 Browserbase 配合良好。这种集成模式为主规划器提供廉价、无状态的工具用于搜索和页面检索,同时将昂贵的渲染和交互式浏览器工作委托给专门的 browser-specialist subagent.

安装

pip install deepagents browserbase stagehand langchain-openai python-dotenv

架构

推荐的模式使用四个工具:

工具描述使用场景
browserbase_search通过 Browserbase Search API 进行 Web 搜索发现——始终先尝试这个
browserbase_fetch通过 Browserbase Fetch API 获取无头页面静态页面、快速读取、不需要 JS
browserbase_rendered_extract通过 Stagehand 的完整浏览器会话JS 密集型或渲染页面(只读)
browserbase_interactive_task用于多步骤浏览器任务的 Stagehand 代理点击、表单填写、登录流程

廉价工具 (search, fetch) 位于主规划器上。昂贵的浏览器工具位于 browser-specialist 子代理中,将嘈杂的会话输出与主线程隔离。

环境变量

# Optional: model overrides

定义工具

from browserbase import Browserbase
from bs4 import BeautifulSoup
from langchain.tools import tool
from stagehand import AsyncStagehand

@tool
def browserbase_search(query: str, num_results: int = 5) -> str:
    """Search the web with Browserbase. Use this first for discovery before opening pages."""
    bb = Browserbase(api_key=os.environ["BROWSERBASE_API_KEY"])
    response = bb.search.web(query=query, num_results=max(1, min(num_results, 10)))
    results = [
        {"title": r.title, "url": r.url}
        for r in getattr(response, "results", [])
    ]
    return str({"query": query, "results": results})


@tool
def browserbase_fetch(url: str, use_proxy: bool = False, max_chars: int = 12000) -> str:
    """Fetch page content without a browser session. Best for static pages and quick reads."""
    bb = Browserbase(api_key=os.environ["BROWSERBASE_API_KEY"])
    response = bb.fetch_api.create(url=url, proxies=use_proxy)
    content = str(getattr(response, "content", ""))
    soup = BeautifulSoup(content, "html.parser")
    for tag in soup(["script", "style"]):
        tag.decompose()
    text = (soup.body or soup).get_text("\n", strip=True)[:max_chars]
    return str({"url": url, "status_code": response.status_code, "text": text})


@tool
def browserbase_rendered_extract(start_url: str, instruction: str) -> str:
    """Open a full Browserbase browser session and extract rendered content with Stagehand."""
    return asyncio.run(_rendered_extract(start_url, instruction))


@tool
def browserbase_interactive_task(start_url: str, task: str) -> str:
    """Open a Browserbase-hosted Stagehand session and execute a multi-step browser task."""
    return asyncio.run(_interactive_task(start_url, task))


async def _rendered_extract(start_url: str, instruction: str) -> str:
    client = AsyncStagehand(browserbase_api_key=os.environ["BROWSERBASE_API_KEY"])
    resp = await client.sessions.start(model_name=os.getenv("STAGEHAND_MODEL", "google/gemini-3-flash-preview"))
    session_id = resp.data.session_id
    try:
        await client.sessions.navigate(id=session_id, url=start_url, frame_id="")
        result = await client.sessions.extract(id=session_id, instruction=instruction)
        return str({"session_id": session_id, "result": getattr(getattr(result, "data", None), "result", None)})
    finally:
        await client.sessions.end(id=session_id)


async def _interactive_task(start_url: str, task: str) -> str:
    model = os.getenv("STAGEHAND_AGENT_MODEL", "anthropic/claude-sonnet-4-6")
    client = AsyncStagehand(browserbase_api_key=os.environ["BROWSERBASE_API_KEY"])
    resp = await client.sessions.start(model_name=model)
    session_id = resp.data.session_id
    try:
        await client.sessions.navigate(id=session_id, url=start_url, frame_id="")
        result = await client.sessions.execute(
            id=session_id,
            execute_options={"instruction": task, "max_steps": 20},
            agent_config={"model": model, "instructions": "Execute the browser task precisely and stop when done."},
            timeout=300.0,
        )
        return str(result)
    finally:
        await client.sessions.end(id=session_id)

构建代理

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

BROWSER_SUBAGENT = {
    "name": "browser-specialist",
    "description": "Handles JS-heavy browsing, rendered extraction, and interactive browser tasks through Browserbase.",
    "system_prompt": (
        "You are a Browserbase browsing specialist. "
        "Use browserbase_rendered_extract for read-only work on rendered pages. "
        "Use browserbase_interactive_task only for stateful actions such as clicking, typing, or submitting forms. "
        "Return concise summaries with the relevant page URL and whether the task succeeded."
    ),
    "tools": [browserbase_rendered_extract, browserbase_interactive_task],
}

agent = create_deep_agent(
    model=ChatOpenAI(model="gpt-5.5", api_key=os.environ["OPENAI_API_KEY"]),
    tools=[browserbase_search, browserbase_fetch],
    subagents=[BROWSER_SUBAGENT],
    system_prompt=(
        "You are a research-oriented Deep Agent with Browserbase tools. "
        "Start with browserbase_search for discovery. "
        "Prefer browserbase_fetch for quick reads of static pages. "
        "Delegate JS-heavy or interactive work to the browser-specialist subagent."
    ),
    # Gate interactive browser actions behind human approval
    interrupt_on={
        "browserbase_interactive_task": {
            "allowed_decisions": ["approve", "edit", "reject"]
        }
    },
    checkpointer=MemorySaver(),
)

在人工审批下运行

interrupt_on 在任何 browserbase_interactive_task 调用前暂停执行,以便您可以在代理提交之前审查提议的操作。

from langgraph.types import Command

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Research the Browserbase Fetch API and explain when to escalate to a full browser session."}]},
    config=config,
    version="v2",
)

# Resume after each interrupt (approve / edit / reject)
while result.interrupts:
    interrupt_value = result.interrupts[0].value
    for action in interrupt_value["action_requests"]:
        print(f"Pending: {action['name']}\nArgs: {action['args']}")
    decision = input("Decision [approve/edit/reject]: ").strip().lower()
    result = agent.invoke(
        Command(resume={"decisions": [{"type": decision}]}),
        config=config,
        version="v2",
    )

决策树:使用哪个工具

User query
  └─ Is a specific URL already known?
       ├─ No  → browserbase_search (discover URLs first)
       └─ Yes ↓
            Is the page JavaScript-rendered?
              ├─ No  → browserbase_fetch (fast, stateless)
              └─ Yes ↓
                   Does the task require interaction (click / login / form)?
                     ├─ No  → browserbase_rendered_extract (read-only full browser)
                     └─ Yes → browserbase_interactive_task (Stagehand agent, triggers interrupt_on)

完整示例

请参阅 deepagents 仓库中的完整可运行示例.