19 Tools in LangChain: Die Schnittstelle zwischen Modell und Welt

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Durchsucht die Kundendatenbank nach passenden Einträgen.
    
    Args:
        query: Suchbegriffe
        limit: Maximale Anzahl der Ergebnisse
    """
    return f"Gefunden: {limit} Ergebnisse für '{query}'"

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    """Input-Schema für Wetterabfragen."""
    location: str = Field(description="Stadt oder Koordinaten")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="Temperatureinheit"
    )
    include_forecast: bool = Field(
        default=False,
        description="5-Tage-Vorhersage einbeziehen"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """Liefert aktuelle Wetterdaten und optionale Vorhersage."""
    temp = 22 if units == "celsius" else 72
    result = f"Aktuelles Wetter in {location}: {temp}°{units[0].upper()}"
    if include_forecast:
        result += "\nNächste 5 Tage: Sonnig"
    return result

from langchain.tools import tool, ToolRuntime

@tool
def get_user_preference(
    pref_name: str,
    runtime: ToolRuntime
) -> str:
    """Liest eine Nutzer-Präferenz aus."""
    preferences = runtime.state.get("user_preferences", {})
    return preferences.get(pref_name, "Nicht gesetzt")

from langgraph.types import Command

@tool
def update_user_name(new_name: str, runtime: ToolRuntime) -> Command:
    """Aktualisiert den Namen des Nutzers im State."""
    return Command(update={"user_name": new_name})

@tool
def analyze_document(doc_id: str, runtime: ToolRuntime) -> str:
    """Analysiert ein Dokument und extrahiert Metadaten."""
    writer = runtime.stream_writer
    
    writer("Lade Dokument...")
    # Document laden (2 Sekunden)
    
    writer("Extrahiere Text...")
    # Text-Extraktion (3 Sekunden)
    
    writer("Analysiere Inhalt...")
    # NLP-Analyse (5 Sekunden)
    
    return "Analyse abgeschlossen: 42 Entitäten gefunden"

from langchain.tools import tool

@tool
def get_server_time() -> str:
    """Gibt die aktuelle Serverzeit zurück."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S Uhr")

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-4"),
    tools=[get_server_time]
)

response = agent.invoke({
    "messages": [{"role": "user", "content": "Wie spät ist es?"}]
})

print(response["messages"][-1].content)
# "Es ist aktuell 14:23:17 Uhr."

