实时获取 Google 相关股票新闻并完成自动化总结
@TOC
🌈你好呀！我是 是Yu欸🚀 感谢你的陪伴与支持~ 欢迎添加文末好友🌌 在所有感兴趣的领域扩展知识，不定期掉落福利资讯(*^▽^*)
写在最前面
版权声明：本文为原创，遵循 CC 4.0 BY-SA 协议。转载请注明出处。
本次演示围绕 Bright Data 与 Haystack 的集成实操 展开，完整展示了从获取 Bright Data API 密钥、创建 SERP API 与 Web Unlocker API Zone，到在 Haystack 中封装为 Tool 并接入 Agent 的全流程；
同时通过实际示例，演示了 AI Agent 在推理过程中按需调用搜索与网页抓取能力，实时获取 Google 相关股票新闻并完成自动化总结的效果。
实测结果表明，借助 Bright Data，可在无需自行维护代理和反爬逻辑的情况下，稳定获取搜索结果与网页正文数据，适合生产级 AI Agent 场景。
实时获取股票新闻并完成自动化总结
#爬虫API #数据采集 #亮数据 #BrightData #效率工具 #科研 #大数据 #人工智能 #WebScraping #开发者 #数据分析
Bright Data 官方注册活动链接：注册点我，额外赠送30刀试用金
可参考代码
# pip install haystack-ai brightdata-sdkimport 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 # 常见字段优先级forkin["data", "text", "content", "html", "body", "result", "response"]: v= getattr(obj, k, None)ifv is not None: returnv# 有些 SDK 对象实现了 dict-like try: d = dict(obj)# type: ignoreif 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 =[]forrin 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)forqin 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)foruin 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)])formsgin response["messages"]: role = msg._role.value if role =="tool":forcontentin msg._content: # print("=== Tool Call ===")# print(content) print("=== Tool Output ===") print(content.result)elif role =="assistant":forcontentin msg._content: if hasattr(content, "text"): print("=== Assistant Response ===") print(content.text)hello，我是 是Yu欸 。如果你喜欢我的文章，欢迎三连给我鼓励和支持：👍点赞 📁 关注 💬评论，我会给大家带来更多有用有趣的文章。
原文链接 👉 ，⚡️更新更及时。
欢迎大家点开下面名片，添加好友交流。
