LangChain
什么是 LangChain
通过 langchain-scrapeless 将 Scrapeless 接入 Python LangChain 应用。你可以直接调用工具来搜索或提取数据,也可以将工具注册到智能体,让模型选择工具并使用其返回结果。
先从 Google 搜索快速入门开始,再将其接入智能体,或选择下方的其他产品。
可用工具
此集成涵盖四类产品。请从 langchain_scrapeless 导入工具类。
| 产品 | 适用场景 | 工具类 |
|---|---|---|
| Scraping API — Google 搜索 | 获取搜索结果和页面 URL | ScrapelessDeepSerpGoogleSearchTool |
| 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/activateWindows 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 版本和已脱敏的异常堆栈。 |
| 智能体未执行搜索 | 先验证直接调用的快速入门示例。检查工具注册和模型的工具调用支持,再使用明确的搜索请求检查执行消息。 |