@tool(
    name="current_time",
    description="Nutze dies, um die aktuelle Uhrzeit zu erfahren. Wichtig für zeitbasierte Anfragen."
)
def get_server_time() -> str:
    """Gibt die aktuelle Serverzeit zurück."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S Uhr")

from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Literal, Optional

class ProductSearchInput(BaseModel):
    """Schema für Produktsuchanfragen."""
    query: str = Field(description="Suchbegriff oder Produktname")
    category: Optional[Literal["electronics", "books", "clothing"]] = Field(
        default=None,
        description="Optional: Produktkategorie zur Einschränkung"
    )
    max_price: Optional[float] = Field(
        default=None,
        description="Optional: Maximaler Preis in Euro"
    )
    sort_by: Literal["relevance", "price", "rating"] = Field(
        default="relevance",
        description="Sortierreihenfolge der Ergebnisse"
    )

@tool(args_schema=ProductSearchInput)
def search_products(
    query: str,
    category: Optional[str] = None,
    max_price: Optional[float] = None,
    sort_by: str = "relevance"
) -> str:
    """Durchsucht den Produktkatalog nach passenden Artikeln."""
    
    # Simulierte Produktsuche
    results = []
    
    if category:
        results.append(f"Gefiltert nach Kategorie: {category}")
    
    if max_price:
        results.append(f"Maximaler Preis: {max_price}€")
    
    results.append(f"Sortierung: {sort_by}")
    results.append(f"Gefunden: 12 Produkte für '{query}'")
    
    return "\n".join(results)

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict

# Custom State mit User-Präferenzen
class ShoppingAgentState(AgentState):
    user_preferences: dict
    cart_items: list[str]

@tool
def recommend_products(
    category: str,
    runtime: ToolRuntime[None, ShoppingAgentState]
) -> str:
    """Empfiehlt Produkte basierend auf Nutzerpräferenzen."""
    
    # Präferenzen aus dem State lesen
    preferences = runtime.state.get("user_preferences", {})
    preferred_brands = preferences.get("brands", [])
    budget = preferences.get("budget", "medium")
    
    # Empfehlungen generieren (simuliert)
    recommendations = []
    
    if preferred_brands:
        recommendations.append(f"Basierend auf deinen bevorzugten Marken: {', '.join(preferred_brands)}")
    
    recommendations.append(f"Budget-Level: {budget}")
    recommendations.append(f"Top-Empfehlung in {category}: Premium Widget Pro")
    
    return "\n".join(recommendations)

# Agent mit Custom State erstellen
agent = create_agent(
    model="openai:gpt-4",
    tools=[recommend_products],
    state_schema=ShoppingAgentState,
    checkpointer=InMemorySaver()
)

# Erste Interaktion: Präferenzen setzen
response1 = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Ich mag die Marken Apple und Sony, Budget ist hoch"}],
        "user_preferences": {"brands": ["Apple", "Sony"], "budget": "high"},
        "cart_items": []
    },
    {"configurable": {"thread_id": "user_001"}}
)

# Zweite Interaktion: Tool nutzt gespeicherte Präferenzen
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "Empfiehl mir Elektronik"}]},
    {"configurable": {"thread_id": "user_001"}}
)

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime

# Context-Schema definieren
@dataclass
class UserContext:
    user_id: str
    api_key: str
    region: str

# Simulierte Datenbank
USER_DATABASE = {
    "user_123": {
        "name": "Anna Schmidt",
        "orders": ["Order-001", "Order-042", "Order-103"],
        "membership": "Gold"
    },
    "user_456": {
        "name": "Thomas Müller",
        "orders": ["Order-234"],
        "membership": "Standard"
    }
}

@tool
def get_order_history(
    runtime: ToolRuntime[UserContext, AgentState]
) -> str:
    """Ruft die Bestellhistorie des aktuellen Nutzers ab."""
    
    # User-ID aus Context holen
    user_id = runtime.context.user_id
    
    if user_id not in USER_DATABASE:
        return "Nutzer nicht gefunden"
    
    user_data = USER_DATABASE[user_id]
    orders = user_data["orders"]
    membership = user_data["membership"]
    
    return f"Mitgliedschaft: {membership}\nBestellungen: {', '.join(orders)}"

@tool
def get_account_status(
    runtime: ToolRuntime[UserContext, AgentState]
) -> str:
    """Zeigt den Status des Nutzer-Accounts an."""
    
    user_id = runtime.context.user_id
    region = runtime.context.region
    
    if user_id not in USER_DATABASE:
        return "Nutzer nicht gefunden"
    
    user_data = USER_DATABASE[user_id]
    name = user_data["name"]
    membership = user_data["membership"]
    
    return f"Account: {name}\nMitgliedschaft: {membership}\nRegion: {region}"

# Agent mit Context-Schema erstellen
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-4"),
    tools=[get_order_history, get_account_status],
    context_schema=UserContext
)

# Context bei jedem Request übergeben
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Zeig mir meine Bestellungen"}]},
    context=UserContext(
        user_id="user_123",
        api_key="secret_key_xyz",
        region="EU"
    )
)

print(response["messages"][-1].content)
# "Du hast folgende Bestellungen: Order-001, Order-042, Order-103. 
#  Deine Mitgliedschaft ist Gold."

from langchain.tools import tool, ToolRuntime
from langgraph.types import Command
from langchain.agents import AgentState
from typing import TypedDict

class ShoppingState(AgentState):
    cart_items: list[str]
    cart_total: float

@tool
def add_to_cart(
    product_id: str,
    price: float,
    runtime: ToolRuntime[None, ShoppingState]
) -> Command:
    """Fügt ein Produkt zum Warenkorb hinzu."""
    
    # Aktuellen Cart aus State lesen
    current_items = runtime.state.get("cart_items", [])
    current_total = runtime.state.get("cart_total", 0.0)
    
    # Neue Werte berechnen
    new_items = current_items + [product_id]
    new_total = current_total + price
    
    # State-Update via Command zurückgeben
    return Command(
        update={
            "cart_items": new_items,
            "cart_total": new_total
        },
        # Tool-Response (wird auch ans Modell zurückgegeben)
        result=f"Produkt {product_id} für {price}€ hinzugefügt. Warenkorbsumme: {new_total}€"
    )

@tool
def clear_cart(runtime: ToolRuntime[None, ShoppingState]) -> Command:
    """Leert den Warenkorb komplett."""
    
    return Command(
        update={
            "cart_items": [],
            "cart_total": 0.0
        },
        result="Warenkorb wurde geleert."
    )

@tool
def show_cart(runtime: ToolRuntime[None, ShoppingState]) -> str:
    """Zeigt den aktuellen Warenkorbinhalt."""
    
    items = runtime.state.get("cart_items", [])
    total = runtime.state.get("cart_total", 0.0)
    
    if not items:
        return "Dein Warenkorb ist leer."
    
    return f"Warenkorb: {', '.join(items)}\nGesamtsumme: {total}€"

from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from typing import Any

@tool
def save_user_preference(
    key: str,
    value: str,
    runtime: ToolRuntime
) -> str:
    """Speichert eine Nutzer-Präferenz dauerhaft."""
    
    store = runtime.store
    user_id = "default_user"  # In Production aus Context holen
    
    # Im Store speichern: (namespace, key) -> value
    store.put(("preferences",), user_id, {key: value})
    
    return f"Präferenz gespeichert: {key} = {value}"

@tool
def get_user_preference(
    key: str,
    runtime: ToolRuntime
) -> str:
    """Liest eine gespeicherte Nutzer-Präferenz."""
    
    store = runtime.store
    user_id = "default_user"
    
    # Aus Store lesen
    item = store.get(("preferences",), user_id)
    
    if item is None:
        return f"Keine Präferenz für '{key}' gefunden"
    
    prefs = item.value
    value = prefs.get(key, "Nicht gesetzt")
    
    return f"{key}: {value}"

@tool
def list_preferences(runtime: ToolRuntime) -> str:
    """Listet alle gespeicherten Präferenzen auf."""
    
    store = runtime.store
    user_id = "default_user"
    
    item = store.get(("preferences",), user_id)
    
    if item is None or not item.value:
        return "Keine Präferenzen gespeichert"
    
    prefs = item.value
    lines = [f"{k}: {v}" for k, v in prefs.items()]
    
    return "Gespeicherte Präferenzen:\n" + "\n".join(lines)

# Agent mit Store erstellen
store = InMemoryStore()

agent = create_agent(
    model=ChatOpenAI(model="gpt-4"),
    tools=[save_user_preference, get_user_preference, list_preferences],
    store=store
)

# Session 1: Präferenzen speichern
response1 = agent.invoke({
    "messages": [{"role": "user", "content": "Meine Lieblingsfarbe ist Blau"}]
})

# Session 2: Präferenzen abrufen (komplett neue Konversation)
response2 = agent.invoke({
    "messages": [{"role": "user", "content": "Was ist meine Lieblingsfarbe?"}]
})

print(response2["messages"][-1].content)
# "Deine Lieblingsfarbe ist Blau."

from langchain.tools import tool, ToolRuntime
import time

@tool
def analyze_large_dataset(
    dataset_id: str,
    runtime: ToolRuntime
) -> str:
    """Analysiert einen großen Datensatz (dauert einige Sekunden)."""
    
    writer = runtime.stream_writer
    
    # Phase 1
    writer("📥 Lade Datensatz...")
    time.sleep(1)  # Simulierte Ladezeit
    
    # Phase 2
    writer("🔍 Extrahiere Features...")
    time.sleep(1.5)
    
    # Phase 3
    writer("📊 Führe statistische Analyse durch...")
    time.sleep(1.5)
    
    # Phase 4
    writer("🧮 Berechne Korrelationen...")
    time.sleep(1)
    
    # Finales Ergebnis
    return f"Analyse von {dataset_id} abgeschlossen: 42 signifikante Muster gefunden"

from langchain.tools import tool
import requests

@tool
def fetch_api_data(endpoint: str) -> str:
    """Ruft Daten von einer externen API ab."""
    
    base_url = "https://api.example.com"
    
    try:
        response = requests.get(
            f"{base_url}/{endpoint}",
            timeout=5
        )
        response.raise_for_status()
        
        data = response.json()
        return f"Daten erhalten: {data}"
        
    except requests.Timeout:
        return "Fehler: API-Anfrage hat das Zeitlimit überschritten. Bitte versuche es später erneut."
    
    except requests.ConnectionError:
        return "Fehler: Verbindung zur API konnte nicht hergestellt werden. Prüfe deine Netzwerkverbindung."
    
    except requests.HTTPError as e:
        if response.status_code == 404:
            return f"Fehler: Endpoint '{endpoint}' existiert nicht."
        elif response.status_code == 429:
            return "Fehler: Zu viele Anfragen. Rate Limit erreicht. Bitte warte eine Minute."
        else:
            return f"Fehler: API gab Statuscode {response.status_code} zurück."
    
    except Exception as e:
        return f"Unerwarteter Fehler: {str(e)}"

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent, AgentState
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from typing import TypedDict
from dataclasses import dataclass

# Context: Unveränderliche Session-Daten
@dataclass
class CustomerContext:
    customer_id: str
    session_id: str

# State: Veränderliche Konversationsdaten
class CustomerServiceState(AgentState):
    customer_name: str
    pending_issues: list[str]
    satisfaction_score: int

# Simulierte Datenbanken
CUSTOMER_DB = {
    "cust_001": {
        "name": "Lisa Meier",
        "orders": ["ORD-123", "ORD-456"],
        "email": "lisa@example.com"
    },
    "cust_002": {
        "name": "Max Weber",
        "orders": ["ORD-789"],
        "email": "max@example.com"
    }
}

ORDER_DB = {
    "ORD-123": {"status": "Versandt", "tracking": "DE123456789"},
    "ORD-456": {"status": "In Bearbeitung", "tracking": None},
    "ORD-789": {"status": "Zugestellt", "tracking": "DE987654321"}
}

# Tools definieren

@tool
def get_customer_info(
    runtime: ToolRuntime[CustomerContext, CustomerServiceState]
) -> Command:
    """Lädt Kundeninformationen aus der Datenbank."""
    
    customer_id = runtime.context.customer_id
    
    if customer_id not in CUSTOMER_DB:
        return Command(result="Kunde nicht gefunden")
    
    customer = CUSTOMER_DB[customer_id]
    
    # State aktualisieren mit Kundenname
    return Command(
        update={"customer_name": customer["name"]},
        result=f"Kunde: {customer['name']}\nBestellungen: {', '.join(customer['orders'])}\nE-Mail: {customer['email']}"
    )

@tool
def track_order(order_id: str) -> str:
    """Verfolgt den Status einer Bestellung."""
    
    if order_id not in ORDER_DB:
        return f"Bestellung {order_id} nicht gefunden"
    
    order = ORDER_DB[order_id]
    status = order["status"]
    tracking = order["tracking"]
    
    result = f"Bestellung {order_id}:\nStatus: {status}"
    
    if tracking:
        result += f"\nTracking-Nummer: {tracking}"
    
    return result

@tool
def report_issue(
    issue_description: str,
    runtime: ToolRuntime[CustomerContext, CustomerServiceState]
) -> Command:
    """Meldet ein Problem und fügt es zur Liste offener Issues hinzu."""
    
    current_issues = runtime.state.get("pending_issues", [])
    new_issues = current_issues + [issue_description]
    
    return Command(
        update={"pending_issues": new_issues},
        result=f"Issue gemeldet: {issue_description}\nIssue-Nummer: #{len(new_issues)}"
    )

@tool
def show_pending_issues(
    runtime: ToolRuntime[CustomerContext, CustomerServiceState]
) -> str:
    """Zeigt alle offenen Issues des Kunden."""
    
    issues = runtime.state.get("pending_issues", [])
    
    if not issues:
        return "Keine offenen Issues"
    
    lines = [f"#{i+1}: {issue}" for i, issue in enumerate(issues)]
    return "Offene Issues:\n" + "\n".join(lines)

@tool
def rate_service(
    rating: int,
    runtime: ToolRuntime[CustomerContext, CustomerServiceState]
) -> Command:
    """Bewertet den Kundenservice (1-5 Sterne)."""
    
    if rating < 1 or rating > 5:
        return Command(result="Bewertung muss zwischen 1 und 5 liegen")
    
    return Command(
        update={"satisfaction_score": rating},
        result=f"Vielen Dank für deine Bewertung: {rating}/5 Sternen"
    )

# Agent erstellen
agent = create_agent(
    model=ChatOpenAI(model="gpt-4", temperature=0),
    tools=[
        get_customer_info,
        track_order,
        report_issue,
        show_pending_issues,
        rate_service
    ],
    state_schema=CustomerServiceState,
    context_schema=CustomerContext,
    checkpointer=InMemorySaver(),
    system_prompt="""Du bist ein freundlicher Kundenservice-Agent.
    Nutze die verfügbaren Tools, um Kundenanfragen zu bearbeiten.
    Lade zuerst immer die Kundeninformationen, bevor du andere Aktionen durchführst."""
)

# Verwendung
config = {"configurable": {"thread_id": "session_001"}}
context = CustomerContext(customer_id="cust_001", session_id="sess_xyz")

# Interaktion 1: Kundeninfo laden
response1 = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hallo, ich brauche Hilfe"}],
        "customer_name": "",
        "pending_issues": [],
        "satisfaction_score": 0
    },
    context=context,
    config=config
)

print("Bot:", response1["messages"][-1].content)

# Interaktion 2: Bestellung tracken
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "Wo ist meine Bestellung ORD-123?"}]},
    context=context,
    config=config
)

print("Bot:", response2["messages"][-1].content)

# Interaktion 3: Issue melden
response3 = agent.invoke(
    {"messages": [{"role": "user", "content": "Die Sendung ist beschädigt angekommen"}]},
    context=context,
    config=config
)

print("Bot:", response3["messages"][-1].content)

# Interaktion 4: Service bewerten
response4 = agent.invoke(
    {"messages": [{"role": "user", "content": "Ich gebe 5 Sterne"}]},
    context=context,
    config=config
)

print("Bot:", response4["messages"][-1].content)