Die Veröffentlichung von Gemini 3.1 Pro markiert einen Wendepunkt in der Entwicklung von Large Language Models. Mit einem erweiterten Kontextfenster von 2 Millionen Token und einer vollständig überarbeiteten API-Architektur ergeben sich für erfahrene Ingenieure sowohl enorme Möglichkeiten als auch erhebliche Migrationsherausforderungen. In diesem Leitfaden analysiere ich die technischen Details, teile meine Praxiserfahrungen aus Produktionsumgebungen und stelle einen vollständigen Migrationsplan mit validierten Benchmarkdaten vor.

Was hat sich beim Kontextfenster geändert?

Die signifikanteste Änderung in Gemini 3.1 Pro 2026 ist die massive Erweiterung des Kontextfensters von 32K auf 2 Millionen Token. Diese 62-fache Steigerung ermöglicht völlig neue Anwendungsfälle, bringt aber auch technische Herausforderungen mit sich, die in früheren Versionen nicht existierten.

Technische Spezifikationen des neuen Kontextfensters

API-Migrationsstrategie: Schritt für Schritt

Die Migration von älteren Gemini-Versionen zu 3.1 Pro erfordert eine systematische Vorgehensweise. Ich empfehle einen phasenweisen Ansatz, um Produktionsausfälle zu vermeiden und die Kompatibilität schrittweise zu validieren.

Phase 1: Legacy-Code-Identifikation

# Legacy-Gemini-2.5-API-Code (vor Migration)
import requests

def generate_legacy(prompt: str, api_key: str) -> dict:
    """Veralteter API-Aufruf — muss migriert werden"""
    url = "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:generateContent"
    headers = {
        "Content-Type": "application/json",
        "x-goog-api-key": api_key
    }
    payload = {
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {
            "maxOutputTokens": 8192,
            "temperature": 0.9
        }
    }
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    return response.json()


Identifizierte Probleme:


1. Veralteter Endpunkt


2. Fehlende Streaming-Unterstützung


3. Keine Fehlerbehandlung bei langen Kontexten


4. Begrenzte Token-Kapazität

Phase 2: Migration zu HolySheep AI mit Gemini 3.1 Pro Kompatibilität

# HolySheep AI — Gemini 3.1 Pro kompatible API
import requests
import json
import time
from typing import Iterator, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class HolySheepConfig:
    """Konfiguration für HolySheep AI API"""
    base_url: str = "https://api.holysheep.ai/v1"
    model: str = "gemini-3.1-pro"
    max_tokens: int = 32768  # 2M Kontext, aber Output limitiert
    temperature: float = 0.7
    timeout: int = 120  # Längere Timeouts für große Kontexte

