演示了如何利用 Bright Data 的 SERP API 和 Web Unlocker 结合 Haystack 框架构建 AI Agent。通过封装异步工具为同步接口，Agent 能够自动搜索 Google 股票新闻并抓取网页内容生成摘要。实测表明该方案无需维护代理即可稳定获取数据，适用于生产级场景。
本次演示围绕 Bright Data 与 Haystack 的集成实操展开，完整展示了从获取 Bright Data API 密钥、创建 SERP API 与 Web Unlocker API Zone，到在 Haystack 中封装为 Tool 并接入 Agent 的全流程。
同时通过实际示例，演示了 AI Agent 在推理过程中按需调用搜索与网页抓取能力，实时获取 Google 相关股票新闻并完成自动化总结的效果。
实测结果表明，借助 Bright Data，可在无需自行维护代理和反爬逻辑的情况下，稳定获取搜索结果与网页正文数据，适合生产级 AI Agent 场景。
import os
import json
import asyncio
import traceback
from typing import Union, List, Any, Optional
from brightdata import BrightDataClient
from haystack.tools import Tool
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
# ========== 你的配置（按你写的，不改'变量名/结构'，但不再硬编码密钥）==========
# ✅ 运行前请在系统环境变量里设置：
# BRIGHTDATA_API_TOKEN=...
# OPENAI_API_KEY=...
#
# Windows PowerShell 示例：
# setx BRIGHTDATA_API_TOKEN "xxx"
# setx OPENAI_API_KEY "yyy"
#
# 然后重开一个终端再运行脚本。
os.environ["BRIGHTDATA_API_TOKEN"] = "XX"
os.environ["OPENAI_API_KEY"] = "sk-XX"
SERP_ZONE_NAME = "serp_api2"
WEB_UNLOCKER_ZONE_NAME = "web_unlocker1" # <-- 换成你控制台真实的 Web Unlocker zone 名
ENABLE_WEB_UNLOCKER = True
# ========== 工具封装：把 async 变 sync（Haystack Tool 只收同步函数）==========
def _run_async(coro):
"""在已存在事件循环（如 Jupyter）中用新 loop 跑；普通脚本里直接 asyncio.run。"""
try:
asyncio.get_running_loop()
new_loop = asyncio.new_event_loop()
try:
return new_loop.run_until_complete(coro)
finally:
new_loop.close()
except RuntimeError:
return asyncio.run(coro)
def _as_error_payload(e: Exception) -> dict:
return {"error": str(e), "traceback": traceback.format_exc(limit=25)}
# ========== 结果提取（避免 outputs_to_string 变 null）==========
def _extract_payload(obj: Any) -> Any:
"""BrightData SDK 可能返回对象/字典。优先取常见字段，否则兜底成字符串。"""
if obj is None:
return None
if isinstance(obj, dict):
return obj
# 常见字段优先级
for k in ["data", "text", "content", "html", "body", "result", "response"]:
v = getattr(obj, k, None)
if v is not None:
return v
# 有些 SDK 对象实现了 dict-like
try:
d = dict(obj)
if d:
return d
except Exception:
pass
return str(obj)
def _results_to_json(results: Any) -> str:
"""Tool 的 outputs_to_string handler：保证永远可 JSON 序列化、并保留 error/traceback。"""
if results is None:
return json.dumps({"error": "BrightData returned None"}, ensure_ascii=False)
if isinstance(results, dict) and "error" in results:
return json.dumps(results, ensure_ascii=False)
if isinstance(results, list):
out = []
for r in results:
if r is None:
out.append(None)
elif isinstance(r, dict) and "error" in r:
out.append(r)
else:
out.append(_extract_payload(r))
return json.dumps(out, ensure_ascii=False)
return json.dumps(_extract_payload(results), ensure_ascii=False)
# ========== SERP 查询（Google Search via BrightData SERP）==========
def google_search_sync(query: Union[str, List[str]], **kwargs):
async def _inner():
try:
async with BrightDataClient(serp_zone=SERP_ZONE_NAME) as c:
if isinstance(query, list):
tasks = [c.search.google(query=q, **kwargs) for q in query]
return await asyncio.gather(*tasks)
else:
return await c.search.google(query=query, **kwargs)
except Exception as e:
return _as_error_payload(e)
return _run_async(_inner())
# ========== Web Unlocker 抓取（修复点：用 scrape_url 而不是 scrape.generic.url）==========
def scrape_url_sync(url: Union[str, List[str]], country: Optional[str] = None):
async def _inner():
try:
async with BrightDataClient(web_unlocker_zone=WEB_UNLOCKER_ZONE_NAME) as c:
if isinstance(url, list):
tasks = [c.scrape_url(u, country=country) for u in url]
return await asyncio.gather(*tasks)
else:
return await c.scrape_url(url, country=country)
except Exception as e:
return _as_error_payload(e)
return _run_async(_inner())
# ========== SERP Tool==========
serp_parameters = {
"type": "object",
"properties": {
"query": {"type": ["string", "array"], "items": {"type": "string"}},
"kwargs": {"type": "object"},
},
"required": ["query"],
}
def serp_api_tool_entry(query: Union[str, List[str]], kwargs: dict = None):
kwargs = kwargs or {}
kwargs.setdefault("num_results", 10)
kwargs.setdefault("language", "en")
return google_search_sync(query, **kwargs)
serp_api_tool = Tool(
name="serp_api_tool",
description="调用 Bright Data SERP API（Google）进行搜索，返回 SERP 结果。",
parameters=serp_parameters,
function=serp_api_tool_entry,
outputs_to_string={"handler": _results_to_json},
)
# ========== Web Unlocker Tool（可选）==========
unlocker_parameters = {
"type": "object",
"properties": {
"url": {"type": ["string", "array"], "items": {"type": "string"}},
"country": {"type": "string"},
},
"required": ["url"],
}
def web_unlocker_tool_entry(url: Union[str, List[str]], country: str = None):
return scrape_url_sync(url, country=country)
web_unlocker_tool = Tool(
name="web_unlocker_tool",
description="调用 Bright Data Web Unlocker 抓取网页内容。",
parameters=unlocker_parameters,
function=web_unlocker_tool_entry,
outputs_to_string={"handler": _results_to_json},
)
# ========== LLM==========
chat_generator = OpenAIChatGenerator(model="gpt-3.5-turbo", api_base_url="https://poloai.top/v1")
tools = [serp_api_tool] + ([web_unlocker_tool] if ENABLE_WEB_UNLOCKER else [])
agent = Agent(chat_generator=chat_generator, tools=tools)
agent.warm_up()
prompt = """
请先用 serp_api_tool 搜索 'Google stock market news' 和 'Alphabet stock market news'，各取 10 条结果。
从 SERP 结果中筛选真正的'新闻文章'链接（避免 quote/股票报价/公司主页），选 3 条主题不同且尽量是最近的。
如果启用了 web_unlocker_tool，则抓取这 3 条链接正文并给出中文摘要（每条 3-5 句）。
如果未启用 web_unlocker_tool，则仅基于 SERP 的 title/description 做保守摘要，并明确说明未抓取正文。
如果工具返回包含 error 字段，请把 error 和 traceback 的关键原因一起输出。
"""
response = agent.run(messages=[ChatMessage.from_user(prompt)])
for msg in response["messages"]:
role = msg._role.value
if role == "tool":
for content in msg._content:
print("=== Tool Output ===")
print(content.result)
elif role == "assistant":
for content in msg._content:
if hasattr(content, "text"):
print("=== Assistant Response ===")
print(content.text)
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online