23 APIs als dynamische Datenquelle: Wenn Wissen sich bewegt

import requests
from typing import Optional, Dict, Any

def get_order_status(order_id: str) -> Optional[Dict[str, Any]]:
    """
    Ruft den aktuellen Status einer Bestellung von einer REST-API ab.
    
    Args:
        order_id: Die eindeutige Bestell-ID
        
    Returns:
        Dict mit Bestelldetails oder None bei Fehler
    """
    api_url = f"https://api.example.com/orders/{order_id}"
    headers = {"Authorization": "Bearer YOUR_API_TOKEN"}
    
    try:
        response = requests.get(api_url, headers=headers, timeout=5)
        response.raise_for_status()  # Wirft Exception bei HTTP-Fehler
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"API-Fehler: {e}")
        return None

# Verwendung
order_data = get_order_status("ORD-12345")
if order_data:
    status = order_data.get("status")
    estimated_delivery = order_data.get("estimated_delivery")
    print(f"Status: {status}, Lieferung: {estimated_delivery}")

from typing import List, Dict, Any
import requests
from langchain_core.tools import tool

@tool
def get_product_info(product_id: str, fields: List[str]) -> Dict[str, Any]:
    """
    Ruft spezifische Produktinformationen von der API ab.
    
    Args:
        product_id: Die Produkt-ID
        fields: Liste der gewünschten Felder (z.B. ["price", "availability"])
        
    Returns:
        Dict mit den angeforderten Produktdaten
    """
    # API-Aufruf
    api_url = f"https://api.shop.com/products/{product_id}"
    response = requests.get(api_url)
    full_data = response.json()
    
    # Nur angeforderte Felder extrahieren
    filtered_data = {
        field: full_data.get(field, "Nicht verfügbar")
        for field in fields
    }
    
    return filtered_data

import asyncio
from typing import AsyncIterator, Dict, Any
from collections import deque
import json

class StreamAggregator:
    """
    Konsumiert einen Daten-Stream und aggregiert Werte über Zeitfenster.
    """
    def __init__(self, window_size: int = 10):
        self.window_size = window_size
        self.buffer = deque(maxlen=window_size)
    
    async def consume_stream(self, stream_url: str) -> AsyncIterator[Dict[str, Any]]:
        """
        Konsumiert einen SSE-Stream und liefert aggregierte Daten.
        
        In einer realen Implementierung würde hier eine Bibliothek wie
        httpx oder aiohttp zum Einsatz kommen.
        """
        # Vereinfachte Simulation eines Streams
        async for event in self._mock_stream():
            self.buffer.append(event)
            
            # Alle 10 Events eine Aggregation liefern
            if len(self.buffer) == self.window_size:
                yield self._aggregate()
    
    def _aggregate(self) -> Dict[str, Any]:
        """
        Aggregiert die Werte im Buffer.
        """
        if not self.buffer:
            return {}
        
        # Beispiel: Durchschnittswert berechnen
        values = [event['value'] for event in self.buffer]
        return {
            "avg_value": sum(values) / len(values),
            "min_value": min(values),
            "max_value": max(values),
            "data_points": len(values)
        }
    
    async def _mock_stream(self):
        """Mock-Stream für Demonstration."""
        import random
        for i in range(50):
            yield {"timestamp": i, "value": random.uniform(10, 100)}
            await asyncio.sleep(0.1)

# Verwendung
async def main():
    aggregator = StreamAggregator(window_size=10)
    
    async for snapshot in aggregator.consume_stream("https://api.example.com/stream"):
        print(f"Aggregierte Daten: {snapshot}")
        # Hier könnte man den Snapshot an ein LLM weitergeben

# asyncio.run(main())

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
import requests

# Tool 1: Wetter-API
@tool
def get_weather(location: str) -> str:
    """
    Ruft die aktuelle Wettervorhersage für einen Ort ab.
    
    Args:
        location: Der Ort, für den das Wetter abgerufen werden soll
    """
    # In Realität: echter API-Aufruf
    # Hier: Simulation
    api_url = f"https://api.weather.com/v1/forecast?location={location}"
    # response = requests.get(api_url)
    # return response.json()["forecast"]
    
    # Mock-Antwort
    return f"Das Wetter in {location}: Sonnig, 22°C, leichter Wind"

# Tool 2: Kalender-API
@tool
def check_calendar(date: str) -> str:
    """
    Prüft, ob an einem bestimmten Datum Termine vorliegen.
    
    Args:
        date: Datum im Format YYYY-MM-DD
    """
    # Mock-Antwort
    return f"Am {date} sind 2 Termine: Meeting um 10:00, Review um 14:00"

# Tool 3: Bestellstatus-API
@tool
def get_order_status(order_id: str) -> str:
    """
    Ruft den aktuellen Status einer Bestellung ab.
    
    Args:
        order_id: Die Bestellnummer
    """
    # Mock-Antwort
    return f"Bestellung {order_id}: In Zustellung, geschätzte Ankunft morgen"

# Tools zusammenstellen
tools = [get_weather, check_calendar, get_order_status]

# Modell mit Tool-Calling-Fähigkeit
model = ChatOpenAI(model="gpt-4o", temperature=0)

# Prompt-Template für den Agent
prompt = ChatPromptTemplate.from_messages([
    ("system", """Du bist ein hilfreicher Assistent mit Zugriff auf verschiedene 
    Tools. Nutze die verfügbaren Tools, um Fragen zu beantworten. Wenn du ein Tool 
    brauchst, rufe es auf. Wenn du alle nötigen Informationen hast, gib eine 
    finale Antwort."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# Agent erstellen
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Agent nutzen
result = agent_executor.invoke({
    "input": "Wie wird das Wetter morgen in Berlin und habe ich an diesem Tag Termine?"
})

print("\n=== FINALE ANTWORT ===")
print(result["output"])

from typing import List, Iterator
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
import requests

class CustomAPILoader(BaseLoader):
    """
    Custom Loader für eine proprietäre REST-API, die Artikel zurückgibt.
    """
    
    def __init__(self, api_url: str, api_key: str, category: str = None):
        """
        Args:
            api_url: Basis-URL der API
            api_key: Authentifizierungs-Token
            category: Optional, filtert nach Kategorie
        """
        self.api_url = api_url
        self.api_key = api_key
        self.category = category
    
    def lazy_load(self) -> Iterator[Document]:
        """
        Lazy Loading: Generiert Dokumente on-demand.
        
        Wichtig für große Datenmengen, um Speicher zu sparen.
        """
        headers = {"Authorization": f"Bearer {self.api_key}"}
        params = {"category": self.category} if self.category else {}
        
        # Pagination handhaben
        page = 1
        while True:
            params["page"] = page
            response = requests.get(
                f"{self.api_url}/articles",
                headers=headers,
                params=params
            )
            
            if response.status_code != 200:
                break
            
            data = response.json()
            articles = data.get("articles", [])
            
            if not articles:
                break
            
            # Jeder Artikel wird zu einem Document
            for article in articles:
                yield Document(
                    page_content=article["content"],
                    metadata={
                        "source": f"{self.api_url}/articles/{article['id']}",
                        "title": article["title"],
                        "author": article["author"],
                        "published": article["published_date"],
                        "category": article.get("category"),
                    }
                )
            
            page += 1
            
            # Abbruchbedingung: keine weitere Seite
            if not data.get("has_next_page"):
                break
    
    def load(self) -> List[Document]:
        """
        Lädt alle Dokumente auf einmal.
        
        Nutzt lazy_load intern, sammelt aber alle Resultate.
        """
        return list(self.lazy_load())

# Verwendung
loader = CustomAPILoader(
    api_url="https://api.company.com",
    api_key="your-api-key-here",
    category="technical"
)

# Dokumente laden
documents = loader.load()
print(f"Geladen: {len(documents)} Dokumente")

# Diese können jetzt in einen Vector Store eingefügt werden
# oder für RAG genutzt werden

from functools import wraps
from datetime import datetime, timedelta
from typing import Any, Dict, Optional, Callable
import hashlib
import json

class APICache:
    """
    Einfacher In-Memory-Cache für API-Responses.
    """
    
    def __init__(self, default_ttl: int = 300):
        """
        Args:
            default_ttl: Time-to-live in Sekunden (Standard: 5 Minuten)
        """
        self.cache: Dict[str, tuple[Any, datetime]] = {}
        self.default_ttl = default_ttl
    
    def _make_key(self, func_name: str, args: tuple, kwargs: dict) -> str:
        """
        Generiert einen eindeutigen Cache-Key aus Funktionsname und Parametern.
        """
        # Serialisiere Parameter zu String
        key_data = f"{func_name}:{args}:{sorted(kwargs.items())}"
        # Hash für kompakte Keys
        return hashlib.md5(key_data.encode()).hexdigest()
    
    def get(self, key: str) -> Optional[Any]:
        """
        Holt einen Wert aus dem Cache, wenn er existiert und nicht abgelaufen ist.
        """
        if key not in self.cache:
            return None
        
        value, expiry = self.cache[key]
        if datetime.now() > expiry:
            # Abgelaufen, entfernen
            del self.cache[key]
            return None
        
        return value
    
    def set(self, key: str, value: Any, ttl: Optional[int] = None):
        """
        Speichert einen Wert im Cache.
        """
        ttl = ttl or self.default_ttl
        expiry = datetime.now() + timedelta(seconds=ttl)
        self.cache[key] = (value, expiry)
    
    def cached_call(self, ttl: Optional[int] = None):
        """
        Decorator, der API-Aufrufe cached.
        """
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Cache-Key generieren
                cache_key = self._make_key(func.__name__, args, kwargs)
                
                # Prüfen, ob gecached
                cached_result = self.get(cache_key)
                if cached_result is not None:
                    print(f"Cache Hit für {func.__name__}")
                    return cached_result
                
                # Nicht gecached, API aufrufen
                print(f"Cache Miss für {func.__name__}, rufe API auf")
                result = func(*args, **kwargs)
                
                # Resultat cachen
                self.set(cache_key, result, ttl)
                
                return result
            
            return wrapper
        return decorator

# Globale Cache-Instanz
api_cache = APICache(default_ttl=300)

# Verwendung mit Tool
@tool
@api_cache.cached_call(ttl=600)  # 10 Minuten Cache
def get_stock_price(symbol: str) -> str:
    """
    Ruft den aktuellen Aktienkurs ab.
    
    Args:
        symbol: Das Aktiensymbol (z.B. 'AAPL')
    """
    # Simulierter API-Aufruf
    import random
    price = random.uniform(100, 200)
    return f"Aktienkurs für {symbol}: ${price:.2f}"

# Erster Aufruf: Cache Miss, API wird aufgerufen
print(get_stock_price("AAPL"))

# Zweiter Aufruf innerhalb von 10 Minuten: Cache Hit
print(get_stock_price("AAPL"))

from typing import Optional, Dict, Any
import requests
from requests.exceptions import RequestException, Timeout
from langchain_core.tools import tool

def safe_api_call(
    url: str,
    method: str = "GET",
    timeout: int = 5,
    retries: int = 3,
    **kwargs
) -> Optional[Dict[str, Any]]:
    """
    Wrapper für robuste API-Aufrufe mit Retry-Logik.
    
    Args:
        url: Die API-URL
        method: HTTP-Methode (GET, POST, etc.)
        timeout: Timeout in Sekunden
        retries: Anzahl Wiederholungsversuche bei Fehler
        **kwargs: Weitere Parameter für requests (headers, params, etc.)
    
    Returns:
        Response-JSON oder None bei Fehler
    """
    for attempt in range(retries):
        try:
            response = requests.request(
                method=method,
                url=url,
                timeout=timeout,
                **kwargs
            )
            
            # Erfolgreicher Response
            if 200 <= response.status_code < 300:
                return response.json()
            
            # Client-Fehler (4xx) - kein Retry sinnvoll
            if 400 <= response.status_code < 500:
                print(f"Client-Fehler {response.status_code}: {response.text}")
                return None
            
            # Server-Fehler (5xx) - Retry könnte helfen
            print(f"Server-Fehler {response.status_code}, Versuch {attempt + 1}/{retries}")
            
        except Timeout:
            print(f"Timeout bei Versuch {attempt + 1}/{retries}")
        except RequestException as e:
            print(f"Request-Fehler: {e}, Versuch {attempt + 1}/{retries}")
        
        # Exponentielles Backoff zwischen Retries
        if attempt < retries - 1:
            import time
            time.sleep(2 ** attempt)
    
    # Alle Retries fehlgeschlagen
    return None

@tool
def get_weather_robust(location: str) -> str:
    """
    Ruft Wetterdaten ab mit robustem Error Handling.
    
    Args:
        location: Der Ort für die Wettervorhersage
    """
    api_url = f"https://api.weather.com/v1/forecast"
    params = {"location": location}
    headers = {"Authorization": "Bearer API_KEY"}
    
    result = safe_api_call(
        url=api_url,
        method="GET",
        params=params,
        headers=headers,
        timeout=5,
        retries=3
    )
    
    if result is None:
        return f"Entschuldigung, die Wetter-API ist momentan nicht verfügbar für {location}."
    
    # Daten extrahieren und formatieren
    forecast = result.get("forecast", {})
    temp = forecast.get("temperature", "unbekannt")
    condition = forecast.get("condition", "unbekannt")
    
    return f"Wetter in {location}: {condition}, {temp}°C"

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from datetime import datetime, timedelta

# Tools definieren
@tool
def check_calendar(date: str) -> str:
    """Prüft Termine an einem Datum (YYYY-MM-DD)."""
    # Mock
    return f"Am {date}: 1 Meeting um 15:00"

@tool
def get_weather_forecast(location: str, date: str) -> str:
    """Holt Wettervorhersage für Ort und Datum."""
    # Mock
    return f"Wetter in {location} am {date}: Sonnig, 24°C"

@tool
def book_flight(destination: str, date: str) -> str:
    """Bucht einen Flug zum Ziel am Datum."""
    # Mock
    return f"Flug nach {destination} am {date} gebucht (Buchungsnummer: FL-12345)"

@tool
def get_next_week_date() -> str:
    """Gibt das Datum in einer Woche zurück."""
    next_week = datetime.now() + timedelta(days=7)
    return next_week.strftime("%Y-%m-%d")

# Agent konfigurieren
tools = [check_calendar, get_weather_forecast, book_flight, get_next_week_date]
model = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Du bist ein Reise-Assistent. Nutze die verfügbaren Tools, um
    Reisen zu planen. Prüfe immer erst Kalender und Wetter, bevor du Flüge buchst.
    Wenn Termine vorhanden sind oder das Wetter schlecht ist, rate vom Flug ab."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10  # Verhindert Endlosschleifen
)

# Komplexe Anfrage
result = agent_executor.invoke({
    "input": """Buche mir einen Flug nach Berlin nächste Woche, aber nur wenn
    das Wetter gut ist und ich keine wichtigen Termine habe."""
})

print("\n=== FINALE ANTWORT ===")
print(result["output"])