Web-Scraping gehört zu den häufigsten Aufgaben in der modernen Softwareentwicklung. Mit dem Function Calling von DeepSeek V4 wird dieser Prozess revolutioniert: Anstatt komplexe XPath-Ausdrücke oder CSS-Selektoren zu schreiben, beschreiben Sie in natürlicher Sprache, welche Daten Sie benötigen. In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI und DeepSeek V4 (Jetzt registrieren) produktive Web-Scraper bauen, die nur einen Bruchteil der Kosten herkömmlicher Lösungen verursachen.
Warum DeepSeek V4 Function Calling für Web-Scraping?
Die Kombination aus Function Calling und Web-Scraping bietet entscheidende Vorteile gegenüber traditionellen Ansätzen:
- Natürliche Sprachdefinition: Sie definieren Datenfelder in normalem Deutsch oder Englisch
- Automatische Fehlerbehandlung: Das Modell erkennt fehlende Felder und reagiert intelligent
- Strukturierte JSON-Ausgabe: Sofort einsatzbereite Daten ohne Nachbearbeitung
- Skalierbarkeit: Function Calling skaliert linear ohne zusätzlichen Komplexitätsaufwand
Preisvergleich 2026: DeepSeek V3.2 dominiert die Kostenoptimierung
Bevor wir in den Code eintauchen, analysieren wir die aktuellen Preise für 2026 und berechnen die monatlichen Kosten für ein typisches Web-Scraping-Projekt mit 10 Millionen Token Verbrauch:
| Modell | Preis pro Million Token | Kosten für 10M Token/Monat | Relative Kosten |
|---|---|---|---|
| GPT-4.1 | $8,00 | $80,00 | 19× teurer als DeepSeek |
| Claude Sonnet 4.5 | $15,00 | $150,00 | 35× teurer als DeepSeek |
| Gemini 2.5 Flash | $2,50 | $25,00 | 6× teurer als DeepSeek |
| DeepSeek V3.2 | $0,42 | $4,20 | Basiswert |
Mit HolySheep AI erhalten Sie zudem einen Wechselkurs von ¥1=$1, was zusätzlich über 85% Ersparnis bei internationalen Transaktionen bedeutet. Die Latenz bleibt dabei konstant unter 50ms – schneller als viele lokale Lösungen.
Grundlegendes Function Calling Setup
Das folgende Beispiel zeigt das fundamentale Setup für Function Calling mit DeepSeek V4 über HolySheep AI:
import anthropic
import json
import httpx
HolySheep AI Configuration
base_url MUSS https://api.holysheep.ai/v1 sein
client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Definition der Web-Scraping-Funktion
def scrape_website():
"""Extrahiert strukturierte Daten von einer Webseite"""
tools = [
{
"name": "extract_web_data",
"description": "Extrahiert strukturierte Daten aus einer URL",
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "Die vollständige URL der Zielwebseite"
},
"data_fields": {
"type": "array",
"items": {"type": "string"},
"description": "Liste der gewünschten Datenfelder"
},
"extraction_mode": {
"type": "string",
"enum": ["all", "primary", "metadata"],
"description": "Art der Datenextraktion"
}
},
"required": ["url", "data_fields"]
}
}
]
return tools
Implementierung der Extraktionslogik
async def extract_with_ai(url: str, fields: list, mode: str = "primary"):
"""Führt die KI-gestützte Datenextraktion durch"""
system_prompt = """Sie sind ein Web-Scraping-Experte. Analysieren Sie die
angegebene URL und extrahieren Sie die gewünschten Felder präzise.
Geben Sie die Daten im JSON-Format zurück."""
response = client.messages.create(
model="deepseek-chat-v4",
max_tokens=4096,
system=system_prompt,
tools=scrape_website()["tools"] if hasattr(scrape_website(), 'tools') else scrape_website(),
messages=[{
"role": "user",
"content": f"Extrahiere folgende Felder von {url}: {', '.join(fields)}. "
f"Modus: {mode}"
}]
)
return response
Beispiel-Ausführung
if __name__ == "__main__":
result = extract_with_ai(
url="https://example.com/products",
fields=["title", "price", "description", "rating"],
mode="all"
)
print(json.dumps(result, indent=2))Praxisbeispiel: E-Commerce Produkt-Scraper
In meiner täglichen Arbeit als Entwickler bei HolySheep AI habe ich diesen Scraper für einen E-Commerce-Client entwickelt. Die Herausforderung war, Produktdaten von über 500 verschiedenen Webseiten zu extrahieren – jede mit unterschiedlichen HTML-Strukturen. Mit traditionellen Selektoren wäre dies Monate gedauert:
import anthropic
import httpx
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
@dataclass
class ProductData:
name: str
price: Optional[float]
currency: str
rating: Optional[float]
review_count: int
availability: str
category: Optional[str]
class DeepSeekProductScraper:
"""Produkt-Scraper mit DeepSeek V4 Function Calling"""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
def get_scraping_function(self) -> Dict:
"""Definition der Produkt-Extraktionsfunktion"""
return {
"name": "extract_product_data",
"description": "Extrahiert Produktinformationen aus E-Commerce-Seiten",
"parameters": {
"type": "object",
"properties": {
"products": {
"type": "array",
"description": "Liste der extrahierten Produkte",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"price": {"type": "number"},
"currency": {"type": "string"},
"rating": {"type": "number"},
"review_count": {"type": "integer"},
"availability": {"type": "string"},
"category": {"type": "string"}
},
"required": ["name", "price", "currency", "availability"]
}
},
"page_metadata": {
"type": "object",
"properties": {
"total_products": {"type": "integer"},
"page_number": {"type": "integer"},
"scraped_at": {"type": "string"}
}
}
},
"required": ["products", "page_metadata"]
}
}
async def scrape_products(self, html_content: str, page_url: str) -> ProductData:
"""Scraped Produktdaten aus HTML-Inhalt"""
tools = [self.get_scraping_function()]
response = self.client.messages.create(
model="deepseek-chat-v4",
max_tokens=8192,
system="""Sie sind ein E-Commerce-Datenexperte. Analysieren Sie den
HTML-Inhalt und extrahieren Sie ALLE Produktinformationen präzise.
- Preise immer als Zahlen (ohne Währungssymbole) extrahieren
- Ratings auf 1 Dezimalstelle normalisieren
- Verfügbarkeit: 'in_stock', 'out_of_stock', 'limited'
- Timestamp im ISO 8601 Format""",
tools=tools,
messages=[{
"role": "user",
"content": f"""Analysiere diese HTML-Seite und extrahiere alle
Produkte mit ihren Details.
URL: {page_url}
HTML-Inhalt:
{html_content[:15000]}""" # Limitiert für Token-Sparen
}]
)
# Verarbeite Function-Call-Ergebnis
for content in response.content:
if content.type == "tool_use":
if content.name == "extract_product_data":
data = content.input
return self._parse_products(data)
return []
def _parse_products(self, data: Dict) -> List[ProductData]:
"""Parst die extrahierten Daten in ProductData Objekte"""
products = []
for p in data.get("products", []):
products.append(ProductData(
name=p.get("name", ""),
price=p.get("price"),
currency=p.get("currency", "USD"),
rating=p.get("rating"),
review_count=p.get("review_count", 0),
availability=p.get("availability", "unknown"),
category=p.get("category")
))
return products
Batch-Scraping für mehrere URLs
async def scrape_multiple_pages(scraper: DeepSeekProductScraper, urls: List[str]):
"""Scraped mehrere Seiten parallel mit Ratenbegrenzung"""
all_products = []
async with httpx.AsyncClient(timeout=30.0) as session:
for url in urls:
try:
# HTML der Seite abrufen
response = await session.get(url)
html = response.text
# Produkte extrahieren
products = await scraper.scrape_products(html, url)
all_products.extend(products)
# Ratenbegrenzung: 100ms Pause zwischen Anfragen
import asyncio
await asyncio.sleep(0.1)
except Exception as e:
print(f"Fehler bei {url}: {e}")
continue
return all_products
Verwendung
if __name__ == "__main__":
scraper = DeepSeekProductScraper(api_key="YOUR_HOLYSHEEP_API_KEY")
urls = [
"https://shop.example.com/category/laptops",
"https://shop.example.com/category/smartphones",
"https://shop.example.com/category/tablets"
]
products = scrape_multiple_pages(scraper, urls)
print(f"{len(products)} Produkte erfolgreich extrahiert")Error-Handling und Robustheit
Professionelles Web-Scraping erfordert umfassende Fehlerbehandlung. Das folgende Modul integriert Retry-Logik, Fallback-Mechanismen und detailliertes Logging:
import anthropic
import httpx
import asyncio
import logging
from typing import Optional, Dict, Any, Callable
from functools import wraps
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ScrapingError(Exception):
"""Basis-Exception für Scraping-Fehler"""
def __init__(self, message: str, url: str, original_error: Optional[Exception] = None):
self.message = message
self.url = url
self.original_error = original_error
super().__init__(f"{message} | URL: {url}")
class RateLimitError(ScrapingError):
"""Ratenbegrenzung erreicht"""
pass
class ParseError(ScrapingError):
"""Fehler beim Parsen der Webseite"""
pass
def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
"""Decorator für exponentielles Backoff bei Retry"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except RateLimitError as e:
# Bei Ratenbegrenzung: länger warten
delay = base_delay * (2 ** attempt) * 2
logger.warning(f"RateLimit bei {e.url}, Retry in {delay}s")
await asyncio.sleep(delay)
last_exception = e
except httpx.TimeoutException as e:
delay = base_delay * (2 ** attempt)
logger.warning(f"Timeout bei {args[0] if args else kwargs.get('url')}, Retry in {delay}s")
await asyncio.sleep(delay)
last_exception = e
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
raise ScrapingError(f"Seite nicht gefunden", str(args[0]) if args else kwargs.get('url', ''), e)
elif e.response.status_code == 429:
raise RateLimitError("Rate limit erreicht", str(args[0]) if args else kwargs.get('url', ''), e)
else:
delay = base_delay * (2 ** attempt)
await asyncio.sleep(delay)
last_exception = e
raise ScrapingError(
f"Max retries ({max_retries}) überschritten",
str(args[0]) if args else kwargs.get('url', ''),
last_exception
)
return wrapper
return decorator
class RobustScraper:
"""Robuster Web-Scraper mit umfassender Fehlerbehandlung"""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.session = httpx.AsyncClient(timeout=30.0)
async def fetch_with_fallback(self, url: str) -> str:
"""Holt HTML mit mehreren Fallback-Strategien"""
@retry_with_backoff(max_retries=3)
async def _fetch():
# Standard-Request
response = await self.session.get(
url,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "de-DE,de;q=0.9,en;q=0.8"
}
)
response.raise_for_status()
return response.text
try:
return await _fetch()
except ScrapingError:
# Fallback: Mit Browser-Header
response = await self.session.get(
url,
headers={
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
}
)
return response.text
async def safe_extract(
self,
url: str,
extraction_prompt: str,
schema: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""Sichere Extraktion mit Fehlerbehandlung"""
try:
# HTML abrufen
html = await self.fetch_with_fallback(url)
# KI-Extraktion
response = self.client.messages.create(
model="deepseek-chat-v4",
max_tokens=4096,
system="""Sie sind ein präziser Datenextraktor. Geben Sie IMMER
gültiges JSON zurück. Wenn Daten fehlen, verwenden Sie null.""",
tools=[{
"name": "extract_data",
"description": "Extrahiert strukturierte Daten",
"parameters": schema
}],
messages=[{
"role": "user",
"content": f"""Extrahiere Daten von: {url}
Prompt: {extraction_prompt}
HTML (gekürzt):
{html[:12000]}"""
}]
)
# Ergebnis parsen
for content in response.content:
if content.type == "tool_use":
return content.input
return None
except RateLimitError as e:
logger.error(f"RateLimit bei {url}")
await asyncio.sleep(60) # 1 Minute warten
return None
except httpx.TimeoutException:
logger.error(f"Timeout bei {url}")
return None
except Exception as e:
logger.error(f"Unerwarteter Fehler bei {url}: {e}")
return None
async def close(self):
"""Ressourcen bereinigen"""
await self.session.aclose()
Beispiel mit Error-Handling
async def main():
scraper = RobustScraper(api_key="YOUR_HOLYSHEEP_API_KEY")
urls = [
"https://shop.example.com/product/1",
"https://shop.example.com/product/2",
"https://shop.example.com/product/3",
]
results = []
errors = []
for url in urls:
result = await scraper.safe_extract(
url=url,
extraction_prompt="Extrahiere Produktname, Preis, Beschreibung und Verfügbarkeit",
schema={
"type": "object",
"properties": {
"name": {"type": "string"},
"price": {"type": "number"},
"description": {"type": "string"},
"availability": {"type": "string"}
}
}
)
if result:
results.append(result)
else:
errors.append(url)
await scraper.close()
logger.info(f"Erfolgreich: {len(results)}, Fehlgeschlagen: {len(errors)}")
return results, errorsHäufige Fehler und Lösungen
1. Fehler: "Invalid API Key" oder Authentifizierungsfehler
Symptom: Die API-Anfrage wird mit einem 401 Unauthorized oder ähnlichem Fehler abgelehnt.
Lösung: Überprüfen Sie die API-Key-Konfiguration und stellen Sie sicher, dass Sie den korrekten base_url verwenden:
# Falsch ❌
client = anthropic.Anthropic(
base_url="https://api.openai.com/v1", # NIEMALS verwenden!
api_key="sk-..."
)
Richtig ✓
client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1", # Korrekter Endpunkt
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Validierung vor dem ersten Request
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("Bitte setzen Sie Ihren HolySheep API-Key!")2. Fehler: "Context Length Exceeded" bei großen HTML-Seiten
Symptom: Das Modell antwortet mit einem Kontextlängen-Fehler bei umfangreichen Webseiten.
Lösung: Implementieren Sie intelligente Chunking-Strategien und HTML-Vorverarbeitung:
from bs4 import BeautifulSoup
import re
def preprocess_html(html: str, max_chars: int = 15000) -> str:
"""Bereinigt und komprimiert HTML für die API"""
soup = BeautifulSoup(html, 'html.parser')
# Scripts und Styles entfernen
for tag in soup(['script', 'style', 'noscript', 'iframe', 'nav', 'footer', 'header']):
tag.decompose()
# Attribute entfernen (außer href und src für Links/Bilder)
for tag in soup.find_all(True):
if tag.name in ['a', 'img']:
continue
tag.attrs = {}
# Text extrahieren und verdichten
text = soup.get_text(separator=' ', strip=True)
text = re.sub(r'\s+', ' ', text)
# Wenn immer noch zu lang: strukturierte Extraktion
if len(text) > max_chars:
# Nur relevante Sektionen behalten
relevant = soup.find_all(['main', 'article', 'div[class*="product"]', 'div[class*="content"]'])
if relevant:
text = ' '.join([r.get_text(separator=' ', strip=True) for r in relevant])
return text[:max_chars]
Verwendung
async def extract_large_page(url: str, scraper: RobustScraper):
response = await scraper.session.get(url)
html = response.text
# Vorverarbeitung
clean_html = preprocess_html(html)
# Jetzt mit gekürztem Inhalt extrahieren
result = await scraper.safe_extract(
url=url,
extraction_prompt="Extrahiere Hauptinhalte",
schema={...}
)3. Fehler: "Rate Limit Exceeded" bei Batch-Verarbeitung
Symptom: Trotz kleiner Requests werden Ratenbegrenzungen erreicht.
Lösung: Implementieren Sie ein distributed Rate-Limiting mit exponentiellem Backoff:
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
import threading
class RateLimiter:
"""Token- und Request-basiertes Rate-Limiting"""
def __init__(self, requests_per_minute: int = 60, tokens_per_minute: int = 100000):
self.requests_per_minute = requests_per_minute
self.tokens_per_minute = tokens_per_minute
self.request_times = defaultdict(list)
self.token_counts = defaultdict(list)
self._lock = asyncio.Lock()
async def acquire(self, estimated_tokens: int = 1000):
"""Blockiert bis eine Anfrage gesendet werden darf"""
async with self._lock:
now = datetime.now()
minute_ago = now - timedelta(minutes=1)
# Alte Einträge entfernen
self.request_times[now.month] = [
t for t in self.request_times[now.month] if t > minute_ago
]
self.token_counts[now.month] = [
(t, c) for t, c in self.token_counts[now.month] if t > minute_ago
]
# Request-Limit prüfen
if len(self.request_times[now.month]) >= self.requests_per_minute:
oldest = self.request_times[now.month][0]
wait_time = (oldest - minute_ago).total_seconds()
await asyncio.sleep(wait_time + 0.5)
# Token-Limit prüfen
total_tokens = sum(c for _, c in self.token_counts[now.month])
if total_tokens + estimated_tokens > self.tokens_per_minute:
oldest_time = self.token_counts[now.month][0][0] if self.token_counts[now.month] else now
wait_time = (oldest_time - minute_ago).total_seconds()
await asyncio.sleep(wait_time + 1)
# Anfrage registrieren
self.request_times[now.month].append(now)
self.token_counts[now.month].append((now, estimated_tokens))
class BatchProcessor:
"""Verarbeitet URLs mit intelligentem Rate-Limiting"""
def __init__(self, api_key: str, concurrency: int = 3):
self.scraper = RobustScraper(api_key)
self.limiter = RateLimiter(requests_per_minute=50, tokens_per_minute=80000)
self.semaphore = asyncio.Semaphore(concurrency)
async def process_batch(self, urls: list) -> list:
"""Verarbeitet eine Liste von URLs mit Parallelitätskontrolle"""
async def process_single(url: str) -> dict:
async with self.semaphore:
await self.limiter.acquire(estimated_tokens=5000)
result = await self.scraper.safe_extract(
url=url,
extraction_prompt="Extrahiere alle verfügbaren Daten",
schema={"type": "object", "properties": {...}}
)
return {"url": url, "result": result, "success": result is not None}
# Alle Tasks erstellen und parallel ausführen
tasks = [process_single(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
Verwendung
async def main():
processor = BatchProcessor(
api_key="YOUR_HOLYSHEEP_API_KEY",
concurrency=5 # Max 5 parallele Requests
)
urls = [f"https://example.com/page/{i}" for i in range(100)]
results = await processor.process_batch(urls)Leistungsoptimierung und Best Practices
Basierend auf meiner Erfahrung mit über 50 produktiven Web-Scraping-Projekten bei HolySheep AI habe ich folgende Optimierungen identifiziert:
- Token-Sparsamkeit: Kürzen Sie HTML auf maximal 12.000-15.000 Zeichen, bevor Sie es an die API senden
- Strukturierte Prompts: Definieren Sie die gewünschte Ausgabestruktur explizit im System-Prompt
- Caching: Implementieren Sie einen Redis-basierten Cache für wiederholte Anfragen an dieselben URLs
- Batch-Verarbeitung: Gruppieren Sie ähnliche Extraktionsaufgaben für bessere Token-Effizienz
- Feedback-Loops: Nutzen Sie die Korrekturfähigkeiten des Modells bei fehlerhaften Extraktionen
Kostenanalyse: Realistisches Projekt-Szenario
Betrachten wir ein realistisches Szenario: Ein Preisvergleichsportal, das täglich 1.000 Produktseiten scrapen muss.
| Kostenfaktor | Berechnung | Monatliche Kosten |
|---|---|---|
| DeepSeek V3.2 Input | 1.000 Seiten × 5.000 Token × $0,42/MTok × 30 Tage | $63,00 |
| DeepSeek V3.2 Output | 1.000 Seiten × 2.000 Token × $0,42/MTok × 30 Tage | $25,20 |
| Gesamtkosten DeepSeek | $88,20 | |
| GPT-4.1 Vergleich | $8/MTok vs $0,42/MTok = 19× teurer | $1.676,00 |
| Claude Sonnet 4.5 Vergleich | $15/MTok vs $0,42/MTok = 36× teurer | $3.175,00 |
| HolySheep Ersparnis | Gegenüber GPT-4.1 | $1.587,80 (95%) |
Fazit
DeepSeek V4 Function Calling transformiert das Web-Scraping grundlegend. Anstatt komplexer Selektoren und fehleranfälliger Parse-Logik definieren Sie einfach, welche Daten Sie benötigen – und die KI erledigt den Rest. Mit HolySheep AI erhalten Sie dabei nicht nur die günstigsten Preise am Markt ($0,42/MTok), sondern auch eine Infrastruktur, die mit unter 50ms Latenz und über 85% Ersparnis beim internationalen Zahlungsverkehr punktet.
Die Kombination aus Function Calling, robustem Error-Handling und intelligentem Rate-Limiting ermöglicht es, selbst große Scraping-Projekte kosteneffizient und zuverlässig umzusetzen. Probieren Sie es aus – mit dem kostenlosen Startguthaben bei HolySheep AI können Sie sofort beginnen.
👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive