使用 Apify 工具进行网页抓取
本笔记本展示了如何使用 Apify 工具和 AutoGen 代理来从网站上抓取数据并格式化输出。
首先，我们需要安装 Apify SDK 和 AutoGen 库。
! pip install -qqq pyautogen apify-client
还需要设置 LLM 配置和 Apify API 密钥。
import os
config_list = [
{"model": "gpt-4", "api_key": os.getenv("OPENAI_API_KEY")},
]
apify_api_key = os.getenv("APIFY_API_KEY")
让我们定义使用 Apify actor 从网站上抓取数据的工具。在这个 教程章节 中可以了解更多关于工具使用的内容。
from apify_client import ApifyClient
from typing_extensions import Annotated
def scrape_page(url: Annotated[str, "要抓取的网页的URL"]) -> Annotated[str, "抓取到的内容"]:
# 使用你的API令牌初始化ApifyClient
client = ApifyClient(token=apify_api_key)
# 准备Actor的输入
run_input = {
"startUrls": [{"url": url}],
"useSitemaps": False,
"crawlerType": "playwright:firefox",
"includeUrlGlobs": [],
"excludeUrlGlobs": [],
"ignoreCanonicalUrl": False,
"maxCrawlDepth": 0,
"maxCrawlPages": 1,
"initialConcurrency": 0,
"maxConcurrency": 200,
"initialCookies": [],
"proxyConfiguration": {"useApifyProxy": True},
"maxSessionRotations": 10,
"maxRequestRetries": 5,
"requestTimeoutSecs": 60,
"dynamicContentWaitSecs": 10,
"maxScrollHeightPixels": 5000,
"removeElementsCssSelector": """nav, footer, script, style, noscript, svg,
[role=\"alert\"],
[role=\"banner\"],
[role=\"dialog\"],
[role=\"alertdialog\"],
[role=\"region\"][aria-label*=\"skip\" i],
[aria-modal=\"true\"]""",
"removeCookieWarnings": True,
"clickElementsCssSelector": '[aria-expanded="false"]',
"htmlTransformer": "readableText",
"readableTextCharThreshold": 100,
"aggressivePrune": False,
"debugMode": True,
"debugLog": True,
"saveHtml": True,
"saveMarkdown": True,
"saveFiles": False,
"saveScreenshots": False,
"maxResults": 9999999,
"clientSideMinChangePercentage": 15,
"renderingTypeDetectionPercentage": 10,
}
# 运行Actor并等待其完成
run = client.actor("aYG0l9s7dbB7j3gbS").call(run_input=run_input)
# 从运行的数据集中获取并打印Actor的结果（如果有的话）
text_data = ""
for item in client.dataset(run["defaultDatasetId"]).iterate_items():
text_data += item.get("text", "") + "\n"
average_token = 0.75
max_tokens = 20000 # 稍微小 于最大值32k以确保安全
text_data = text_data[: int(average_token * max_tokens)]
return text_data
创建代理并注册工具。
from autogen import ConversableAgent, register_function
# 创建网络爬虫代理。
scraper_agent = ConversableAgent(
"WebScraper",
llm_config={"config_list": config_list},
system_message="你是一个网络爬虫，可以使用提供的工具爬取任何网页。当爬取完成时返回“TERMINATE”。",
)
# 创建用户代理。
user_proxy_agent = ConversableAgent(
"UserProxy",
llm_config=False, # 该代理不需要LLM。
human_input_mode="NEVER",
code_execution_config=False, # 该代理不支持代码执行。
is_termination_msg=lambda x: x.get("content", "") is not None and "terminate" in x["content"].lower(),
default_auto_reply="请继续，如果完成了，请返回“TERMINATE”。",
)
# 使用代理注册函数。
register_function(
scrape_page,
caller=scraper_agent,
executor=user_proxy_agent,
name="scrape_page",
description="爬取网页并返回内容。",
)
开始对网页数据进行爬取的对话。我们使用了reflection_with_llm选项作为summary方法，以便将输出格式化为所需的格式。summary方法在对话完成后调用，给出完整的对话历史记录。
chat_result = user_proxy_agent.initiate_chat(
scraper_agent,
message="你能帮我爬取agentops.ai吗？",
summary_method="reflection_with_llm",
summary_args={
"summary_prompt": """将爬取的内容进行总结，并按照以下格式进行格式化：
---
*公司名称*：
`Acme Corp`
---
*网站*：
`acmecorp.com`
---
*描述*：
`做一些事情的公司。`
---
*标签*：
`制造业。零售业。电子商务。`
---
*要点*：
`通过销售产品为股东提供价值。`
---
*问题*：
`他们销售什么产品？他们如何盈利？他们的市场份额是多少？`
---
"""
},
)
UserProxy（给WebScraper）：
你能帮我爬取agentops.ai的网站吗？
--------------------------------------------------------------------------------
>>>>>>>> 使用自动回复...
WebScraper（给UserProxy）：
***** 建议的工具调用（call_0qok2jvCxOfv7HOA0oxPWneM）：scrape_page *****
参数：
{
"url": "https://www.agentops.ai"
}
****************************************************************************
--------------------------------------------------------------------------------
>>>>>>>> 执行函数scrape_page...
UserProxy（给WebScraper）：
UserProxy（给WebScraper）：
***** 调用工具的响应（call_0qok2jvCxOfv7HOA0oxPWneM） *****
现在开始
通过我们的功能将您的业务提升到新的水平
AI代理很糟糕。
我们正在解决这个问题。
构建符合规定的AI代理，具备可观察性、评估和重放分析。不再有黑盒子和猜测。
新！介绍AgentOps
三行代码。无限测试。
即时测试+调试=可工作的符合规定的AI代理
5
# 程序代码的开头（例如main.py，__init__.py）
6
ao_client = agentops.Client(<在此处插入您的API密钥>)
9
#（可选：记录特定函数）
10
@ao_client.record_action('正在记录的示例函数')
11
def sample_function(...):
15
ao_client.end_session('成功')
从原型到生产
慷慨的免费限制，只在需要时升级。
**********************************************************************
--------------------------------------------------------------------------------
>>>>>>>> 使用自动回复...
WebScraper（给UserProxy）：
当然，这是来自agentops.ai网站的信息：
- 他们的主要价值主张是修复糟糕的AI代理，并用符合规定、可观察的AI代理替代黑盒子和猜测，同时提供评估和重放分析。
- 他们的最新产品是AgentOps。提供简单即时的测试和调试，承诺提供性能更好的符合规定的AI代理。
- 只需三行代码即可轻松集成。
- 他们允许您记录特定函数。
- 他们提供慷慨的免费限制，只有在必要时才需要升级。
这是他们代码的示例：
```python
ao_client = agentops.Client(<在此处插入您的 API 密钥>)
# 可选：记录特定函数
@ao_client.record_action('被记录的示例函数')
def sample_function(...):
...
ao_client.end_session('成功')
```
这段代码是他们库/函数的示例用法。
如果您需要更具体的细节，请告诉我。
--------------------------------------------------------------------------------
UserProxy（给 WebScraper）：
如果还没有完成，请继续，否则返回“终止”。
--------------------------------------------------------------------------------
>>>>>>>> 使用自动回复...
WebScraper（给 UserProxy）：
终止
--------------------------------------------------------------------------------
输出结果存储在摘要中。
print(chat_result.summary)
---
*公司名称*：
`AgentOps`
---
*网站*：
`agentops.ai`
---
*描述*：
`该公司旨在改进 AI 代理。他们提供可观察和可评估的 AI 代理，通过回放分析来替代黑盒模型和盲目提示。`
---
*标签*：
`人工智能，AI 代理，可观察性，分析。`
---
*要点*：
`他们的产品 AgentOps 允许轻松快速地测试和调试 AI 代理。集成只需编写三行代码即可。他们还提供慷慨的免费限额，并且只在必要时强制升级。`
---
*问题*：
`AgentOps 与其他类似产品有何不同？他们的定价如何随使用量变化？他们的“慷慨的免费限额”有哪些详细信息？`
---