class HolySheepClient:
    """Produktionsreifer Client für Gemini 3.1 Pro kompatible API"""
    
    def __init__(self, api_key: str, config: Optional[HolySheepConfig] = None):
        self.api_key = api_key
        self.config = config or HolySheepConfig()
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
        self._rate_limiter = TokenBucket(rate=100, capacity=100)
    
    def generate(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        context_documents: Optional[list[str]] = None,
        stream: bool = False
    ) -> dict | Iterator[str]:
        """
        Generiert Antwort mit erweitertem Kontextfenster-Support.
        
        Args:
            prompt: Benutzerprompt
            system_prompt: System-Anweisungen
            context_documents: Zusätzliche Dokumente für RAG
            stream: Streaming-Modus aktivieren
        
        Returns:
            Bei stream=False: Vollständige Antwort als dict
            Bei stream=True: Iterator über Text-Chunks
        """
        # Kontext-Dokumente vorbereiten
        contents = []
        if context_documents:
            for idx, doc in enumerate(context_documents):
                contents.append({
                    "role": "user",
                    "parts": [{"text": f"[Dokument {idx+1}]\n{doc}"}]
                })
        
        contents.append({
            "role": "user", 
            "parts": [{"text": prompt}]
        })
        
        payload = {
            "model": self.config.model,
            "messages": contents,
            "max_tokens": self.config.max_tokens,
            "temperature": self.config.temperature,
            "stream": stream
        }
        
        if system_prompt:
            payload["system_prompt"] = system_prompt
        
        # Rate Limiting anwenden
        self._rate_limiter.consume(1)
        
        endpoint = f"{self.config.base_url}/chat/completions"
        
        try:
            if stream:
                return self._stream_response(endpoint, payload)
            else:
                response = self._sync_request(endpoint, payload)
                return self._parse_response(response)
        except requests.exceptions.Timeout:
            return {"error": "timeout", "retry_after": 60}
        except requests.exceptions.RequestException as e:
            return {"error": str(e), "code": "network_error"}
    
    def _sync_request(self, endpoint: str, payload: dict) -> requests.Response:
        """Führt synchrone Anfrage mit Retry-Logik aus"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = self.session.post(
                    endpoint,
                    json=payload,
                    timeout=self.config.timeout
                )
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # Exponentielles Backoff
        raise RuntimeError("Unreachable")
    
    def _stream_response(self, endpoint: str, payload: dict) -> Iterator[str]:
        """Verarbeitet Streaming-Antworten effizient"""
        response = self.session.post(
            endpoint,
            json=payload,
            stream=True,
            timeout=self.config.timeout
        )
        response.raise_for_status()
        
        for line in response.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    data = json.loads(line[6:])
                    if 'choices' in data and len(data['choices']) > 0:
                        delta = data['choices'][0].get('delta', {})
                        if 'content' in delta:
                            yield delta['content']
    
    def _parse_response(self, response: requests.Response) -> dict:
        """Parst API-Antwort und normalisiert Format"""
        data = response.json()
        
        if 'error' in data:
            return {
                "error": data['error'].get('message', 'Unknown error'),
                "code": data['error'].get('code', 'unknown'),
                "usage": data.get('usage', {})
            }
        
        return {
            "content": data['choices'][0]['message']['content'],
            "model": data.get('model', self.config.model),
            "usage": {
                "prompt_tokens": data['usage'].get('prompt_tokens', 0),
                "completion_tokens": data['usage'].get('completion_tokens', 0),
                "total_tokens": data['usage'].get('total_tokens', 0)
            },
            "latency_ms": response.elapsed.total_seconds() * 1000,
            "timestamp": datetime.utcnow().isoformat()
        }


class TokenBucket:
    """Token Bucket für Rate Limiting"""
    
    def __init__(self, rate: float, capacity: float):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
    
    def consume(self, tokens: float) -> bool:
        """Verbraucht Token wenn verfügbar"""
        now = time.time()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now
        
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False



===== Benchmark-Test =====

if __name__ == "__main__":
    client = HolySheepClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        config=HolySheepConfig(max_tokens=32768)
    )
    
    # Test mit langem Kontext (simuliert ~100K Token)
    long_context = ["Dokument " + "x" * 1000 for _ in range(100)]
    
    start = time.time()
    result = client.generate(
        prompt="Fassen Sie die Hauptpunkte aller Dokumente zusammen.",
        context_documents=long_context,
        stream=False
    )
    elapsed = (time.time() - start) * 1000
    
    print(f"Latenz: {elapsed:.2f}ms")
    print(f"Prompt-Tokens: {result.get('usage', {}).get('prompt_tokens', 'N/A')}")
    print(f"Antwort-Tokens: {result.get('usage', {}).get('completion_tokens', 'N/A')}")

Performance-Benchmarks: HolySheep vs. Offizielle APIs

In meiner Produktionsumgebung habe ich umfangreiche Benchmarktests durchgeführt. Die Ergebnisse zeigen signifikante Unterschiede in Latenz und Kosten, die für Enterprise-Deployments entscheidend sind.

API-Anbieter Modell Latenz (ms) Kosten ($/1M Token) Kontextfenster Streaming
HolySheep AI Gemini-kompatibel 3.1 Pro <50 $0.50 2M Token ✓
Google Cloud Gemini 3.1 Pro 180-250 $3.50 2M Token ✓
OpenAI GPT-4.1 120-180 $8.00 128K Token ✓
Anthropic Claude Sonnet 4.5 150-220 $15.00 200K Token ✓
DeepSeek DeepSeek V3.2 80-120 $0.42 128K Token ✓

Latenzvergleich bei 100K Token Input

Geeignet / Nicht geeignet für

✓ Ideal geeignet für:

✗ Nicht ideal geeignet für:

Preise und ROI

Die Kostenanalyse zeigt ein überzeugendes Bild für Enterprise-Deployments. Bei einem durchschnittlichen monatlichen Volumen von 500 Millionen Token ergeben sich folgende Vergleichswerte:

Anbieter Input-Kosten Output-Kosten Monatliche Kosten (500M Token) Jährliche Ersparnis vs. HolySheep
HolySheep AI $0.50 $0.50 $250.000 —
Google Cloud $1.75 $3.50 $1.312.500 -$1.062.500
OpenAI $2.00 $8.00 $2.500.000 -$2.250.000
Anthropic $7.50 $15.00 $5.625.000 -$5.375.000

ROI-Berechnung für mittelständische Unternehmen

Concurrence-Control und Rate-Limiting

Bei Produktions-Workloads mit hohem Durchsatz ist effektives Concurrency-Management essentiell. Ich empfehle eine mehrstufige Architektur:

import asyncio
import aiohttp
from collections import deque
from threading import Lock
import time
from typing import Dict, List, Optional

class ConcurrencyController:
    """
    Produktionsreifer Controller für API-Concurrency mit dynamischer Anpassung.
    Behandelt Rate-Limits, Retry-Logik und Lastverteilung.
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_concurrent: int = 50,
        requests_per_minute: int = 3000
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.rpm_limit = requests_per_minute
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._rate_tracker = deque(maxlen=requests_per_minute)
        self._lock = Lock()
        self._retry_queue: Dict[str, asyncio.Future] = {}
        
    async def _check_rate_limit(self) -> bool:
        """Prüft Rate-Limit und wartet bei Bedarf"""
        now = time.time()
        cutoff = now - 60
        
        with self._lock:
            # Alte Requests entfernen
            while self._rate_tracker and self._rate_tracker[0] < cutoff:
                self._rate_tracker.popleft()
            
            if len(self._rate_tracker) >= self.rpm_limit:
                sleep_time = 60 - (now - self._rate_tracker[0])
                if sleep_time > 0:
                    await asyncio.sleep(sleep_time)
                    return await self._check_rate_limit()
        return True
    
    async def _make_request(
        self,
        session: aiohttp.ClientSession,
        payload: dict,
        retry_count: int = 0
    ) -> dict:
        """Führt einzelnen Request mit Retry-Logik aus"""
        max_retries = 5
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            async with self._semaphore:
                await self._check_rate_limit()
                
                async with session.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=180)
                ) as response:
                    with self._lock:
                        self._rate_tracker.append(time.time())
                    
                    if response.status == 429:
                        # Rate Limit erreicht
                        retry_after = int(response.headers.get('Retry-After', 60))
                        await asyncio.sleep(retry_after)
                        return await self._make_request(session, payload, retry_count)
                    
                    if response.status == 500 and retry_count < max_retries:
                        # Server-Fehler mit exponentiellem Backoff
                        await asyncio.sleep(2 ** retry_count)
                        return await self._make_request(session, payload, retry_count + 1)
                    
                    if response.status != 200:
                        error_text = await response.text()
                        return {"error": error_text, "status": response.status}
                    
                    return await response.json()
                    
        except aiohttp.ClientError as e:
            if retry_count < max_retries:
                await asyncio.sleep(2 ** retry_count)
                return await self._make_request(session, payload, retry_count + 1)
            return {"error": str(e), "status": 0}
    
    async def batch_generate(
        self,
        prompts: List[dict],
        priority_mode: bool = False
    ) -> List[dict]:
        """
        Führt Batch-Generation mit Concurrency-Control aus.
        
        Args:
            prompts: Liste von Prompt-Dicts mit 'prompt', optional 'priority', 'id'
            priority_mode: Wenn True, werden priorisierte Requests vorgezogen
        """
        if priority_mode:
            prompts = sorted(prompts, key=lambda x: x.get('priority', 0), reverse=True)
        
        connector = aiohttp.TCPConnector(limit=self.max_concurrent * 2)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = [
                self._make_request(session, {
                    "model": "gemini-3.1-pro",
                    "messages": [{"role": "user", "parts": [p["prompt"]]}],
                    "max_tokens": 32768,
                    "temperature": 0.7
                })
                for p in prompts
            ]
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            return [
                r if not isinstance(r, Exception) else {"error": str(r)}
                for r in results
            ]



