集成LangChain

LangChain

什么是 LangChain

通过 langchain-scrapeless 将 Scrapeless 接入 Python LangChain 应用。你可以直接调用工具来搜索或提取数据,也可以将工具注册到智能体,让模型选择工具并使用其返回结果。

先从 Google 搜索快速入门开始,再将其接入智能体,或选择下方的其他产品。

可用工具

此集成涵盖四类产品。请从 langchain_scrapeless 导入工具类。

产品适用场景工具类
Scraping API — Google 搜索获取搜索结果和页面 URLScrapelessDeepSerpGoogleSearchTool
Crawl提取已知 URL 的内容或发现网站中的页面ScrapelessCrawlerScrapeTool, ScrapelessCrawlerCrawlTool
Web Unlocker提取页面,并可配置 JavaScript 渲染等选项ScrapelessUniversalScrapingTool
AI Scraper获取 AI 平台的回答及可用来源ScrapelessAIScraperTool、ScrapelessChatGPTScraperTool 等特定引擎工具,以及 ScrapelessAIScraperResultTool

Web Unlocker 命名说明: 产品名称为 Web Unlocker。在 langchain-scrapeless 0.1.4 中,其 Python 类仍名为 ScrapelessUniversalScrapingTool;导入和编写代码时请使用此名称。

环境配置

你需要 Python 3.10 或更高版本,以及具有所选产品访问权限和可用额度的 Scrapeless API 密钥。直接调用工具只需要此密钥。可选的智能体示例还需要模型提供商的密钥。

本指南适用于 langchain-scrapeless 0.1.4,该版本会安装 Scrapeless Python SDK 1.3.0。智能体示例使用 LangChain 1.x。

安装软件包

创建并激活虚拟环境。

macOS / Linux:

python3 -m venv .venv
source .venv/bin/activate

Windows PowerShell:

py -3 -m venv .venv
.\.venv\Scripts\Activate.ps1

然后安装集成包:

python -m pip install "langchain-scrapeless==0.1.4"
python -m pip check

配置 API 密钥

替换占位符,并在运行 Python 的终端中设置变量。

macOS / Linux:

export SCRAPELESS_API_KEY="YOUR_SCRAPELESS_API_KEY"

Windows PowerShell:

$env:SCRAPELESS_API_KEY = "YOUR_SCRAPELESS_API_KEY"

工具会自动读取 SCRAPELESS_API_KEY。请勿将密钥写入源文件或共享日志。

快速入门:搜索 Google

将以下代码保存为 search.py:

from langchain_scrapeless import ScrapelessDeepSerpGoogleSearchTool

tool = ScrapelessDeepSerpGoogleSearchTool(handle_tool_error=False)
result = tool.invoke({
    "q": "Scrapeless documentation",
    "hl": "en",
    "gl": "us",
})

items = result.get("organic_results") or []
if not items:
    print("No organic results returned. Inspect the response:", result)
for item in items[:3]:
    print(item.get("title"), item.get("link"))
    print(item.get("snippet", ""))

在已激活的环境中运行:

python search.py

有自然搜索结果时,脚本最多输出三条结果的标题、URL 和摘要。hl 指定搜索语言,gl 指定国家或地区。仅凭空列表无法判定请求失败,请检查完整响应。

搜索摘要是内容概述。需要页面正文时,将返回的 URL 传给 Crawl。示例使用 handle_tool_error=False,使 LangChain 工具异常向上传播,而非转为工具输出。

接入 LangChain 智能体

智能体使用模型选择工具并组织回答,Scrapeless 负责执行搜索。模型提供商的用量与 Scrapeless 用量分别计算。

安装并配置模型集成

在同一环境中安装:

python -m pip install "langchain>=1,<2" "langchain-openai>=1,<2"
python -m pip check

保留 SCRAPELESS_API_KEY 的设置,并添加 OpenAI 凭据。

macOS / Linux:

export OPENAI_API_KEY="YOUR_OPENAI_API_KEY"
export OPENAI_MODEL="YOUR_TOOL_CALLING_MODEL"

Windows PowerShell:

$env:OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
$env:OPENAI_MODEL = "YOUR_TOOL_CALLING_MODEL"

将模型占位符替换为你的账户可用且支持工具调用的模型 ID。OPENAI_MODEL 是本示例显式读取的变量。

创建并运行智能体

保存为 agent.py:

import os

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_scrapeless import ScrapelessDeepSerpGoogleSearchTool

model = ChatOpenAI(model=os.environ["OPENAI_MODEL"])
search_tool = ScrapelessDeepSerpGoogleSearchTool(handle_tool_error=False)
agent = create_agent(model=model, tools=[search_tool])

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": (
            "Search the web for Scrapeless documentation and return three "
            "relevant documentation links with a short description of each."
        ),
    }],
})

# Show the model's tool calls, tool responses, and final answer.
for message in result["messages"]:
    message.pretty_print()

运行 python agent.py。在输出的消息中,查找对 scrapeless_deepserp_google_search 的调用、工具响应,以及基于这些结果生成的最终回答。检查响应中是否包含可用数据;仅有模型回答并不能确认 Scrapeless 已被调用。

注册工具会使其可供使用,但不会强制每次请求都调用它。要添加其他能力,请创建对应的工具实例并将其加入 tools=[...]。

更多产品示例

以下示例相互独立。将每个代码块保存为 .py 文件,并在配置好的环境中运行 python filename.py。

Crawl

