검색 강화 생성 (RAG) 애플리케이션을 구축 중이라면, 튜토리얼의 정적인 PDF가 아닌 주제에 대한 최신 데이터가 필요합니다. 하지만 실제 기사를 스크래핑하려면 봇 방지 장벽과 차단된 요청을 극복해야 합니다. 데이터가 있다 해도 여전히 데이터를 분할하고, 임베딩하고, 색인화하며, 검색 기능을 연결해야 합니다.
이 튜토리얼은 이 모든 과정을 처리합니다. Bright Data가 주제에 관한 기사를 찾아 스크래핑하고, Weaviate가 이를 저장 및 검색하며, 여러분은 단 하나의 Python 스크립트로 인용 출처가 포함된 답변을 얻을 수 있습니다.
요약
어떤 주제든 검색 가능한 질문-답변 지식 기반으로 전환하세요. 구식 훈련 데이터 대신 실시간 웹 데이터를 활용합니다.
- Bright Data SERP API는 주제에 맞는 실제 기사 URL을 찾아내고, Web Unlocker는 이를 크롤링합니다(봇 방지 기능이 적용된 사이트도 포함).
- Weaviate는 Cohere를 통해 청크를 자동 벡터화하고, 하이브리드 검색으로 색인을 생성하며, 단 한 번의 API 호출로 인용된 답변을 생성합니다.
python3 pipeline.py를실행하고 주제를 입력하면 몇 분 안에 인용된 RAG 답변을 얻을 수 있습니다.- GitHub에서 전체 소스 코드를 확인하세요 – 클론하여 실행해 보세요
API 키를 받아 직접 주제를 입력해 보세요.
최종 출력 결과는 다음과 같습니다:
3~5분 내에 파이프라인을 실행하세요
이미 API 키가 있다면 지금 바로 파이프라인을 실행하세요:
# 1. 저장소 복제 (Python 3.10 이상 필요)
git clone https://github.com/triposat/weaviate-bright-data-rag.git
cd weaviate-bright-data-rag
# 2. 종속성 설치
pip3 install -r requirements.txt
# 3. .env 파일 생성
cp .env.example .env
# .env 파일을 편집하고 API 키를 입력하세요(아래 "API 키 받기" 참조)
# 4. 실행
python3 pipeline.py파이프라인은 주제를 요청하고 Bright Data 존을 자동으로 검색합니다. 실제 기사를 찾아 스크래핑합니다. 기사를 쪼개어 Weaviate에 저장하고(Cohere를 통해 자동 벡터화), 데모 쿼리를 실행한 후, 사용자가 직접 질문할 수 있도록 대화형 모드로 전환합니다.
API 키 받기 (무료 시작 가능)
각 서비스별로 1개씩, 총 3개의 API 키가 필요합니다. Cohere와 Weaviate는 신용카드 정보가 필요하지 않으며, Bright Data는 가입 시 무료 체험 크레딧을 제공합니다.
1. Bright Data API 키
API 키와 2개의 존을 생성하세요:
- brightdata.com에서 가입하세요
- 계정 설정 → 사용자 및 API 키로 이동
- 새 API 키 생성 → 복사 →
.env파일의BRIGHT_DATA_API_TOKEN값으로 붙여넣기
파이프라인에는 SERP API와 Web Unlocker라는 2개의 존이 필요합니다. 프록시 및 스크래핑 → 내 존에서 이미 생성되어 있는지 확인하세요. 보이지 않는다면 다음과 같이 생성하세요:
- ‘프록시 및 스크래핑 ‘ → ‘내 존‘으로 이동
- 추가 선택 → 존 유형으로 SERP API 선택 → 임의의 이름 지정(예:
serp) → 저장 - 다시 ‘추가’를 선택 → 존 유형으로 ‘Unlocker API ‘ 선택 → 임의의 이름 지정(예:
unlocker) → 저장
존 이름이나 비밀번호를 복사할 필요는 없습니다. 파이프라인이 API 키를 사용하여 자동으로 감지합니다.
2. Cohere API 키 (무료)
이 파이프라인에서 Cohere는 임베딩과 생성을 모두 처리합니다:
- dashboard.cohere.com으로 이동
- Google, GitHub 또는 이메일로 가입하세요 – 신용카드 필요 없음
- 대시보드에 체험용 API 키가 표시됩니다. 이를 복사하세요
- 체험판 플랜은 사용량 제한이 있지만 넉넉한 편입니다(자동 실행은 20회 미만의 호출을 사용하며, 대화형 질문 하나당 2회씩 추가됩니다)
3. Weaviate Cloud 자격 증명(무료)
벡터를 저장하고 쿼리할 수 있는 무료 샌드박스 클러스터를 생성하세요:
- console.weaviate.cloud로 이동하세요
- Google 또는 GitHub로 가입하세요
- 클러스터 생성(Create Cluster ) 선택 → 샌드박스(무료) 선택 → 리전 선택 → 생성
- 약 30초 정도 기다린 후, 클러스터를 선택 → ‘Details’ 탭
- REST 엔드포인트 (클러스터 URL)와 API 키를 복사하세요
참고: 샌드박스 클러스터는 14일 후에 만료됩니다. 클러스터가 만료되면 새 클러스터를 생성하고 .env 파일의 URL과 키를 업데이트하세요. pipeline.py를 다시 실행하여 데이터를 다시 가져오세요.
3가지 키를 모두 확보했다면, “3~5분 내에 파이프라인 실행” 섹션으로 돌아가 복제/설치 단계를 따르세요.
RAG 파이프라인의 전체 작동 방식
파이프라인은 데이터 수집, 처리, 벡터 저장, 생성이라는 4단계로 구성됩니다:
각 단계에서는 다음과 같은 API 호출을 수행합니다:
| 단계 | 실행 내용 | 소요 시간 | API 호출 |
|---|---|---|---|
| 1. 검색 + 스크래핑 | Bright Data SERP + Web Unlocker | 약 2–3분 | 2개의 SERP + 6개의 스크래핑 요청 |
| 2. 처리 + 청크 | 로컬 (BeautifulSoup + chunker) | 1초 미만 | 0 |
| 3. 임베드 + 저장 | Weaviate → Cohere embed-v4.0 | 약 30–60초 | ~150–400개 임베딩 (배치 처리) |
| 4. 쿼리 (3가지 데모) | Weaviate → Cohere command-a-03-2025 | 쿼리당 약 5초 | 쿼리당 검색 1회 + 생성 1회 |
파이프라인에서 Bright Data의 역할
Bright Data는 웹 데이터 플랫폼입니다. 이 파이프라인에서 Bright Data는 두 가지 역할을 수행합니다:
| 제품 | 이 파이프라인에서 수행하는 작업 |
|---|---|
| SERP API | 주제를 입력하면 SERP API가 Google을 검색하여 실제 기사 URL을 반환합니다. 하드코딩된 URL이 필요하지 않습니다 |
| 웹 언락커 | 주제당 6개의 기사를 수집하며, 봇 방지 기능이 있는 사이트도 포함됩니다. 각 기사의 길이는 20만~180만 자입니다 |
이 파이프라인은 SERP API와 Web Unlocker를 사용합니다. 다른 데이터 수집 방식에 대해서는 Bright Data의 전체 제품 목록을 참조하십시오.
RAG에 Bright Data를 사용하는 이유
RAG용 스크래핑 시 고려해야 할 몇 가지 사항은 다음과 같습니다:
- 신뢰할 수 있는 스크래핑. Web Unlocker는 재시도, IP 로테이션, 브라우저 지문 인식 기능을 자동으로 처리하므로, 실행 도중 봇 방지 페이지에서 파이프라인이 중단되지 않습니다.
- LLM에 바로 활용 가능한 출력. Crawl API는 원시 HTML 대신 정리된 마크다운을 반환하므로 임베딩 파이프라인을 위한 전처리 과정을 생략할 수 있습니다(이 튜토리얼에서는 Web Unlocker + BeautifulSoup을 사용하지만, 원시 HTML이 필요하지 않다면 Crawl API가 더 빠른 방법입니다).
- 확장성. 이 튜토리얼에서는 6개의 기사를 스크래핑합니다. 실제 운영 환경에서는 6,000개가 필요할 수도 있습니다. Bright Data의 AI 인프라는 사용자 측에서 코드를 변경할 필요 없이 이러한 규모의 동시 스크래핑을 지원합니다.
- 규정 준수. Bright Data는 GDPR 및 CCPA를 준수하며, 전체 네트워크 액세스 권한을 부여하기 전에 신원 확인을 요구합니다.
파이프라인에서 Weaviate의 역할
Weaviate는 오픈소스 벡터 데이터베이스입니다. 단일 API 호출로 검색과 생성을 모두 수행하므로, LLM을 별도로 호출할 필요가 없습니다.
여기서 Weaviate는 스크래핑된 데이터 조각을 저장하고 Cohere를 통해 벡터화합니다. 쿼리를 실행하면 하이브리드 검색을 수행하고 생성형 검색 API를 통해 답변을 생성합니다.
| 기능 | 이 파이프라인에서 작동 방식 |
|---|---|
| 하이브리드 검색 | 조정 가능한 알파 매개변수를 통해 의미 벡터(70%)와 BM25 키워드 매칭(30%)을 결합합니다 |
| 통합 생성형 검색 | 단일 generate.hybrid() 호출로 상위 5개 청크를 검색하고 인용된 답변을 생성합니다 |
| 자동 벡터화 | Weaviate는 데이터 가져오기 시 자동으로 Cohere 임베딩 API를 호출하므로, 사용자가 별도의 임베딩 코드를 작성할 필요가 없습니다 |
| 메타데이터 필터링 | 각 청크와 함께 소스 URL, 도메인, 스크래핑 타임스탬프, 콘텐츠 유형을 저장합니다 |
대규모 환경에서의 Weaviate
Weaviate에는 이 파이프라인에서는 사용하지 않지만 대규모 환경에서 중요한 기능들도 있습니다:
- BSD 3조항 라이선스 적용 – 필요 시 자체 호스팅 또는 포크 가능
- 다양한 배포 옵션 – Weaviate Cloud(무료 샌드박스), 전용 클라우드, 자체 호스팅 쿠버네티스
- 다중 테넌시 – SaaS 애플리케이션을 위한 노드당 50,000명 이상의 테넌트 지원
- 회전 양자화 – 98~99%의 리콜률로 벡터 4배 압축
RAG 파이프라인을 단계별로 구축하기
아래의 각 단계는 pipeline.py의 핵심 로직을 보여줍니다. 전체 소스 코드는 GitHub에 있습니다.
프로젝트 설정 및 임포트
먼저 종속성을 임포트하고 .env 파일에서 자격 증명을 불러옵니다:
import os
import sys
import time
import hashlib
import requests
import urllib3
from urllib.parse import quote
from datetime import datetime, timezone
from dotenv import load_dotenv
from bs4 import BeautifulSoup
import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
urllib3.disable_warnings()
load_dotenv()
# .env 파일에서 자격 증명 로드
COHERE_API_KEY = os.getenv("COHERE_API_KEY")
WEAVIATE_URL = os.getenv("WEAVIATE_URL")
WEAVIATE_API_KEY = os.getenv("WEAVIATE_API_KEY")
BD_API_TOKEN = os.getenv("BRIGHT_DATA_API_TOKEN")
COLLECTION_NAME = "WebResearch"
def clean_url(url):
"""URL 내의 nbsp 아티팩트 수정 (일부 사이트의 인코딩 문제로 인해 발생)."""
cleaned = url.replace("nbsp", "-")
while "--" in cleaned:
cleaned = cleaned.replace("--", "-")
return cleaned
def clean_generated_text(text):
"""터미널에 표시하기 위해 LLM이 생성한 텍스트를 정리합니다."""
text = text.replace("**", "")
text = text.replace("nbsp", "-")
while "--" in text:
text = text.replace("--", "-")
return text어떤 작업을 수행하기 전에, 파이프라인은 .env 파일에 필요한 모든 자격 증명이 설정되어 있는지 확인합니다:
def validate_env():
"""필요한 모든 환경 변수가 설정되었는지 확인합니다."""
missing = []
if not BD_API_TOKEN:
missing.append("BRIGHT_DATA_API_TOKEN")
if not COHERE_API_KEY:
missing.append("COHERE_API_KEY")
if not WEAVIATE_URL:
missing.append("WEAVIATE_URL")
if not WEAVIATE_API_KEY:
missing.append("WEAVIATE_API_KEY")
if missing:
print("ERROR: .env 파일에 환경 변수가 누락되었습니다:")
for var in missing:
print(f" - {var}")
# ... 예시 .env 형식 출력 ...
print("n각 키를 얻는 방법은 블로그 게시물을 참조하세요 (모두 무료로 시작할 수 있습니다).")
sys.exit(1)존 이름이나 비밀번호를 직접 설정할 필요는 없습니다. 파이프라인이 API 키에서 이를 자동으로 탐지합니다:
def discover_bright_data_credentials():
"""
API 키에서 Bright Data 프록시 자격 증명을 자동으로 탐지합니다.
모든 Bright Data 계정에서 작동합니다. 하드코딩된 값이 필요하지 않습니다.
"""
headers = {"Authorization": f"Bearer {BD_API_TOKEN}"}
# 1. 활성 존 가져오기
zones = requests.get(
"https://api.brightdata.com/zone/get_active_zones", headers=headers
).json()
# 각 유형의 첫 번째 존 선택 (여러 개가 있는 경우 이름을 명시적으로 설정)
zone_names = {}
for z in zones:
if z["type"] not in zone_names:
zone_names[z["type"]] = z["name"]
# "unblocker"는 Web Unlocker 제품의 API 이름입니다
unlocker_zone = zone_names.get("unblocker")
serp_zone = zone_names.get("serp")
# 2. 영역 비밀번호 가져오기
unlocker_pwd = requests.get(
f"https://api.brightdata.com/zone/passwords?zone={unlocker_zone}",
headers=headers,
).json()["passwords"][0]
serp_pwd = requests.get(
f"https://api.brightdata.com/zone/passwords?zone={serp_zone}",
headers=headers,
).json()["passwords"][0]
# 3. 고객 ID 가져오기 (비용 엔드포인트는 {customer_id: cost_data}를 반환함)
cost = requests.get(
f"https://api.brightdata.com/zone/cost?zone={unlocker_zone}",
headers=headers,
).json()
customer_id = list(cost.keys())[0]
return customer_id, unlocker_zone, unlocker_pwd, serp_zone, serp_pwd리포지토리를 클론하고 API 키를 추가하면, 파이프라인이 나머지를 처리합니다.
1단계: Bright Data를 사용하여 기사 찾기 및 스크래핑
이 파이프라인은 SERP API를 사용하여 주제와 관련된 기사 URL을 찾은 다음, Web Unlocker를 통해 각 기사를 스크래핑합니다:
def get_bd_proxy(customer_id, zone, password):
"""Bright Data 프록시 URL 생성."""
proxy = f"http://brd-customer-{customer_id}-zone-{zone}:{password}@brd.superproxy.io:33335"
return {"http": proxy, "https": proxy}
def search_serp(query, customer_id, zone, password, num=10):
"""Bright Data SERP API를 통해 Google을 검색하고 자연 검색 결과를 반환합니다."""
proxies = get_bd_proxy(customer_id, zone, password)
# brd_json=1은 Bright Data가 원시 HTML 대신 구조화된 JSON을 반환하도록 지시합니다.
search_url = f"https://www.google.com/search?q={quote(query)}&brd_json=1&num={num}"
try:
# verify=False는 BD 프록시에 대한 SSL 인증을 우회합니다.
# 프로덕션 환경에서는 대신 Bright Data CA 인증서를 설치하십시오:
# https://docs.brightdata.com/general/account/ssl-certificate
response = requests.get(search_url, proxies=proxies, timeout=30, verify=False)
if response.status_code == 200:
data = response.json()
return [
{
"title": item.get("title", ""),
"url": item.get("link", ""),
"description": item.get("description", ""),
}
for item in data.get("organic", [])
]
except Exception as e:
print(f"SERP 오류: {str(e)[:60]}", end=" ", flush=True)
return []search_serp() 는 쿼리를 Bright Data SERP 프록시를 통해 전송하고 구조화된 JSON(제목, URL, 설명)을 반환합니다. brd_json=1 매개변수는 Bright Data가 Google HTML을 깔끔한 JSON으로 파싱하도록 지시합니다.
다음으로, find_articles_for_topic() 은 주제당 2개의 SERP 쿼리를 실행하고 결과를 필터링하며, scrape_url() 은 Web Unlocker를 통해 각 기사를 가져옵니다:
def find_articles_for_topic(topic, customer_id, serp_zone, serp_pwd):
"""Bright Data SERP API를 사용하여 주제에 대한 실제 기사 URL을 찾습니다."""
search_queries = [
f"{topic} 최신 뉴스 및 트렌드",
f"{topic} 심층 분석 가이드",
]
# 기사 콘텐츠가 아닌 결과(동영상, 피드, 소셜 미디어)를 반환하는 도메인은 건너뜁니다
skip_domains = {
"youtube.com", "twitter.com", "x.com", "facebook.com", "instagram.com",
"reddit.com", "linkedin.com", "wikipedia.org", "amazon.com", "tiktok.com",
}
skip_extensions = (".pdf", ".doc", ".ppt", ".xls", ".zip", ".mp4", ".mp3")
all_urls = []
seen_domains = set()
serp_docs = []
for query in search_queries:
results = search_serp(query, customer_id, serp_zone, serp_pwd, num=10)
if results:
# 전체 스크래핑이 실패하더라도 LLM이 기사 요약을 참조할 수 있도록
# SERP 제목과 설명을 문서로 저장
serp_text = f"Google 검색 결과: {query}nn"
for r in results:
serp_text += f"제목: {r['title']}nURL: {r['url']}n"
serp_text += f"요약: {r['description']}nn"
serp_docs.append({
"url": f"https://google.com/search?q={quote(query)}",
"html": serp_text,
"scraped_at": datetime.now(timezone.utc).isoformat(),
"is_serp": True,
})
# 기사 URL 추출 (다양성을 위해 도메인당 1개)
for r in results:
url = r.get("url", "")
if not url:
continue
domain = url.split("/")[2] if "://" in url else ""
base_domain = ".".join(domain.split(".")[-2:])
if base_domain in skip_domains:
continue
if any(url.lower().endswith(ext) for ext in skip_extensions):
continue
if base_domain in seen_domains:
continue # 다양성을 위해 도메인당 기사 1개만 추출
seen_domains.add(base_domain)
all_urls.append(url)
return all_urls[:6], serp_docs # 상위 6개 URL
def scrape_url(url, customer_id, zone, password, retries=2):
"""자동 재시도 기능을 사용하여 Bright Data Web Unlocker로 URL을 스크래핑합니다."""
proxies = get_bd_proxy(customer_id, zone, password)
# 사용자 정의 헤더가 필요 없음: Web Unlocker가 User-Agent,
# 쿠키 및 지문을 자동으로 관리합니다.
for attempt in range(retries + 1):
try:
# verify=False는 BD 프록시의 SSL 인증을 우회합니다.
# 프로덕션 환경에서는 대신 Bright Data CA 인증서를 설치하십시오:
# https://docs.brightdata.com/general/account/ssl-certificate
response = requests.get(
url, proxies=proxies, timeout=60, verify=False
)
if response.status_code == 200:
return {
"url": url,
"html": response.text,
"scraped_at": datetime.now(timezone.utc).isoformat(),
}
else:
print(f"HTTP {response.status_code}", end=" → ", flush=True)
except Exception as e:
print(f"Error: {str(e)[:60]}", end=" → ", flush=True)
if attempt < retries:
time.sleep(2)
return Nonecollect_data()는 두 단계를 결합합니다 – SERP가 URL을 찾고, Web Unlocker가 이를 크롤링합니다:
def collect_data(topic, customer_id, unlocker_zone, unlocker_pwd, serp_zone, serp_pwd):
"""SERP를 통해 주제에 대한 기사를 찾은 후, Web Unlocker로 스크래핑합니다."""
documents = []
# 1. SERP API를 사용하여 기사 URL 찾기
urls_to_scrape, serp_docs = find_articles_for_topic(
topic, customer_id, serp_zone, serp_pwd
)
if not urls_to_scrape:
return []
# 2. Web Unlocker를 사용하여 찾은 기사 스크래핑
for i, url in enumerate(urls_to_scrape):
domain = url.split("/")[2] if "://" in url else url
print(f" ({i+1}/{len(urls_to_scrape)}) {domain}... ", end="", flush=True)
result = scrape_url(url, customer_id, unlocker_zone, unlocker_pwd)
if result:
documents.append(result)
print(f"OK ({len(result['html']):,} 자)")
else:
print("실패 (건너뜀)")
# 3. SERP 결과를 추가 문서로 추가
documents.extend(serp_docs)
return documents“OpenAI vs Google vs Anthropic AI race”로 실행하면 다음과 같은 결과가 출력됩니다:
[SERP API] 'OpenAI vs Google vs Anthropic AI 경쟁'에 관한 기사 검색 중...
검색: 'OpenAI vs Google vs Anthropic AI 경쟁 최신 뉴스 및 동향'... OK (9개 결과)
검색: 'OpenAI vs Google vs Anthropic AI 경쟁 심층 분석 가이드'... OK (9개 결과)
스크래핑할 기사 URL 6개 발견:
- www.hashmeta.ai
- ramp.com
- shawnkanungo.com
- www.androidheadlines.com
- www.youreverydayai.com
- www.storyboard18.com
[Web Unlocker] 기사 6개 스크래핑 중...
(1/6) www.hashmeta.ai... OK (216,770자)
(2/6) ramp.com... OK (1,421,933자)
(3/6) shawnkanungo.com... OK (224,141자)
(4/6) www.androidheadlines.com... OK (591,022 자)
(5/6) www.youreverydayai.com... OK (450,519 자)
(6/6) www.storyboard18.com... OK (357,053 자)
수집된 문서 총계: 86개 모두 성공적으로 수집됨 – 2개의 SERP 결과 페이지를 포함하여 총 8개의 문서가 수집되었습니다.
Web Unlocker가 URL에 대해 3회 시도 후 실패하면, 파이프라인은 해당 URL을 건너뛰고 나머지 기사들을 처리합니다.
현재 8개의 원본 문서(기사 6개 + SERP 결과 페이지 2개)가 있습니다. 이제 임베딩을 위해 데이터를 정리하고 쪼개세요.
2단계: 데이터 정리 및 청크화
원본 HTML은 약 90%가 노이즈입니다. 처리 단계에서는 이를 제거하여 깨끗한 텍스트로 만들고, 가능한 경우 문장 경계를 따라 500자 단위(약 125 토큰)의 청크로 분할합니다.
청크 크기는 RAG의 핵심적인 절충점을 결정합니다. 작은 청크(200–500자)는 사실별 정밀한 검색 결과를 제공하는 반면, 큰 청크(1000–2000자)는 LLM에 더 많은 주변 맥락을 제공하지만 검색 결과의 잡음이 증가합니다. 500자 기본값은 사실 확인 질문(“기업용 시장에서 Anthropic의 OpenAI 상대 승률은 얼마인가?”)에 적합합니다. 요약이나 비교와 같이 더 넓은 맥락이 필요한 쿼리의 경우 chunk_size를 1500~2000으로 늘리십시오.
50자 중복 설정은 경계에서 정보 손실을 방지합니다. 이 설정이 없으면 두 청크에 걸쳐 있는 문장이 분할되어 어느 청크에도 완전한 문장이 담기지 않게 됩니다.
def clean_html(html, is_serp=False):
"""HTML을 정리하여 텍스트로 변환하고, 내비게이션, 광고 및 상투적인 내용을 제거합니다."""
if is_serp:
return html # SERP 결과는 이미 정리된 텍스트입니다
soup = BeautifulSoup(html, "html.parser")
# 불필요한 요소 제거
for tag in soup(["nav", "footer", "header", "script", "style",
"aside", "iframe", "noscript", "svg", "form", "button"]):
tag.decompose()
# 일반적인 광고/쿠키/팝업 컨테이너 제거
for selector in [".ad", ".ads", ".cookie", ".popup", ".modal", ".sidebar",
"#cookie-banner", "#ad-container", "[role='banner']",
"[role='navigation']", "[role='complementary']"]:
for el in soup.select(selector):
el.decompose()
text = soup.get_text(separator="n", strip=True)
lines = [line.strip() for line in text.splitlines() if line.strip()]
return "n".join(lines)
def chunk_text(text, chunk_size=500, chunk_overlap=50):
"""문장 경계를 기준으로 텍스트를 겹치는 청크로 분할합니다.
overlap 매개변수는 청크 경계에서 문장이 청크 사이에 분실되지 않도록 보장합니다."""
if len(text) <= chunk_size:
return [text]
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
# 문장 경계에서 분할 시도
if end < len(text):
for sep in [". ", ".n", "nn", "n", " "]:
last_sep = text[max(start, end - 100):end].rfind(sep)
if last_sep != -1:
end = max(start, end - 100) + last_sep + len(sep)
break
chunk = text[start:end].strip()
if chunk and len(chunk) > 50:
chunks.append(chunk)
start = end - chunk_overlap
return chunks
def process_documents(documents):
"""모든 문서를 정리하고, 청크로 나누며, 메타데이터를 추가합니다."""
all_chunks = []
for doc in documents:
is_serp = doc.get("is_serp", False)
clean_text = clean_html(doc["html"], is_serp=is_serp)
if len(clean_text) < 100:
continue
chunks = chunk_text(clean_text)
domain = doc["url"].split("/")[2] if "://" in doc["url"] else "unknown"
for i, chunk in enumerate(chunks):
all_chunks.append({
"text": chunk,
"source_url": doc["url"],
"source_domain": domain,
"scraped_at": doc["scraped_at"],
"chunk_index": i,
"total_chunks": len(chunks),
"content_hash": hashlib.md5(chunk.encode()).hexdigest(),
"content_type": "serp_result" if is_serp else "article",
})
return all_chunks처리 후, 8개의 문서가 (기사 길이에 따라) 약 150~400개의 정제된 텍스트 청크로 변환되며, 각 청크에는 메타데이터(출처 URL, 도메인, 타임스탬프, 콘텐츠 해시)가 포함됩니다.
3단계: Weaviate에 임베딩 및 저장
Weaviate Cloud에 연결하고, Cohere 벡터화 기능을 사용하여 컬렉션을 생성한 후, 모든 청크를 일괄 가져옵니다.
def connect_weaviate():
"""시간 초과 설정을 확장하여 Weaviate Cloud에 연결합니다."""
client = weaviate.connect_to_weaviate_cloud(
cluster_url=WEAVIATE_URL,
auth_credentials=Auth.api_key(WEAVIATE_API_KEY),
headers={"X-Cohere-Api-Key": COHERE_API_KEY},
additional_config=weaviate.classes.init.AdditionalConfig(
timeout=weaviate.classes.init.Timeout(init=30, query=60, insert=120),
),
skip_init_checks=True, # 유휴 샌드박스에서 gRPC 타임아웃 방지
)
if not client.is_ready():
print(" ERROR: Weaviate 클러스터가 준비되지 않았습니다.")
print(" .env 파일에서 WEAVIATE_URL 및 WEAVIATE_API_KEY를 확인하십시오")
print(" console.weaviate.cloud에서 샌드박스 클러스터가 실행 중인지 확인하십시오")
sys.exit(1)
return client
def setup_collection(client):
"""하이브리드 검색 + 생성형 구성으로 컬렉션을 생성합니다."""
# 이 이름의 기존 컬렉션을 삭제합니다. 새 토픽으로 다시 실행하면
# 이전 지식 기반이 추가되는 것이 아니라 대체됩니다.
if client.collections.exists(COLLECTION_NAME):
client.collections.delete(COLLECTION_NAME)
print(f" 기존 '{COLLECTION_NAME}' 컬렉션 삭제됨")
client.collections.create(
name=COLLECTION_NAME,
description="RAG용 Bright Data를 통해 수집된 웹 기사",
# Cohere embed-v4.0: 가져올 때 텍스트를 자동으로 벡터화
vector_config=Configure.Vectors.text2vec_cohere(
model="embed-v4.0",
),
# Cohere command-a-03-2025: 쿼리 시점에 RAG 답변 생성
generative_config=Configure.Generative.cohere(
model="command-a-03-2025",
),
properties=[
Property(name="text", data_type=DataType.TEXT,
description="청크 텍스트 콘텐츠"),
Property(name="source_url", data_type=DataType.TEXT,
skip_vectorization=True),
Property(name="source_domain", data_type=DataType.TEXT,
skip_vectorization=True),
Property(name="scraped_at", data_type=DataType.TEXT,
skip_vectorization=True),
Property(name="chunk_index", data_type=DataType.INT,
skip_vectorization=True),
Property(name="total_chunks", data_type=DataType.INT,
skip_vectorization=True),
Property(name="content_hash", data_type=DataType.TEXT,
skip_vectorization=True),
Property(name="content_type", data_type=DataType.TEXT,
skip_vectorization=True),
],
)
print(f" '{COLLECTION_NAME}' 컬렉션 생성됨")주목할 만한 몇 가지 사항:
- 메타데이터 필드에
skip_vectorization=True설정 –텍스트필드만 임베딩되어 API 호출을 줄이고 더 깔끔한 벡터를 생성합니다 content_hash는청크별로 저장됩니다. 증분 재스크래핑 로직을 추가할 때 변경되지 않은 콘텐츠의 재임베딩을 건너뛰는 데 사용하세요(현재 파이프라인은 매 실행 시마다 새로 가져옵니다).
재실행 동작: 파이프라인은 매 실행 시 컬렉션을 삭제하고 새로 생성합니다. “AI race”를 실행한 후 “quantum computing”을 실행하면 AI race 데이터가 대체됩니다. 여러 주제를 유지하려면
COLLECTION_NAME을주제별로 고유한 이름으로 변경하십시오(예:WebResearch_ai_race,WebResearch_quantum).
AI용 벡터 데이터셋 준비에 대한 자세한 내용은 Bright Data 가이드에서 확인하세요.
store_chunks() 함수는 모든 청크를 컬렉션에 일괄 삽입합니다:
def store_chunks(client, chunks):
"""Weaviate로 청크를 일괄 가져옵니다(Cohere를 통해 자동 벡터화됨)."""
collection = client.collections.use(COLLECTION_NAME)
with collection.batch.fixed_size(batch_size=50) as batch:
for chunk in chunks:
batch.add_object(properties=chunk)
failed = len(collection.batch.failed_objects) if collection.batch.failed_objects else 0
if failed > 0:
print(f" 첫 번째 오류: {collection.batch.failed_objects[0].message[:120]}")
return failedbatch.fixed_size(50)은 하나씩 삽입하는 대신 처리량을 높이기 위해 일괄 가져오기를 수행합니다. 테스트 실행에서 모든 청크는 오류 없이 가져오기에 성공했습니다. Weaviate는 가져오기 시점에 각 청크를 임베딩하기 위해 Cohere를 호출합니다.
4단계: 하이브리드 검색 및 생성 쿼리
모든 청크가 임베딩되고 색인된 상태에서 rag_query() 함수를 사용하여 쿼리합니다. 이 함수는 generate.hybrid()를 호출하여 단일 요청으로 검색과 생성을 수행합니다:
def rag_query(client, question, alpha=0.7, limit=5):
"""Weaviate 하이브리드 검색 + 생성형 AI를 사용하여 RAG 쿼리를 실행합니다."""
collection = client.collections.use(COLLECTION_NAME)
response = collection.generate.hybrid(
query=question,
alpha=alpha, # 0.7 = 70% 의미적, 30% 키워드
limit=limit,
grouped_task=f"""아래 검색된 문서를 바탕으로 다음 질문에 답하십시오:
"{question}"
지침:
- 명확하고 포괄적인 답변을 제공하십시오
- 주요 주장마다 출처 URL을 인용하십시오
- 정보가 오래되었거나 상충되는 경우 이를 명시하십시오
- 답변은 간결하면서도 유익하게 작성하십시오 (2~4단락)""",
)
print(f"n Q: {question}")
print(f" {'─' * 60}")
if response.generated:
print(f" A: {clean_generated_text(response.generated)}")
else:
print(" A: (응답 생성되지 않음 — Cohere API 키를 확인하세요)")
# 기사 출처와 SERP 요약 부분을 분리
article_sources = []
serp_sources = []
seen_urls = set()
for obj in response.objects:
url = obj.properties.get("source_url", "unknown")
if url in seen_urls:
continue
seen_urls.add(url)
content_type = obj.properties.get("content_type", "")
domain = obj.properties.get("source_domain", "")
if content_type == "serp_result":
serp_sources.append((domain, url))
else:
article_sources.append((domain, clean_url(url)))
print(f"n 출처 ({len(response.objects)} 개의 청크 검색됨):")
for domain, url in article_sources:
print(f" - [{domain}] {url}")
if not article_sources and serp_sources:
print(" (SERP 요약본 기준 — 일치하는 기사 청크 없음)")
return response순수 벡터 검색은 “GPT-5″나 “Claude Code”와 같은 정확한 용어를 놓칠 수 있습니다. 순수 키워드 검색은 의미적으로 관련된 콘텐츠를 놓칩니다. alpha=0.7 혼합 방식은 두 가지를 모두 제공합니다. Weaviate의 BlockMax WAND 알고리즘은 대규모 환경에서도 BM25 키워드 구성 요소의 속도를 유지합니다.
limit=5로 설정하면 쿼리는 상위 5개 청크를 검색합니다. 이는 LLM에 불필요한 정보로 과부하를 주지 않으면서도 상세한 답변을 작성하기에 충분한 맥락을 제공합니다. 여러 하위 주제를 아우르는 광범위한 질문의 경우 10으로 늘리고, 정확한 사실 확인이 필요한 경우 3으로 줄이세요. grouped_task 매개변수는 검색된 모든 청크를 하나의 프롬프트로 Cohere에 전송하여 단일 답변을 작성할 수 있게 합니다. 대안인 single_prompt는 청크마다 별도의 응답을 생성하므로, 문서별 요약에는 유용하지만 여러 출처에 걸친 답변에는 적합하지 않습니다.
더 많은 옵션을 확인하려면 Bright Data의 시맨틱 검색 API 종합 가이드를 참조하세요.
4단계 모두 결합하기
main() 함수는 전체 파이프라인을 실행합니다. 주제를 선택하면 나머지는 이 함수가 처리합니다:
def main():
print("=" * 65)
print(" RAG 파이프라인: Bright Data + Weaviate")
print(" 주제 스크래핑 → 지식베이스 구축 → 질문하기")
print("=" * 65)
# ── 환경 검증 ──
validate_env()
# ── 사용자에게 주제 요청 ──
print()
try:
topic = input(" 어떤 주제를 조사하시겠습니까? ").strip()
except (EOFError, KeyboardInterrupt):
print("n 안녕히 가세요!")
return
if not topic:
print(" 주제가 입력되지 않았습니다. 종료합니다.")
return
print(f'n 다음 주제에 대한 RAG 지식 기반 구축 중: "{topic}"')
# ── Bright Data 자격 증명 자동 검색 ──
print("n[SETUP] Bright Data에 연결 중...")
cust_id, unlocker_zone, unlocker_pwd, serp_zone, serp_pwd = (
discover_bright_data_credentials()
)
# ── 1단계: 주제에 대한 기사 찾기 및 스크래핑 ──
print(f'n[STEP 1/4] "{topic}"에 대한 기사 찾기 및 스크래핑 중...')
documents = collect_data(
topic, cust_id, unlocker_zone, unlocker_pwd, serp_zone, serp_pwd
)
print(f"n 수집된 총 문서 수: {len(documents)}")
if not documents:
print(" 오류: 수집된 문서가 없습니다. 다른 주제를 시도해 보세요.")
return
# ── 2단계: 문서 처리 및 청크 분할 ──
print("n[STEP 2/4] 문서 처리 및 청크 분할 중...")
chunks = process_documents(documents)
print(f" {len(documents)}개의 문서에서 {len(chunks)}개의 청크를 생성했습니다.")
if not chunks:
print(" 오류: 생성된 청크가 없습니다. 문서가 너무 짧을 수 있습니다.")
return
# ── 3단계: Weaviate에 저장 ──
print("n[STEP 3/4] Weaviate에 저장 중 (임베딩 + 인덱싱)...")
print(" Weaviate Cloud에 연결 중...", end=" ", flush=True)
client = connect_weaviate()
print("OK")
print(" 컬렉션 설정 중...")
setup_collection(client)
print(f" {len(chunks)} 개의 청크 가져오기 (Cohere를 통한 자동 벡터화)...")
failed = store_chunks(client, chunks)
print(f" 가져온 개수: {len(chunks) - failed} 성공, {failed} 실패")
# 개수 확인
collection = client.collections.use(COLLECTION_NAME)
count = collection.aggregate.over_all(total_count=True).total_count
print(f" Weaviate의 총 객체 수: {count}")
# ── 4단계: 데모 쿼리 + 대화형 모드 ──
print(f'n[STEP 4/4] "{topic}"에 대한 RAG 쿼리...')
print("=" * 65)
demo_queries = [
f"{topic}의 최신 동향과 트렌드는 무엇인가요?",
f"{topic}의 가장 큰 과제와 위험 요소는 무엇인가요?",
f"{topic}의 향후 전망은 어떻게 되나요?",
]
for question in demo_queries:
rag_query(client, question)
print()
# ── 요약 ──
print("=" * 65)
print(" 파이프라인 완료!")
print(f' 주제: "{topic}"')
print(f" - Bright Data를 통해 {len(documents)}개의 소스를 수집함")
print(f" - Weaviate에 {count}개의 청크를 저장했습니다")
print(f" - {len(demo_queries)}개의 데모 RAG 쿼리를 실행했습니다")
print("=" * 65)
# ── 대화형 모드 ──
print(f'n "{topic}"에 대한 지식 기반이 준비되었습니다!')
print(" 무엇이든 물어보세요. 종료하려면 'quit'을 입력하세요.n")
while True:
try:
user_question = input(" 질문: ").strip()
except (EOFError, KeyboardInterrupt):
print("n 안녕히 가세요!")
break
if not user_question:
continue
if user_question.lower() in ("quit", "exit", "q"):
print(" 안녕히 가세요!")
break
rag_query(client, user_question)
print()
client.close()
if __name__ == "__main__":
main()실행하기:
python3 pipeline.pyAI 경쟁 테스트 실행 결과 RAG 답변
이 파이프라인은 “OpenAI vs Google vs Anthropic AI 경쟁” 을 주제로 실행되었습니다. 다음은 테스트 실행 시 생성된 RAG 답변 예시입니다. 실제 결과는 실행 시점에 게시된 기사에 따라 달라집니다.
질문 1: “OpenAI, Google, Anthropic 간의 AI 경쟁에서 최근 동향과 트렌드는 무엇인가요?”
OpenAI, Google, Anthropic 간의 AI 경쟁은 각 기업이 고유한 강점을 활용하며 계속해서 빠르게 진화하고 있습니다. OpenAI는 선점 효과 덕분에 매출과 소비자 채택 측면에서 선두를 유지하고 있습니다. Anthropic은 Claude Code와 같은 전문 도구와 AI 서비스를 구매하는 기업 간 직접 비교에서 70%의 승률을 기록하며 기업 채택 부문에서 격차를 좁혀가고 있습니다. Google은 타의 추종을 불허하는 컴퓨팅 리소스와 자사 생태계 전반에 걸친 원활한 통합을 제공합니다.
출처: shawnkanungo[.]com, hashmeta[.]ai, ramp[.]com
질문 2: “OpenAI, Google, Anthropic 간의 AI 경쟁에서 가장 큰 과제와 위험 요소는 무엇인가?”
오픈AI는 특히 컴퓨팅 자원을 파트너십에 의존하고 있는 만큼, 독립성을 유지하면서 혁신 속도를 지속해야 하는 과제에 직면해 있습니다. 구글은 관료주의적 관성으로 어려움을 겪고 있으며, 대화형 AI가 광고 클릭을 감소시킴에 따라 핵심 검색 광고 사업을 잠식할 위험에 처해 있습니다. 안전성을 최우선으로 내세우는 앤트로픽은 해석 가능성에 대한 집중을 역량 중심 시장에서 시장 점유율로 전환해야 합니다.
출처: hashmeta[.]ai, shawnkanungo[.]com
질문 3: “OpenAI, Google, Anthropic AI 간의 경쟁 구도에 대한 미래 전망은 어떠한가?”
오픈AI는 매출과 소비자 채택 측면에서 선두를 달리고 있으며, GPT-5와 추론 비용 절감에 대한 투자가 로드맵에 포함되어 있다. 앤트로픽의 미래 성공은 설명 가능성에 대한 규제 요건이 등장하느냐에 달려 있는데, 안전성과 해석 가능성에 대한 초기 투자가 상당한 우위를 제공할 수 있다. 구글은 특히 제미니(Gemini) 와 같은 도구를 특정 사용 사례에 맞게 조정하고 AI를 일상적인 업무 흐름에 통합하는 데 있어 여전히 강력한 경쟁자로 남아 있다.
출처: hashmeta[.]ai, shawnkanungo[.]com
각 답변은 동일한 파이프라인 실행 중에 수집된 기사를 기반으로 합니다. 각 인용은 1단계에서 수집된 출처를 가리키며, URL을 열어 어떤 주장이라도 확인할 수 있습니다. 수집된 기사에 다루지 않은 내용을 질문하면, 모델은 이를 알리거나 덜 상세한 답변을 제공합니다.
데모 쿼리가 끝나면 파이프라인은 대화형 모드로 전환되어 사용자가 직접 질문을 할 수 있습니다:
실전 환경 적용
이 기능을 실제 운영 환경에 적용하려면 멀티테넌시, 규정 준수 및 비용 관리가 필요합니다. (전체적인 맥락을 파악하려면 RAG가 실제 AI 에이전트 기술 스택에 어떻게 적용되는지 확인해 보세요.)
데이터 격리를 위한 멀티 테넌시
여러 고객을 위해 RAG를 구축하는 경우, Weaviate의 멀티 테넌시 기능을 사용하면 각 테넌트에게 격리된 벡터 인덱스가 포함된 전용 샤드를 할당할 수 있습니다:
from weaviate.classes.config import Configure
from weaviate.classes.tenants import Tenant
# 컬렉션에서 멀티 테넌시 활성화
collection = client.collections.create(
name="WebContent",
multi_tenancy_config=Configure.multi_tenancy(enabled=True),
# ... 벡터라이저 + 생성형 구성
)
# 각 고객은 고유한 격리된 테넌트를 할당받습니다
collection.tenants.create([
Tenant(name="customer_a"),
Tenant(name="customer_b"),
Tenant(name="customer_c"),
])
# 테넌트별로 데이터를 수집하고 저장
tenant_collection = collection.with_tenant("customer_a")
with tenant_collection.batch.dynamic() as batch:
for chunk in customer_a_chunks:
batch.add_object(properties=chunk)단일 노드는 50,000개 이상의 활성 테넌트를 지원하며, 20노드 클러스터는 100만 개를 처리합니다.
비용 최적화
데이터가 증가함에 따라 비용을 절감하는 4가지 기법:
- Weaviate 회전 양자화 – 98~99%의 리콜률로 벡터 압축률을 4배 향상시킵니다.
- 콘텐츠 해싱 –
content_hash필드를 사용하면 변경되지 않은 청크를 다시 임베딩하지 않고 증분 업데이트가 가능합니다(위의 3단계 참조). - 메타데이터 필드에
skip_vectorization=True설정 – 중요한 데이터만 임베딩합니다. - Bright Data 데이터셋 마켓플레이스 – 일반적인 도메인의 경우 스크래핑 대신 미리 수집된 데이터셋을 활용하세요.
단일 사용자 프로토타입 단계를 넘어설 때 이러한 방법들이 중요해집니다.
흔히 발생하는 오류와 해결 방법
문제가 발생하면 먼저 이 표를 확인하세요:
| 문제 | 원인 | 해결 방법 |
|---|---|---|
Weaviate gRPC DEADLINE_EXCEEDED |
스크래핑 중에 샌드박스 클러스터가 유휴 상태가 됨 | pipeline.py를 다시 실행하십시오. 스크립트가 자동으로 재연결됩니다. 문제가 지속되면 Weaviate 콘솔에서 클러스터 상태를 확인하십시오 |
Cohere API 속도 제한 (429) |
체험판 요금제에 속도 제한이 적용됩니다 | 1분 정도 기다렸다가 다시 시도하거나, Cohere 대시보드에서 사용량을 확인하십시오. 자동 실행은 20회 미만의 호출을 사용하며, 대화형 질문 하나당 2회가 추가됩니다 |
Web Unlocker 영역을 찾을 수 없음 |
Bright Data 계정에 Web Unlocker 영역이 없습니다 | Bright Data → 프록시 및 스크래핑 → 내 존으로 이동하여 Web Unlocker 존을 생성하십시오 |
SERP API 영역이 없습니다 |
Bright Data 계정에 SERP 영역이 없습니다 | Bright Data → 프록시 및 스크래핑 → 내 존으로 이동하여 SERP API 존을 생성하십시오 |
모든 URL에서HTTP 403 오류가 발생했습니다 |
Web Unlocker 재시도 횟수 초과 | 다른 주제를 시도해 보세요. 일부 틈새 사이트는 엄격한 봇 차단 기능을 사용합니다. 고급 옵션을 위해 CAPTCHA 우회 방법을 확인하세요 |
Weaviate 클러스터가 준비되지 않았습니다 |
샌드박스 만료됨(14일 제한) | Weaviate 콘솔에서 새 샌드박스를 생성하고 .env 파일을 업데이트하세요 |
| Cohere 모델을 사용할 수 없습니다 | command-a-03-2025 또는 embed-v4.0이 지원 종료되었습니다 |
docs.cohere.com/docs/models에서 사용 가능한 모델을 확인하고 setup_collection()의 model= 매개변수를 업데이트하십시오 |
ModuleNotFoundError: 'weaviate'라는 모듈이 없습니다 |
의존성이 설치되지 않았습니다 | 프로젝트 디렉터리에서 pip3 install -r requirements.txt를 실행하십시오 |
오류가 목록에 없는 경우 전체 출력을 확인하십시오. 파이프라인은 모든 단계를 세부 정보와 함께 기록합니다.
사용 사례
동일한 아키텍처는 어떤 주제에도 적용됩니다. 몇 가지 예시:
- 경쟁 정보 – 주제: “경쟁사 X의 가격 전략”. 파이프라인은 경쟁사 웹사이트, 가격 페이지 및 애널리스트 보고서를 수집합니다. 그런 다음 “경쟁사 X의 기업용 가격 정책은 우리와 어떻게 비교되나요?”라고 질문합니다.
- 시장 조사 – 주제: “동남아시아 핀테크 동향”. 지역 뉴스와 업계 간행물을 수집하여 “동남아시아에서 떠오르는 주요 핀테크 트렌드는 무엇인가?”와 같은 질문을 할 수 있게 해줍니다.
- 전자상거래 – 주제: “지속 가능한 패션 시장”. 시장 보고서와 소비자 조사 자료를 수집합니다. “어떤 지속 가능한 패션 브랜드가 시장 점유율을 높이고 있나요?”
- 기술 연구 – 주제: “쿠버네티스 보안 모범 사례”. 기술 블로그와 보안 권고 사항을 수집하므로, 특정 CVE나 잘못된 구성에 대해 질문할 수 있습니다.
다음 개발 과제
이것은 알려진 제약 사항이 있는 작동하는 프로토타입입니다:
- 실행할 때마다 전체 컬렉션을 교체합니다(증분 업데이트 없음) – diff 기능을 추가하려면
content_hash를사용하세요 - 텍스트만 처리하며, 수집된 페이지의 표, 이미지, PDF는 제외됩니다
- Google 검색을 통해 콘텐츠를 찾습니다 – 특정 URL이 필요한 경우, 해당 URL을
scrape_url()에 직접 전달하십시오 - 단일 사용자 CLI로 실행됩니다
이 단계에서 다음과 같은 작업을 수행할 수 있습니다:
- 스케줄링 – 크론 작업을 통해 파이프라인을 실행하여 지식베이스를 최신 상태로 유지
- 다중 테넌시 – 각 고객에게 독립된 샤드를 할당합니다(위의 “프로덕션 환경으로 이동” 섹션 참조)
- 다양한 데이터 소스 – 구조화된 Amazon 또는 LinkedIn 데이터의 경우 Bright Data Web Scraper API를, 전체 사이트의 Markdown 데이터의 경우 Crawl API를 사용하세요
- 프론트엔드 –
rag_query()를Flask 또는 FastAPI 엔드포인트로 래핑하고 채팅 UI를 연결 - 에이전트형 RAG – 언제, 무엇을 스크래핑할지 스스로 결정하는 에이전트형 RAG 시스템을 구축
- LangChain – 내장된 체인 오케스트레이션 및 메모리 기능을 위해 Bright Data와 함께 파이프라인을 LangChain 으로 이식
자주 묻는 질문
이 파이프라인은 어떤 주제에 사용할 수 있나요?
공개 웹에 기사가 있는 모든 주제입니다. 이 파이프라인은 Bright Data SERP API를 사용하여 Google에서 주제를 검색한 다음, 상위 결과를 스크래핑합니다. 색인된 페이지가 적은 틈새 주제의 경우 반환되는 기사 수가 적지만, 파이프라인은 여전히 작동합니다. 파이프라인은 찾은 내용을 그대로 사용합니다.
운영 비용은 얼마인가요?
3가지 서비스 모두 무료로 시작할 수 있는 방법을 제공합니다. Cohere의 Trial 플랜은 신용카드 없이 무료로 이용할 수 있습니다. Weaviate Cloud는 무료 샌드박스 클러스터를 제공하며, Bright Data는 SERP API와 Web Unlocker에 대한 무료 체험판을 제공합니다.
다른 임베딩 모델이나 LLM을 사용할 수 있나요?
네. 임베딩과 생성 모두에 대해 setup_collection() 에서 모델 매개변수를 변경하면 됩니다. Weaviate는 기본적으로 Cohere, OpenAI, Google, Hugging Face 벡터화기를 지원합니다. 전환하려면 text2vec_cohere를 text2vec_openai로 교체하고, connect_weaviate()에서 API 키 헤더를 업데이트한 후 파이프라인을 다시 실행하세요.
지식 기반을 최신 상태로 유지하려면 어떻게 해야 하나요?
동일한 주제를 사용하여 pipeline.py를 다시 실행하세요. 파이프라인은 기존 컬렉션을 삭제하고 새로 수집된 데이터로 새로운 컬렉션을 생성합니다. 프로덕션 환경에서는 content_hash 검사를 추가하여 변경되지 않은 청크를 재임베딩하지 않도록 할 수 있습니다. 크론 작업을 통해 파이프라인을 예약하여 원하는 간격으로 데이터를 자동으로 갱신하세요.
스크랩할 URL이 이미 있다면 어떻게 하나요?
SERP 검색 단계를 건너뛰세요. collect_data()에서 find_articles_for_topic() 호출을 직접 준비한 URL 목록으로 대체하고, 각 URL을 scrape_url()에 전달하세요. 파이프라인의 나머지 단계(청크 생성, 임베딩, 쿼리)는 동일한 방식으로 작동합니다.
6개 이상의 기사를 스크래핑하려면 어떻게 해야 하나요?
find_articles_for_topic() 끝부분의 [:6] 슬라이스를 더 큰 숫자(예: [:12])로 변경하세요. 또한 search_queries 목록에 검색 쿼리를 더 추가하여 더 광범위한 결과를 얻을 수도 있습니다. 기사가 많아지면 스크래핑 시간이 길어지고 청크 수도 늘어나지만, 파이프라인의 나머지 단계에서는 이를 자동으로 처리합니다.