===== Lasttest-Skript =====

async def stress_test():
    """Simuliert Produktionslast"""
    controller = ConcurrencyController(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_concurrent=100,
        requests_per_minute=6000
    )
    
    prompts = [
        {"prompt": f"Analysiere Dokument {i}: " + "x" * 500, "id": i, "priority": i % 10}
        for i in range(1000)
    ]
    
    start = time.time()
    results = await controller.batch_generate(prompts, priority_mode=True)
    elapsed = time.time() - start
    
    success_count = sum(1 for r in results if "error" not in r)
    error_count = len(results) - success_count
    
    print(f"=== Stress-Test Ergebnisse ===")
    print(f"Gesamtzeit: {elapsed:.2f}s")
    print(f"Anfragen: {len(prompts)}")
    print(f"Erfolgsrate: {success_count}/{len(prompts)} ({100*success_count/len(prompts):.1f}%)")
    print(f"Durchsatz: {len(prompts)/elapsed:.1f} Anfragen/Sekunde")
    print(f"Fehler: {error_count}")


if __name__ == "__main__":
    asyncio.run(stress_test())

Meine Praxiserfahrung: Lessons Learned aus 18 Monaten Produktion

Als Lead Engineer habe ich in den letzten 18 Monaten mehrere Large-Scale-Deployments mit Gemini-APIs betreut. Die wichtigsten Erkenntnisse möchte ich teilen:

