如何使用 Haystack 集成 Bright Data

import os
from haystack import Pipeline, Document
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.embedders import OpenAIDocumentEmbedder, OpenAITextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.dataclasses import ChatMessage
from haystack_brightdata import BrightDataWebScraper
import json

# 设置 API 密钥
os.environ["BRIGHT_DATA_API_KEY"] = "your-brightdata-api-key"
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 初始化组件
scraper = BrightDataWebScraper()
document_store = InMemoryDocumentStore()
docs_embedder = OpenAIDocumentEmbedder()
text_embedder = OpenAITextEmbedder()
retriever = InMemoryEmbeddingRetriever(document_store)
generator = OpenAIChatGenerator()

# 从多个 Amazon 产品爬取产品数据
product_urls = [
    "https://www.amazon.com/dp/B0DRWBJDLJ",
    "https://www.amazon.com/dp/B08B8M5JGN",
    "https://www.amazon.com/dp/B09WTTWH1R",
]

documents = []
for url in product_urls:
    result = scraper.run(dataset="amazon_product", url=url)

    # 解析响应
    if isinstance(result["data"], str):
        product_data = json.loads(result["data"])
    else:
        product_data = result["data"]

    if not isinstance(product_data, list):
        product_data = [product_data]

    for product in product_data:
        content_parts = [
            f"产品: {product.get('title', 'N/A')}",
            f"品牌: {product.get('brand', 'N/A')}",
            f"价格: ${product.get('final_price', 'N/A')} {product.get('currency', '')}",
            f"评级: {product.get('rating', 0)}/5",
            f"评论数: {product.get('reviews_count', 0)}",
        ]

        if product.get('description'):
            content_parts.append(f"描述: {product.get('description')}")

        if product.get('features'):
            features_text = '\n  - '.join(product.get('features', []))
            content_parts.append(f"特性:\n  - {features_text}")

        content = '\n'.join(content_parts)

        documents.append(Document(
            content=content,
            meta={
                "url": product.get('url', url),
                "title": product.get('title', ''),
                "price": product.get('final_price', 0),
                "rating": product.get('rating', 0),
            }
        ))

# 嵌入并存储文档
embeddings = docs_embedder.run(documents)
document_store.write_documents(embeddings["documents"])

# 使用 ChatPromptBuilder 创建 RAG 管道
messages = [
    ChatMessage.from_system("您是一位有帮助的购物助手。"),
    ChatMessage.from_user("""
背景:
{% for document in documents %}
    {{ document.content }}
{% endfor %}

问题: {{question}}
""")
]

prompt_builder = ChatPromptBuilder(template=messages)

# 构建并连接管道
pipe = Pipeline()
pipe.add_component("embedder", text_embedder)
pipe.add_component("retriever", retriever)
pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("llm", generator)

pipe.connect("embedder.embedding", "retriever.query_embedding")
pipe.connect("retriever", "prompt_builder.documents")
pipe.connect("prompt_builder", "llm")

# 询问关于产品的问题
question = "哪个产品的评级最高？"
response = pipe.run({
    "embedder": {"text": question},
    "prompt_builder": {"question": question}
})

print(f"答案: {response['llm']['replies'][0].text}")

import os
from haystack import Pipeline, Document
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.embedders import OpenAIDocumentEmbedder, OpenAITextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.dataclasses import ChatMessage
from haystack_brightdata import BrightDataSERP, BrightDataUnlocker
import json

# 设置 API 密钥
os.environ["BRIGHT_DATA_API_KEY"] = "your-brightdata-api-key"
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 初始化组件
serp = BrightDataSERP()
unlocker = BrightDataUnlocker(default_output_format="markdown", zone="unblocker")
document_store = InMemoryDocumentStore()
docs_embedder = OpenAIDocumentEmbedder()
text_embedder = OpenAITextEmbedder()
retriever = InMemoryEmbeddingRetriever(document_store)
generator = OpenAIChatGenerator(model="gpt-4")

# 搜索信息
search_query = "生产中机器学习的最佳实践"
search_result = serp.run(query=search_query, num_results=5)
search_data = json.loads(search_result["results"])

# 从搜索结果中提取 URL
urls = []
for result in search_data.get("organic", [])[:5]:
    url = result.get("url") or result.get("link")
    if url:
        urls.append(url)

# 从每个 URL 获取内容
documents = []
for url in urls:
    try:
        result = unlocker.run(url=url, output_format="markdown")
        content = result["content"]
        documents.append(Document(
            content=content,
            meta={"url": url}
        ))
    except Exception as e:
        print(f"无法获取 {url}: {e}")

# 嵌入并存储文档
embeddings = docs_embedder.run(documents)
document_store.write_documents(embeddings["documents"])

# 创建 RAG 管道
messages = [
    ChatMessage.from_system("您是一位知识丰富的 AI 助手。"),
    ChatMessage.from_user("""
来自网��资源的背景:
{% for document in documents %}
    来源: {{ document.meta.url }}
    {{ document.content }}
{% endfor %}

问题: {{question}}
""")
]

prompt_builder = ChatPromptBuilder(template=messages)

pipe = Pipeline()
pipe.add_component("embedder", text_embedder)
pipe.add_component("retriever", retriever)
pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("llm", generator)

pipe.connect("embedder.embedding", "retriever.query_embedding")
pipe.connect("retriever", "prompt_builder.documents")
pipe.connect("prompt_builder", "llm")

# 提出问题
question = "在生产中部署 ML 模型的主要挑战是什么？"
response = pipe.run({
    "embedder": {"text": question},
    "prompt_builder": {"question": question}
})

print(f"答案: {response['llm']['replies'][0].text}")