对于已知 URL,使用 ScrapelessCrawlerScrapeTool。在 urls 中传入一个或一批 URL:

from langchain_scrapeless import ScrapelessCrawlerScrapeTool

tool = ScrapelessCrawlerScrapeTool(handle_tool_error=False)
result = tool.invoke({
    "urls": ["https://example.com"],
    "formats": ["markdown"],
    "only_main_content": True,
    "poll_timeout": 180,
})

if result.get("status") != "completed":
    raise RuntimeError(f"Scrape did not complete: {result.get('status')}")

pages = result.get("data") or []
print("Returned pages:", len(pages))
for page in pages:
    metadata = page.get("metadata") or {}
    print(metadata.get("sourceURL"), metadata.get("statusCode"))
    print(page.get("markdown", ""))

该调用会等待批量任务完成。除了任务状态,还应检查每个页面的状态和内容。poll_timeout 的单位为秒;独立的页面参数 timeout 的单位为毫秒。

要发现链接指向的页面,请使用 ScrapelessCrawlerCrawlTool,并传入 {"url": "https://example.com", "limit": 5, "formats": ["markdown"]}。它同样会等待完成,并返回包含 status 和 data 的字典。limit 是上限,因此实际返回的页面可能更少。此工具在 0.1.4 版本中不提供 poll_timeout 参数。

Web Unlocker

启用 JavaScript 渲染并请求 Markdown:

from pathlib import Path

from langchain_scrapeless import ScrapelessUniversalScrapingTool

tool = ScrapelessUniversalScrapingTool(handle_tool_error=False)
content = tool.invoke({
    "url": "https://example.com",
    "js_render": True,
    "response_type": "markdown",
})

if not isinstance(content, str) or not content.strip():
    raise RuntimeError("Expected non-empty Markdown text; inspect the response")

Path("page.md").write_text(content, encoding="utf-8")
print("Saved page.md")

打开 page.md 检查提取的正文。文本格式包括 markdown、html 和 plaintext。直接使用返回的文本;它没有 Crawl 的 status 和 data 字段。请将输入 URL 与内容一同保存,作为来源。

AI Scraper

使用特定引擎的工具,或通过 ScrapelessAIScraperTool 选择引擎。支持的引擎值为 chatgpt、perplexity、copilot、gemini、aimode、overview、grok 和 alexa。

以下示例使用 AI Scraper 中的 ChatGPT 引擎:

from langchain_scrapeless import ScrapelessChatGPTScraperTool

tool = ScrapelessChatGPTScraperTool(handle_tool_error=False)
task = tool.invoke({
    "prompt": "Explain HTTP caching and include sources where available.",
    "country": "US",
    "web_search": True,
    "shopping": False,
    "output_format": "json",
})

if task.get("status") != "success":
    raise RuntimeError(f"Task did not complete: {task.get('task_id')}")

answer = task["task_result"]
print(answer.get("result_text", ""))
print("Links:", answer.get("links") or [])
print("Content references:", answer.get("content_references") or [])

ChatGPT 工具默认返回文本;设置 output_format="json" 后,会返回包含任务状态和结果字段的 Python 字典。统一工具和其他特定引擎的工具默认返回完整的任务字典。输入和回答字段因引擎而异,例如 Copilot 需要 mode。来源可能为空,Google AI Overview 也可能在未生成概览的情况下完成任务。

此调用只需要 SCRAPELESS_API_KEY。AI Scraper 用于获取平台回答;智能体示例中的 ChatOpenAI 模型承担不同的角色,并使用独立的密钥。

任务处理

上方大多数示例都会等待结果。对于耗时较长的工作流:

  • AI Scraper: 在引擎输入中设置 wait=False 和 output_format="json",保存 task_id,并使用 ScrapelessAIScraperResultTool 获取结果。传入相同的 engine。查询支持 wait、poll_timeout 和 poll_interval,后两个参数的单位为秒。

  • 批量抓取: 如果轮询超时,请保留错误中的任务 ID。通过 Python SDK 调用 client.scraping_crawl.scrape.check_batch_scrape_status(task_id) 查询现有批量任务。

  • 网站爬取: 使用 SDK 的 async_crawl_url() 提交任务,并使用 check_crawl_status() 查询已保存的任务。状态查询返回带有 status、data 等属性的对象。

这些是远程任务工作流,与 Python 的 async/await 不同。轮询超时不会取消远程任务。再次提交前,请先查询现有 ID。详细用法见 Python SDK 指南和软件包示例。

故障排查

问题检查事项
导入或依赖错误使用 Python 3.10+,并在同一虚拟环境中安装和运行。执行 python -m pip check 和 python -m pip show langchain-scrapeless scrapeless langchain-core。
缺少密钥或身份验证失败在运行 Python 的进程中设置密钥。检查密钥有效性、产品权限和额度。智能体的模型提供商密钥需单独配置。
参数校验错误检查所选工具和引擎的参数结构。对于统一的 AI Scraper 工具,将引擎专属字段放在 input 中。SDK、REST 和工具的参数名称可能不同。
超时保留任务 ID,并按上文说明查询。重复提交可能创建新任务。
输出为空或不符合预期检查完整响应、任务状态和请求的内容字段。报告 Web Unlocker 文本解码错误时,请提供 SDK 版本和已脱敏的异常堆栈。
智能体未执行搜索先验证直接调用的快速入门示例。检查工具注册和模型的工具调用支持,再使用明确的搜索请求检查执行消息。

相关资源