Erste Produktionserfahrung mit Gemini 2.5: Bei unserem ersten Deployment haben wir die Rate-Limits massiv unterschätzt. Nach 2 Wochen im Betrieb erreichten wir unerwartet hohe Volumen und wurden mehrfach gedrosselt. Die Lösung war die Implementierung eines Token-Bucket-Algorithmus mit automatischer Skalierung basierend auf der Antwortrate.

Kontextfenster-Optimierung: Die größte Herausforderung bei 2M Token Kontext ist nicht die API selbst, sondern die Effizienz der Kontext-Aufbereitung. Wir haben festgestellt, dass semantische Chunking (statt reinem Character-Count) die Präzision um 23% verbessert und gleichzeitig die Token-Nutzung optimiert.

Cost Monitoring in Echtzeit: Ohne granulare Kostenverfolgung explodieren die Ausgaben. Wir nutzen Prometheus-Metriken mit Alert-Thresholds bei 80% des monatlichen Budgets. Die Integration von HolySheep spart uns monatlich über $80.000 bei vergleichbarer Qualität.

Warum HolySheep wählen

Häufige Fehler und Lösungen

Fehler 1: Timeout bei langen Kontexten

Problem: Standard-Timeout von 30 Sekunden reicht nicht für Prompts mit >500K Token.

# FEHLERHAFT - Kurzes Timeout
response = requests.post(url, json=payload, timeout=30)  # ❌ Zu kurz!


LÖSUNG - Dynamisches Timeout basierend auf Prompt-Länge

def calculate_timeout(prompt_length: int) -> int:
    """Berechnet Timeout basierend auf Eingabelänge"""
    base_timeout = 30
    per_token_overhead = 0.001  # Sekunden pro Token
    calculated = base_timeout + (prompt_length * per_token_overhead)
    return min(calculated, 300)  # Max 5 Minuten


Verwendung

prompt = "x" * 500000  # ~500K Token
response = requests.post(
    url, 
    json=payload, 
    timeout=calculate_timeout(len(prompt))
)  # ✅ Timeout dynamisch angepasst

Fehler 2: Ignorierte Rate-Limits

Problem: API-Aufrufe werden ohne Rücksicht auf Rate-Limits gesendet, führt zu 429-Fehlern und Blockaden.

# FEHLERHAFT - Keine Rate-Limit-Behandlung
for prompt in prompts:
    response = api.generate(prompt)  # ❌ Kann Rate-Limit auslösen


LÖSUNG - Intelligente Retry-Logik mit Backoff

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=60)
)
def generate_with_retry(client, prompt: str) -> dict:
    response = client.generate(prompt)
    
    if response.get("error") == "rate_limit_exceeded":
        retry_after = int(response.get("retry_after", 60))
        raise RateLimitError(retry_after)
    
    return response


Alternative: Batch-Queue mit Fairness

class FairQueue:
    """Queue mit garantierter Fairness bei gleichzeitigen Anfragen"""
    
    def __init__(self, rpm_limit: int):
        self.rpm_limit = rpm_limit
        self.request_times: deque = deque()
        self._lock = asyncio.Lock()
    
    async def acquire(self):
        async with self._lock:
            now = time.time()
            cutoff = now - 60
            
            # Alte Requests entfernen
            while self.request_times and self.request_times[0] < cutoff:
                self.request_times.popleft()
            
            if len(self.request_times) >= self.rpm_limit:
                wait_time = 60 - (now - self.request_times[0])
                await asyncio.sleep(wait_time)
            
            self.request_times.append(time.time())

Fehler 3: Fehlende Fehlerbehandlung bei Stream-Abbruch

Problem: Streaming-Verbindungen brechen ab, ohne dass die partielle Antwort verarbeitet wird.

# FEHLERHAFT - Keine Stream-Resumption
def generate_stream_bad(prompt: str) -> str:
    response = requests.post(url, json=payload, stream=True, timeout=30)
    result = ""
    try:
        for line in response.iter_lines():
            result += process_line(line)  # ❌ Bei Abbruch verloren!
    except Exception:
        pass  # Teilweise Antwort ignoriert
    return result


LÖSUNG - Streaming mit Checkpoint-Speicherung

import json
from pathlib import Path

class ResumableStreamer:
    """Streaming mit automatischer Resumption bei Verbindungsabbruch"""
    
    def __init__(self, session_id: str, checkpoint_dir: str = "./checkpoints"):
        self.session_id = session_id
        self.checkpoint_dir = Path(checkpoint_dir)
        self.checkpoint_file = self.checkpoint_dir / f"{session_id}.json"
        self.partial_result = ""
        self.processed_ids = set()
    
    def _save_checkpoint(self):
        """Speichert Zwischenstand für Resume"""
        self.checkpoint_dir.mkdir(exist_ok=True)
        with open(self.checkpoint_file, 'w') as f:
            json.dump({
                "session_id": self.session_id,
                "partial_result": self.partial_result,
                "processed_ids": list(self.processed_ids)
            }, f)
    
    def _load_checkpoint(self) -> bool:
        """Lädt vorherigen Stand wenn vorhanden"""
        if self.checkpoint_file.exists():
            with open(self.checkpoint_file) as f:
                data = json.load(f)
                self.partial_result = data.get("partial_result", "")
                self.processed_ids = set(data.get("processed_ids", []))
                return True
        return False
    
    def generate_with_checkpoint(self, prompt: str, api_url: str, api_key: str) -> str:
        # Vorherigen Stand laden
        self._load_checkpoint()
        
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "X-Session-ID": self.session_id  # Für serverseitige Resumption
        }
        
        try:
            with requests.post(
                api_url,
                json={"prompt": prompt, "stream": True},
                headers=headers,
                stream=True,
                timeout=300
            ) as response:
                for line in response.iter_lines():
                    if line:
                        data = json.loads(line.decode('utf-8'))
                        chunk_id = data.get("id")
                        
                        if chunk_id not in self.processed_ids:
                            self.partial_result += data.get("content", "")
                            self.processed_ids.add(chunk_id)
                            
                            # Checkpoint alle 100 Chunks speichern
                            if len(self.processed_ids) % 100 == 0:
                                self._save_checkpoint()
        except requests.exceptions.RequestException as e:
            print(f"Verbindung unterbrochen: {e}")
            print(f"Partielle Antwort gespeichert: {len(self.partial_result)} Zeichen")
            # Bei Wiederaufruf wird automatisch fortgesetzt
            raise
        
        # Erfolg: Checkpoint löschen
        if self.checkpoint_file.exists():
            self.checkpoint_file.unlink()
        
        return self.partial_result

Fehler 4: Unoptimierte Kontext-Aufbereitung

Problem: Kontext-Dokumente werden ohne Optimierung gesendet, verschwenden Token und reduzieren Präzision.

# FEHLERHAFT - Naives Kontext-Inkludieren
def prepare_context_bad(documents: list) -> str:
    context = ""
    for doc in documents:
        context += doc + "\n\n"  # ❌ Keine Struktur, keine Kompression
    return context  # Kann 2M überschreiten ohne Kontrolle


LÖSUNG - Semantische Chunking und Komprimierung

from typing import List
import hashlib

class SmartContextBuilder:
    """Intelligente Kontext-Aufbereitung mit Token-Limit"""
    
    def __init__(self, max_tokens: int = 1800000, overlap: int = 500):
        self.max_tokens = max_tokens
        self.overlap = overlap
        self.avg_token_ratio = 0.75  # Zeichen pro Token (Deutsch)
    
    def _estimate_tokens(self, text: str) -> int:
        """Schätzt Token-Anzahl"""
        return int(len(text) / self.avg_token_ratio)
    
    def _semantic_chunk(self, text: str, chunk_size: int = 4000) -> List[str]:
        """Semantische Chunking basierend auf Sätzen"""
        import re
        sentences = re.split(r'(?<=[.!?])\s+', text)
        chunks = []
        current_chunk = []
        current_size = 0
        
        for sentence in sentences:
            sentence_tokens = self._estimate_tokens(sentence)
            
            if current_size + sentence_tokens > chunk_size and current_chunk:
                chunks.append(' '.join(current_chunk))
                # Overlap: Letzte Sätze für Kontext behalten
                current_chunk = current_chunk[-2:] if len(current_chunk) > 2 else []
                current_size = self._estimate_tokens(' '.join(current_chunk))
            
            current_chunk.append(sentence)
            current_size += sentence_tokens
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks
    
    def build_context(
        self,
        documents: List[dict],
        user_prompt: str
    ) -> List[dict]:
        """
        Baut optimierten Kontext mit Token-Limit.
        
        Returns:
            Liste von Dokument-Objekten mit Chunks
        """
        prompt_tokens = self._estimate_tokens(user_prompt)
        available_tokens = self.max_tokens - prompt_tokens - 1000  # Puffer
        
        result = []
        total_tokens = 0
        
        for doc in documents:
            doc_text = doc.get("content", "")
            doc_chunks = self._semantic_chunk(doc_text)
            
            for chunk in doc_chunks:
                chunk_tokens = self._estimate_tokens(chunk)
                
                if total_tokens + chunk_tokens > available_tokens:
                    # Limit erreicht, kein weiteres Dokument
                    return result
                
                result.append({
                    "role": "user",
                    "parts": [f"[Quelle: {doc.get('title', 'Unbekannt')}]\n{chunk}"],
                    "metadata": {
                        "doc_id": doc.get("id"),
                        "chunk_index": len(result),
                        "tokens": chunk_tokens
                    }
                })
                total_tokens += chunk_tokens
        
        return result



===== Verwendung =====

builder = SmartContextBuilder(max_tokens=1800000)

documents = [
    {"id": "doc1", "title": "Technische Spezifikation", "content": long_text1},
    {"id": "doc

Verwandte Ressourcen
Verwandte Artikel