Eine Low-Code-Plattform, die die Einfachheit von No-Code mit der Leistung von Full-Code verbindet 🚀
Jetzt kostenlos starten

LangChain Tools: Vollständige Anleitung zum Erstellen und Verwenden benutzerdefinierter LLM-Tools + Codebeispiele 2025

Beschreiben Sie, was Sie automatisieren möchten

Latenode verwandelt Ihre Eingabeaufforderung in Sekundenschnelle in einen einsatzbereiten Workflow

Geben Sie eine Nachricht ein

Unterstützt von Latenode AI

Es dauert einige Sekunden, bis die magische KI Ihr Szenario erstellt hat.

Bereit zu gehen

Benennen Sie Knoten, die in diesem Szenario verwendet werden

Im Arbeitsbereich öffnen

Wie funktioniert es?

Lorem ipsum dolor sitzen amet, consectetur adipiscing elit. Suspendisse varius enim in eros elementum tristique. Duis Cursus, Mi Quis Viverra Ornare, Eros Dolor Interdum Nulla, Ut Commodo Diam Libero Vitae Erat. Aenean faucibus nibh und justo cursus id rutrum lorem imperdiet. Nunc ut sem vitae risus tristique posuere.

Änderungswunsch:

Geben Sie eine Nachricht ein

Schritt 1: Anwendung eins

-

Unterstützt von Latenode AI

Beim Absenden des Formulars ist ein Fehler aufgetreten. Versuchen Sie es später noch einmal.
Versuchen Sie es erneut
Inhaltsverzeichnis
LangChain Tools: Vollständige Anleitung zum Erstellen und Verwenden benutzerdefinierter LLM-Tools + Codebeispiele 2025

LangChain Tools sind Python-basierte Lösungen, die eine nahtlose Interaktion zwischen Large Language Models (LLMs) und externen Systemen wie APIs und Datenbanken ermöglichen. Durch die Unterstützung strukturierter Funktionsaufrufe ermöglichen diese Tools LLMs die Durchführung von Aufgaben wie dem Abrufen von Live-Daten, der Ausführung von Abfragen oder der Automatisierung von Workflows. Dieser Ansatz schließt die Lücke zwischen KI-Argumentation und umsetzbaren Ergebnissen und eignet sich ideal für Szenarien, die Echtzeit-Updates oder Systemintegration erfordern.

LangChain-Tools eignen sich besonders für Anwendungen wie Kundenservice-Bots, technische Supportsysteme und Finanzassistenten. Entwickler können zwischen zwei Erstellungsmethoden wählen: @tool Dekorateur für Einfachheit oder die BaseTool Unterklasse für erweiterte Anpassung. Beide Ansätze legen Wert auf klare Funktionssignaturen, robuste Fehlerbehandlung und präzise Eingabevalidierung, um Zuverlässigkeit zu gewährleisten.

Für Teams, die eine Alternative zur kundenspezifischen Entwicklung suchen, bieten Plattformen wie Latenknoten Vereinfachen Sie die Automatisierung mit vorgefertigten Konnektoren und visuellen Workflows. Beispielsweise können Sie Tools wie Notion, WhatsAppoder Google Blätter wird unkompliziert, ohne dass umfangreiches Programmieren erforderlich ist. Dies reduziert Entwicklungszeit und Wartungsaufwand, sodass sich die Teams auf die Bereitstellung wirkungsvoller Lösungen konzentrieren können.

Erstellen einer benutzerdefinierten Langkette Werkzeug

Langkette

Architektur und Kernkomponenten des LangChain-Tools

LangChain-Tools basieren auf einem strukturierten Framework, das Python-Funktionen in Schnittstellen transformiert, die von großen Sprachmodellen (LLMs) aufgerufen werden können. Dieses Setup ermöglicht eine reibungslose Interaktion zwischen KI-Modellen und externen Systemen. Im Folgenden untersuchen wir die Kernkomponenten, die den effektiven Betrieb dieser Tools ermöglichen.

Schlüsselkomponenten der LangChain-Tools

Die Funktionalität der LangChain-Tools wird durch fünf kritische Komponenten unterstützt. Jede spielt eine besondere Rolle bei der Gewährleistung eines reibungslosen Betriebs und einer zuverlässigen Kommunikation zwischen Tools und LLMs:

  • Funktionssignatur
    Die Funktionssignatur dient als Vorlage für die Benutzeroberfläche des Tools. Sie definiert die vom Tool akzeptierten Parameter und seine Ausgabe. Python-Typhinweise spielen eine entscheidende Rolle bei der Gestaltung dieser Signatur und helfen LLMs, das richtige Tool für bestimmte Aufgaben zu finden. Der Funktionsname dient als eindeutige Kennung und leitet das LLM bei der Ausführung zur Auswahl des richtigen Tools.
  • Werkzeugbeschreibungen
    Werkzeugbeschreibungen vermitteln LLMs ein klares Verständnis der Funktion eines Werkzeugs. In einfacher, natürlicher Sprache verfasst, erklären sie, wann und wie das Werkzeug eingesetzt werden sollte und welche Ergebnisse es liefert. Klare und prägnante Beschreibungen sind unerlässlich, um Missverständnisse oder falsche Werkzeuganwendung zu vermeiden.
  • Parameterschema
    Das Parameterschema definiert die Regeln und die Struktur für die Eingabe des Tools. Mithilfe von Python-Typhinweisen können Entwickler Schemata erstellen, die Einschränkungen, Standardwerte und benutzerdefinierte Validierungsregeln festlegen. Dadurch wird sichergestellt, dass die Eingabedaten korrekt formatiert sind, was das Risiko von Laufzeitfehlern reduziert und die allgemeine Zuverlässigkeit verbessert.
  • Retourenabwicklung
    Diese Komponente bestimmt, wie die Ausgabe eines Tools verarbeitet und an das LLM zurückgegeben wird. Tools können Ausgaben in verschiedenen Formen erzeugen, z. B. als Klartext, strukturierte Daten oder komplexe Objekte. Das Rückgabeformat ist entscheidend, um sicherzustellen, dass das LLM die Ergebnisse effektiv in seinem Workflow nutzen kann.
  • Fehlerbehandlung
    Die Fehlerbehandlung ist ein wichtiger, aber oft übersehener Aspekt. Tools müssen in der Lage sein, Probleme wie Netzwerkausfälle, Ratenbegrenzungen oder ungültige Eingaben zu bewältigen. Effektives Fehlermanagement stellt sicher, dass ein einzelner Fehler nicht den Betrieb eines gesamten Agenten stört, was besonders in Produktionsumgebungen wichtig ist.

Decorator vs. BaseTool-Unterklassenmethoden

LangChain bietet zwei Hauptansätze zur Erstellung von Tools, die jeweils auf unterschiedliche Komplexitätsstufen und Anwendungsfälle zugeschnitten sind. Diese Methoden sind die @tool Dekorateur und der BaseTool Unterklasse.

  • Der @tool Dekorateur
    Dieser Ansatz ist auf Einfachheit ausgelegt. Er ermöglicht Entwicklern die schnelle Konvertierung von Python-Funktionen in LangChain-kompatible Tools mit minimalem Aufwand. Der Dekorator übernimmt Aufgaben wie Schemagenerierung, Parametervalidierung und grundlegende Fehlerumbrüche automatisch. Er eignet sich ideal für einfache Operationen wie API-Aufrufe, einfache Berechnungen oder Datentransformationen, bei denen keine Zustandsverwaltung oder komplexe Initialisierung erforderlich ist.
  • Der BaseTool Unterklasse
    Für erweiterte Anforderungen BaseTool Die Subclass-Methode bietet umfangreiche Anpassungsmöglichkeiten. Sie eignet sich für Tools, die komplexe Logik, zustandsbehaftete Operationen oder erweiterte Fehlerbehandlung erfordern. Entwickler können benutzerdefinierte Initialisierung, asynchrone Operationen und komplexere Rückgabetypen implementieren. Diese Methode erfordert zwar mehr Programmierung, bietet aber die nötige Flexibilität für Tools auf Produktionsebene, insbesondere für Tools mit Authentifizierung, persistenten Verbindungen oder detaillierter Geschäftslogik.

Die Wahl des richtigen Ansatzes

Die Wahl zwischen diesen Methoden hängt von der Komplexität des Tools und dem Verwendungszweck ab. Einfache Tools beginnen oft mit dem Dekorator-Ansatz und können später, wenn die Anforderungen steigen, zu unterklassenbasierten Implementierungen weiterentwickelt werden. Für Tools, die jedoch eine robuste Fehlerbehandlung oder die Integration in komplexe Systeme erfordern, beginnend mit dem BaseTool Unterklassen können Zeit sparen und spätere architektonische Herausforderungen vermeiden.

Erstellen benutzerdefinierter Tools mit ordnungsgemäßer Validierung

Beim Erstellen benutzerdefinierter Tools ist es wichtig, auf eine strenge Eingabevalidierung, eine effektive Fehlerbehandlung und eine klare Dokumentation zu achten. Diese Elemente gewährleisten eine zuverlässige Leistung der Tools und eine nahtlose Integration in große Sprachmodelle (LLMs).

So erstellen Sie ein Basiswerkzeug

Der @tool Decorator bietet eine einfache Methode zum Erstellen von LangChain-Tools. Es generiert automatisch Schemata und übernimmt die grundlegende Validierung und ist somit ideal für einfache Vorgänge.

Hier ist ein Beispiel für ein Tool zur Wettersuche:

from langchain.tools import tool
from typing import Optional
import requests

@tool
def get_weather_data(city: str, country_code: Optional[str] = "US") -> str:
    """
    Fetch current weather information for a specified city.

    Args:
        city: The name of the city to get weather for.
        country_code: Two-letter country code (default: US).

    Returns:
        Weather information as a formatted string.
    """
    try:
        api_key = "your_api_key_here"
        url = "http://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": f"{city},{country_code}",
            "appid": api_key,
            "units": "imperial"
        }
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        temp = data["main"]["temp"]
        description = data["weather"][0]["description"]
        return f"Current weather in {city}: {temp}°F, {description}"
    except requests.exceptions.RequestException as e:
        return f"Error fetching weather data: {str(e)}"
    except KeyError as e:
        return f"Invalid response format: missing {str(e)}"

Für komplexere Szenarien, wie etwa solche, die eine benutzerdefinierte Initialisierung oder die Handhabung interner Zustände erfordern, ist die BaseTool Unterklasse bietet mehr Flexibilität:

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")
    table: str = Field(description="Target table name")

class DatabaseQueryTool(BaseTool):
    name = "database_query"
    description = "Execute SQL queries against the company database"
    args_schema: Type[BaseModel] = DatabaseQueryInput

    def __init__(self, connection_string: str):
        super().__init__()
        self.connection_string = connection_string
        self.connection = None

    def _run(self, query: str, table: str) -> str:
        if not self.connection:
            self.connection = self._establish_connection()
        # Execute query with proper validation
        return self._execute_safe_query(query, table)

Best Practices für Werkzeugnamen und -beschreibungen

Die Wahl klarer und beschreibender Namen für Werkzeuge hilft LLMs, deren Zweck und Verwendung zu verstehen. Verwenden Sie handlungsorientierte Verben in Werkzeugnamen (z. B. search_documents statt docs) und vermeiden Sie Abkürzungen, die das LLM verwirren könnten. Konsistenz zwischen verwandten Tools ist ebenso wichtig; zum Beispiel die Benennung mehrerer API-Tools als api_get_user, api_create_user und api_delete_user erstellt eine logische Gruppierung.

Beschreibungen sollten prägnant und im Aktiv verfasst sein und den Zweck des Tools, die erforderlichen Eingaben und die erwarteten Ergebnisse klar umreißen. Vergleichen Sie diese beiden Beispiele:

# Poor description
@tool
def calc(x: float, y: float) -> float:
    """Does math stuff"""
    return x + y

# Effective description
@tool
def add_numbers(first_number: float, second_number: float) -> float:
    """
    Add two numbers together and return the sum.

    Use this tool when you need to perform basic addition of numeric values.
    Both inputs must be numbers (integers or decimals).

    Args:
        first_number: The first number to add.
        second_number: The second number to add.

    Returns:
        The sum of the two input numbers.
    """
    return first_number + second_number

Parametertypisierung und Eingabevalidierung

Die genaue Parametertypisierung ist entscheidend, um Laufzeitprobleme zu vermeiden und LLM-Interaktionen zu steuern. Die Typhinweise von Python und die Pydantic-Modelle arbeiten gut zusammen, um die Validierung zu erzwingen.

Grundlegende Typvalidierung Beispiel:

from typing import List, Dict, Optional, Union
from datetime import datetime
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@tool
def create_task(
    title: str,
    description: Optional[str] = None,
    priority: Priority = Priority.MEDIUM,
    due_date: Optional[datetime] = None,
    tags: List[str] = []
) -> Dict[str, Union[str, int]]:
    """
    Create a new task in the project management system.

    Args:
        title: Task title (required, max 100 characters).
        description: Detailed task description (optional).
        priority: Task priority level (low, medium, high).
        due_date: When the task should be completed (ISO format).
        tags: List of tags to categorize the task.

    Returns:
        Dictionary containing a task ID and a confirmation message.
    """
    if len(title) > 100:
        raise ValueError("Title must be 100 characters or less")
    if due_date and due_date < datetime.now():
        raise ValueError("Due date cannot be in the past")
    task_id = generate_task_id()
    return {
        "task_id": task_id,
        "message": f"Task '{title}' created successfully"
    }

Erweiterte Validierung unter Verwendung von Pydantic-Modellen:

from pydantic import BaseModel, Field, validator
from typing import List
import re

class EmailInput(BaseModel):
    recipients: List[str] = Field(description="List of email addresses")
    subject: str = Field(description="Email subject line", max_length=200)
    body: str = Field(description="Email body content")

    @validator('recipients')
    def validate_emails(cls, v):
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        for email in v:
            if not re.match(email_pattern, email):
                raise ValueError(f"Invalid email address: {email}")
        return v

    @validator('subject')
    def validate_subject(cls, v):
        if not v.strip():
            raise ValueError("Subject cannot be empty")
        return v.strip()

@tool
def send_email(email_data: EmailInput) -> str:
    """
    Send an email to specified recipients with validation.

    All email addresses are validated before sending.
    The subject line is required and cannot be empty.
    """
    # Send validated email
    return f"Email sent to {len(email_data.recipients)} recipients"

Fehlerbehandlung und Ausnahmeverwaltung

Sobald die Eingaben validiert sind, ist eine robuste Fehlerbehandlung entscheidend, um sicherzustellen, dass die Arbeitsabläufe auch bei auftretenden Problemen intakt bleiben. Eine gut konzipierte Fehlerbehandlung verhindert, dass ein einzelner Fehler den gesamten Prozess unterbricht, und liefert hilfreiches Feedback für die Fehlerbehebung.

Hier ist ein Beispiel für einen Dekorator zur Standardisierung der Fehlerbehandlung über verschiedene Tools hinweg:

import logging
from functools import wraps
import requests

def handle_tool_errors(func):
    """Decorator to standardize error handling across tools."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.Timeout as e:
            logging.error(f"Timeout occurred: {e}")
            return "Request timed out"
        except Exception as e:
            logging.error(f"An error occurred: {e}")
            return f"Error: {e}"
    return wrapper
sbb-itb-23997f1

Integrieren von Tools mit Agenten und Workflows

Die Integration von LangChain-Tools mit Agenten umfasst die Auswahl der richtigen Tools und die Gewährleistung einer reibungslosen Ausführung der Aufgaben.

Verbinden von Tools mit Agenten

Hier ist ein Beispiel für die Einrichtung eines Agenten mit mehreren Tools, die auf ein Kundendienstszenario zugeschnitten sind:

from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
from langchain.tools import tool
import requests
from datetime import datetime

@tool
def lookup_order_status(order_id: str) -> str:
    """
    Retrieve the current status of a customer order using its ID.

    Args:
        order_id: The unique identifier for the order (e.g., ORD-12345).

    Returns:
        Information about the order status, including shipping details.
    """
    # Simulated API call
    api_response = requests.get(f"https://api.company.com/orders/{order_id}")
    if api_response.status_code == 200:
        data = api_response.json()
        return f"Order {order_id}: {data['status']} - Expected delivery: {data['delivery_date']}"
    return f"Order {order_id} not found in system"

@tool
def process_refund_request(order_id: str, reason: str) -> str:
    """
    Handle a customer refund request.

    Args:
        order_id: The order ID for which the refund is requested.
        reason: The reason provided by the customer for requesting the refund.

    Returns:
        Confirmation of the refund request along with a reference number.
    """
    refund_id = f"REF-{datetime.now().strftime('%Y%m%d')}-{order_id[-5:]}"
    return f"Refund initiated for {order_id}. Reference: {refund_id}. Processing time: 3-5 business days."

# Initialize the agent with tools
llm = OpenAI(temperature=0)
tools = [lookup_order_status, process_refund_request]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    max_iterations=3
)

# The agent selects the appropriate tool based on user input
response = agent.run("I need to check order ORD-67890 and request a refund because the item arrived damaged")

In diesem Beispiel verwendet der Agent die Eingabeabfrage, um zu bestimmen, welche Tools aktiviert werden sollen. Wenn mehrere Tools verfügbar sind, kann deren Organisation in spezialisierten Gruppen sowohl die Genauigkeit als auch die Effizienz verbessern. Beispielsweise können Tools nach ihrer Funktion gruppiert werden, z. B. nach auftrags- oder produktbezogenen Aufgaben:

from langchain.agents import AgentExecutor
from langchain.tools import BaseTool

class CustomerServiceAgent:
    def __init__(self):
        self.order_tools = [lookup_order_status, process_refund_request, additional_order_tool]
        self.product_tools = [additional_product_tool, additional_product_tool_2]

    def route_to_specialist(self, query: str) -> AgentExecutor:
        if "order" in query.lower() or "refund" in query.lower():
            return initialize_agent(self.order_tools, llm, AgentType.ZERO_SHOT_REACT_DESCRIPTION)
        elif "product" in query.lower() or "inventory" in query.lower():
            return initialize_agent(self.product_tools, llm, AgentType.ZERO_SHOT_REACT_DESCRIPTION)
        else:
            # Default to general tools
            return initialize_agent(self.order_tools[:2], llm, AgentType.ZERO_SHOT_REACT_DESCRIPTION)

Diese Methode stellt sicher, dass Abfragen an die relevantesten Tools weitergeleitet werden, wodurch eine optimierte Benutzererfahrung entsteht.

Erweiterte Tool-Muster für komplexe Arbeitsabläufe

Bei Workflows, die mehrere unabhängige API-Aufrufe oder Datenbankabfragen erfordern, kann die asynchrone Ausführung die Effizienz erheblich steigern. Anstatt Aufgaben nacheinander abzuarbeiten, ermöglichen asynchrone Muster die parallele Ausführung von Aufgaben:

import asyncio
from langchain.tools import tool
import aiohttp
from typing import List

@tool
async def fetch_user_data_async(user_id: str) -> str:
    """
    Retrieve a user's profile asynchronously.

    Args:
        user_id: The unique identifier for the user.

    Returns:
        Profile details as a JSON string.
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.userservice.com/users/{user_id}") as response:
            if response.status == 200:
                data = await response.json()
                return f"User {user_id}: {data['name']}, {data['email']}, {data['subscription_tier']}"
            return f"User {user_id} not found"

@tool
async def fetch_usage_metrics_async(user_id: str) -> str:
    """
    Obtain usage statistics for a user asynchronously.

    Args:
        user_id: The identifier for the user.

    Returns:
        Usage details, including API call count and storage usage.
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.analytics.com/usage/{user_id}") as response:
            if response.status == 200:
                data = await response.json()
                return f"Usage for {user_id}: {data['api_calls']} calls, {data['storage_gb']}GB storage"
            return f"No usage data for {user_id}"

async def parallel_user_analysis(user_ids: List[str]) -> List[str]:
    """Execute multiple asynchronous tasks for user data retrieval."""
    tasks = []
    for user_id in user_ids:
        tasks.append(fetch_user_data_async(user_id))
        tasks.append(fetch_usage_metrics_async(user_id))
    results = await asyncio.gather(*tasks)
    return results

Dieser Ansatz spart nicht nur Zeit, sondern stellt auch sicher, dass das System komplexe Arbeitsabläufe effizient bewältigen kann.

Für Aufgaben, die Kontext über mehrere Interaktionen hinweg erfordern, können zustandsorientierte Tools verwendet werden. Diese Tools speichern Informationen und ermöglichen so eine kumulative Analyse und ein besseres Tracking:

from typing import Dict, Any
import json
from datetime import datetime
from langchain.tools import BaseTool

class StatefulAnalyticsTool(BaseTool):
    name = "analytics_tracker"
    description = "Track and analyze user behavior patterns across multiple interactions"

    def __init__(self):
        super().__init__()
        self.session_data: Dict[str, Any] = {}
        self.interaction_count = 0

    def _run(self, action: str, data: str) -> str:
        self.interaction_count += 1

        if action == "track_event":
            event_data = json.loads(data)
            event_type = event_data.get("type")
            if event_type not in self.session_data:
                self.session_data[event_type] = []
            self.session_data[event_type].append({
                "timestamp": datetime.now().isoformat(),
                "data": event_data,
                "interaction_number": self.interaction_count
            })
            return f"Tracked {event_type} event. Total interactions: {self.interaction_count}"

        elif action == "analyze_patterns":
            if not self.session_data:
                return "No data collected yet for analysis"
            patterns = {}
            for event_type, events in self.session_data.items():
                patterns[event_type] = {
                    "count": len(events),
                    "frequency": len(events) / self.interaction_count
                }
            return f"Behavior patterns: {json.dumps(patterns, indent=2)}"

        return "Unknown action. Use 'track_event' or 'analyze_patterns'"

Produktionsüberlegungen für LangChain-Tools

Die Überführung von LangChain-Tools von der Entwicklung in die Produktion erfordert einen durchdachten Ansatz, um Herausforderungen in Bezug auf Leistung, Sicherheit und Wartung zu bewältigen. Diese Überlegungen sind entscheidend, um sicherzustellen, dass die Tools in realen Umgebungen effizient und sicher funktionieren.

Optimierung der Werkzeugleistung

Leistungsengpässe in Produktionsumgebungen entstehen oft durch langsame externe API-Antworten, ineffiziente Logik oder übermäßige synchrone Vorgänge [1][2]Diese Probleme treten besonders dann auf, wenn die Tools eine große Anzahl gleichzeitiger Anfragen verarbeiten oder mit APIs interagieren, die Ratenbegrenzungen erzwingen.

Eine Möglichkeit zur Leistungssteigerung besteht darin, asynchrone Ausführungsmuster, wodurch Tools mehrere Anfragen gleichzeitig verarbeiten können. Dieser Ansatz ist besonders effektiv für E/A-gebundene Operationen, wie das folgende Beispiel zeigt:

import asyncio
import aiohttp
from langchain.tools import StructuredTool
from pydantic import BaseModel

class BatchAPITool(BaseModel):
    """Optimized tool for handling multiple API requests concurrently."""

    async def fetch_data_batch(self, endpoints: list[str]) -> dict:
        """Process multiple API endpoints concurrently with error handling."""
        async with aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=10),
            connector=aiohttp.TCPConnector(limit=20)
        ) as session:
            tasks = [self._fetch_single(session, url) for url in endpoints]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return {
                "successful": [r for r in results if not isinstance(r, Exception)],
                "failed": [str(r) for r in results if isinstance(r, Exception)],
                "total_processed": len(results)
            }

    async def _fetch_single(self, session: aiohttp.ClientSession, url: str) -> dict:
        try:
            async with session.get(url) as response:
                if response.status == 200:
                    return await response.json()
                return {"error": f"HTTP {response.status}"}
        except asyncio.TimeoutError:
            return {"error": "Request timeout"}
        except Exception as e:
            return {"error": f"Request failed: {str(e)}"}

# Create the structured tool with async support
batch_tool = StructuredTool.from_function(
    func=BatchAPITool().fetch_data_batch,
    name="batch_api_processor",
    description="Process multiple API endpoints concurrently for improved performance"
)

Zusätzlich zur asynchronen Ausführung Ratenbegrenzung und Caching sind wesentliche Bestandteile einer robusten Produktionsstrategie. Durch Ratenbegrenzung wird verhindert, dass Tools API-Kontingente überschreiten, während Caching die Häufigkeit von API-Aufrufen reduziert, indem Antworten für eine bestimmte Zeit gespeichert werden.

Ratenbegrenzung mit exponentiellem Backoff

Um ratenbegrenzte APIs reibungslos zu handhaben, ist die Implementierung exponentieller Backoff-Strategien wichtig. Das folgende Beispiel zeigt einen Dekorator, der Anfragen mit zunehmender Verzögerung wiederholt:

import random
from functools import wraps
import asyncio

def rate_limited_retry(max_retries=3, base_delay=1.0):
    """Decorator for handling rate limits with exponential backoff."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if "rate limit" in str(e).lower() and attempt < max_retries:
                        delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                        await asyncio.sleep(delay)
                        continue
                    raise e
            return {"error": "Max retries exceeded"}
        return wrapper
    return decorator

Caching-Strategien

Durch Caching kann die Reaktionsfähigkeit deutlich verbessert und die Anzahl der API-Aufrufe reduziert werden, indem häufig abgerufene Daten gespeichert werden. Hier ist ein Beispiel für ein einfaches Caching-System:

from datetime import datetime, timedelta
from typing import Dict, Any, Optional

class ToolCache:
    def __init__(self, default_ttl_minutes: int = 15):
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.default_ttl = timedelta(minutes=default_ttl_minutes)

    def get(self, key: str) -> Optional[Any]:
        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() < entry["expires"]:
                return entry["data"]
            else:
                del self.cache[key]
        return None

    def set(self, key: str, data: Any, ttl_minutes: Optional[int] = None) -> None:
        ttl = timedelta(minutes=ttl_minutes) if ttl_minutes else self.default_ttl
        self.cache[key] = {
            "data": data,
            "expires": datetime.now() + ttl
        }

# Global cache instance
tool_cache = ToolCache()

Durch die Kombination dieser Techniken – asynchrone Ausführung, Ratenbegrenzung und Caching – können LangChain-Tools eine reibungslosere und zuverlässigere Leistung in Produktionsumgebungen erzielen.

Sicherheit in der Tool-Entwicklung

Während die Leistung im Mittelpunkt steht, ist die Sicherheit beim Einsatz von LangChain-Tools in der Produktion ebenso wichtig. Tools greifen häufig mit erhöhten Berechtigungen auf externe Systeme zu und sind dadurch anfällig für Risiken wie unzureichende Eingabevalidierung und zu freizügige Zugriffe.

Eingabevalidierung mit Pydantik Models

Pydantic-Modelle bieten eine robuste Grundlage für die Validierung von Eingaben. Durch die Durchsetzung strenger Regeln verhindern sie, dass schädliche oder ungültige Daten das System gefährden. Hier ist ein Beispiel für ein sicheres Eingabemodell:

from pydantic import BaseModel, Field

class SecureAPIRequest(BaseModel):
    """Secure input model with comprehensive validation."""
    endpoint: str = Field(..., regex=r'^[a-zA-Z0-9/_-]+$', max_length=200)
    user_id: str = Field(..., regex=r'^[a-zA-Z0-9-]+$')
    # Additional fields and validators can be added here to further secure inputs

Sichern von API-Tools

Das folgende Beispiel zeigt ein sicheres API-Tool, das Eingabevalidierung und zusätzliche Sicherheitsüberprüfungen umfasst:

def create_secure_api_tool():
    """Create API tool with built-in security validation."""

    def secure_api_call(request: SecureAPIRequest) -> str:
        # Additional security checks
        if not _is_authorized_user(request.user_id):
            return "Error: Unauthorized access attempt"

        if not _is_allowed_endpoint(request.endpoint):
            return "Error: Endpoint not permitted"

        try:
            # Perform the actual API call with validated inputs
            result = _execute_api_request(request)
            return _sanitize_response(result)
        except Exception:
            # Never expose internal error details
            return "Error: Request processing failed"

    return StructuredTool.from_function(
        func=secure_api_call,
        name="secure_api_tool",
        description="Execute API requests with comprehensive security validation"
    )

Bereinigen von API-Antworten

Um zu verhindern, dass vertrauliche Informationen offengelegt werden, sollten Antworten vor der Rücksendung bereinigt werden:

def _sanitize_response(response: dict) -> str:
    """Remove sensitive information from API responses."""
    sensitive_keys = ['password', 'token', 'secret', 'key', 'credential']

    def clean_dict(obj):
        if isinstance(obj, dict):
            return {k: clean_dict(v) for k, v in obj.items() 
                    if k.lower() not in sensitive_keys}
        elif isinstance(obj, list):
            return [clean_dict(item) for item in obj]
        return obj

    cleaned = clean_dict(response)
    return str(cleaned)[:1000]  # Limit response size

Berechtigungsverwaltung

Das Berechtigungsmanagement sollte dem Prinzip der geringsten Privilegien folgen und sicherstellen, dass Tools nur auf die für ihre Aufgaben erforderlichen Ressourcen zugreifen können. Rollenbasierte Zugriffskontrollen können unbefugte Aktionen weiter einschränken und so die allgemeine Sicherheit erhöhen. [1][3].

LatenknotenVisueller Integrationsansatz

Latenknoten

LangChain-Tools erfordern oft manuelle Programmierung und laufende Wartung, was zeit- und ressourcenintensiv sein kann. Latenode hingegen vereinfacht diesen Prozess mit seinen vorgefertigten Konnektoren zu Hunderten von Diensten und macht so bei den meisten gängigen Integrationen die individuelle Programmierung überflüssig.

Latenode vs. manuelle LangChain-Entwicklung

Der Unterschied zwischen der Erstellung benutzerdefinierter LangChain-Tools und der Nutzung des visuellen Workflow-Designs von Latenode ist eklatant, insbesondere hinsichtlich Entwicklungszeit und Komplexität. Beispielsweise erfordert die Entwicklung eines benutzerdefinierten Tools für Google Sheets in der Regel umfangreiches Coding für Aufgaben wie Fehlerbehandlung, Authentifizierung und Datenvalidierung. Mit Latenode lässt sich die gleiche Funktionalität über eine intuitive Drag-and-Drop-Oberfläche erreichen.

Stellen Sie sich einen Workflow vor, der Kundenfeedback verarbeitet und eine Tabelle aktualisiert. Normalerweise wären dafür separate Tools für die Datenverarbeitung, API-Authentifizierung und Tabellenbearbeitung erforderlich. Latenode vereinfacht dies in einer visuellen Sequenz wie: Webhook → OpenAI GPT-4 → Google Tabellen. Jeder Konnektor in dieser Kette verfügt über integrierte Authentifizierung und Fehlerverwaltung, wodurch ein Großteil des manuellen Aufwands entfällt.

Latenode unterstützt die Integration mit über 300 Apps und mehr als 200 KI-Modellen und deckt damit ein breites Spektrum an Geschäftsautomatisierungsanforderungen ab, ohne dass benutzerdefinierter Code erforderlich ist. Dieser Ansatz ist besonders vorteilhaft für Teams, die zuverlässige Lösungen ohne den zusätzlichen Aufwand laufender Wartung suchen. Durch die Optimierung der Entwicklungs- und Wartungsprozesse erleichtert Latenode den Übergang zu einem effizienten Systemmanagement.

Diese Plattform beschleunigt nicht nur die Entwicklung, sondern reduziert auch die Herausforderungen im Zusammenhang mit der Wartung individuell erstellter Tools erheblich.

Reduzierung des Overheads und Verbesserung der Wartbarkeit

Die Entwicklung programmatischer Tools erfordert häufig das Debuggen komplexer Funktionsaufrufe und die Verwaltung der Parameteranalyse, was ein mühsamer und fehleranfälliger Prozess sein kann. Latenode beseitigt diese Hürden, indem es durch sein visuelles Workflow-Design die Interaktion zwischen LLM und System ermöglicht und Integrationen auch für Nicht-Entwickler zugänglich macht.

Funktionen wie der integrierte Ausführungsverlauf und die Wiederholung von Szenarien ermöglichen es Teams, Probleme schnell zu diagnostizieren und zu beheben, ohne ganze Workflows neu erstellen zu müssen. Dadurch entfällt ein Großteil des Rätselratens, das normalerweise mit dem Debuggen benutzerdefinierter Tools verbunden ist.

Darüber hinaus erhöht Latenodes AI Code Copilot die Flexibilität, indem es Teams ermöglicht, JavaScript direkt in Workflows zu generieren und zu bearbeiten. Diese Funktion schließt die Lücke zwischen visuellem Design und benutzerdefinierter Logik und ermöglicht es Teams, maßgeschneiderte Funktionen hinzuzufügen, ohne zwischen verschiedenen Entwicklungsumgebungen wechseln zu müssen. Diese nahtlose Integration hilft Teams, sich auf die Entwicklung schlanker und effektiver Automatisierungslösungen zu konzentrieren.

Die Wartung wird weiter vereinfacht, da Latenode API-Änderungen, Authentifizierungsupdates und die Zuverlässigkeit des Connectors automatisch übernimmt. Dies erspart den Teams die laufende Überwachung externer API-Updates und die Überarbeitung benutzerdefinierter Implementierungen und reduziert so den langfristigen Aufwand.

Vorgefertigte Konnektoren für schnellere Integrationen

Die vorgefertigten Konnektoren von Latenode bieten Teams eine schnellere und effizientere Möglichkeit, externe Systeme zu integrieren und gleichzeitig den Wartungsaufwand zu minimieren. Diese Integrationen sind so konzipiert, dass sie die Fehlerbehandlung und API-Updates automatisch verwalten und so wertvolle Zeit und Mühe sparen.

Die Connector-Bibliothek der Plattform umfasst beliebte Business-Tools wie Notion, StripeWhatsApp Telegramund LinkedIn. Jeder Connector verfügt über eine vorkonfigurierte Authentifizierung und unterstützt gängige Anwendungsfälle, sodass auch bei der Weiterentwicklung der APIs ein reibungsloser Betrieb gewährleistet ist.

Für die Automatisierung persönlicher Messenger geht Latenode über Standard-API-Integrationen hinaus. Es ermöglicht die Automatisierung für Plattformen wie WhatsApp, LinkedIn und Telegram und ermöglicht so personalisierte Outreach- und CRM-ähnliche Workflows. Die Implementierung von Funktionen wie benutzerdefinierten LangChain-Tools wäre aufgrund der Herausforderungen bei Authentifizierung und Compliance äußerst komplex.

Darüber hinaus ermöglicht die integrierte Datenbank von Latenode strukturiertes Datenmanagement direkt in Workflows. In Kombination mit der Headless-Browser-Automatisierung unterstützt sie komplexe Automatisierungsszenarien, die sonst mehrere benutzerdefinierte Tools und externe Dienste erfordern würden.

Für Teams, die sich zwischen Eigenentwicklung und Kauf entscheiden, erweist sich das Preismodell von Latenode – basierend auf der Ausführungszeit statt auf Einzelkosten – oft als kostengünstigere Option. Dies kann zu erheblichen Einsparungen im Vergleich zu den Entwicklungs- und Wartungskosten kundenspezifischer LangChain-Tools führen.

FAQs

Was ist der Unterschied zwischen der Verwendung des @tool-Dekorators und der Unterklassenbildung von BaseTool zum Erstellen von LangChain-Tools?

Der @tool Decorator bietet eine einfache Möglichkeit, Python-Funktionen in LangChain-Tools umzuwandeln. Durch einfaches Wrapping einer Funktion werden automatisch wichtige Attribute wie Name und Beschreibung des Tools zugewiesen, wobei diese Details aus dem Funktionsnamen und dem Docstring der Funktion abgerufen werden. Dies macht es zu einer hervorragenden Wahl für die Erstellung einfacher Tools oder zum schnellen Testen von Ideen.

Für Szenarien, die mehr Komplexität erfordern, Unterklassen BaseTool ist die bessere Option. Diese Methode bietet mehr Flexibilität und ermöglicht die Definition benutzerdefinierter Attribute, die Implementierung erweiterter Validierungen, die Verwaltung von Fehlern und die Entwicklung komplexerer Verhaltensweisen. Sie eignet sich besonders für Tools, die Produktionsanforderungen erfüllen oder komplexe Workflows verarbeiten müssen.

Im Wesentlichen ist die @tool Decorator ist perfekt für schnelle und einfache Setups, während BaseTool ist die erste Wahl für die Entwicklung fortschrittlicherer und zuverlässigerer Tools.

Wie erleichtert Latenode die Integration externer Systeme im Vergleich zur Entwicklung benutzerdefinierter Tools in LangChain?

Latenode optimiert den Prozess der Integration externer Systeme durch seine visuelles Workflow-Design, wodurch manuelles Codieren überflüssig wird. Die benutzerfreundliche Drag-and-Drop-Oberfläche ermöglicht nahtlose Verbindungen zwischen KI-Agenten, APIs und externen Diensten und umgeht die oft kniffligen Aufgaben der Parametervalidierung und Fehlerbehandlung, die mit der Entwicklung benutzerdefinierter LangChain-Tools einhergehen.

Durch die Automatisierung kritischer Aufgaben wie Fehlerbehandlung und Systemkonfiguration verkürzt Latenode nicht nur die Einrichtungszeit, sondern reduziert auch die Wahrscheinlichkeit von Programmierfehlern. Dies macht es zur perfekten Wahl für alle, die schnelle und zuverlässige Integrationen suchen, insbesondere wenn ihnen fortgeschrittene Programmierkenntnisse fehlen. Mit Latenode können sich Benutzer auf die Erstellung von KI-gestützte Lösungen anstatt Code zu beheben.

Was sind die Best Practices für den sicheren und effizienten Einsatz von LangChain-Tools in der Produktion?

Um LangChain-Tools sicher und effizient in einer Produktionsumgebung einzusetzen, konzentrieren Sie sich zunächst auf Eingabe- und AusgabevalidierungDies schützt vor Injection-Angriffen und anderen Sicherheitslücken. Stellen Sie sicher, dass die Berechtigungen so konfiguriert sind, dass nur der für die Funktionalität erforderliche Mindestzugriff gewährt wird, um potenzielle Risiken zu reduzieren. Integrieren Protokollierung und Überwachung ist ebenso wichtig, da Sie so alle auftretenden Probleme schnell erkennen und beheben können.

Ein weiterer wichtiger Schritt ist die Vorbereitung auf potenziellen Missbrauch durch die Implementierung Ausgabefilterung und -validierung um die Datengenauigkeit und -integrität zu gewährleisten. Um die Systemzuverlässigkeit zu erhöhen, integrieren Ratenbegrenzung den Verkehr zu regeln und Fehlerbehandlungsmechanismen um unerwartete Probleme zu bewältigen, ohne Unterbrechungen zu verursachen. Indem Sie diese Maßnahmen befolgen, können Sie eine sichere und stabile Umgebung für Ihre LangChain-Tools schaffen und gleichzeitig eine optimale Leistung aufrechterhalten.

Ähnliche Blog-Beiträge

Apps austauschen

Anwendung 1

Anwendung 2

Schritt 1: Wählen ein Auslöser

Schritt 2: Wähle eine Aktion

Wenn das passiert ...

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Beschreibung des Auslösers

Name des Knotens

Aktion, zum Beispiel löschen

Vielen Dank! Ihre Einreichung wurde erhalten!
Hoppla! Beim Absenden des Formulars ist ein Fehler aufgetreten.

Mach das.

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Aktion, zum Beispiel löschen

Name des Knotens

Beschreibung des Auslösers

Name des Knotens

Aktion, zum Beispiel löschen

Vielen Dank! Ihre Einreichung wurde erhalten!
Hoppla! Beim Absenden des Formulars ist ein Fehler aufgetreten.
Probieren Sie es jetzt

Keine Kreditkarte notwendig

Ohne Einschränkung

Raian
Forscher, Texter und Usecase-Interviewer
September 2, 2025
18
min lesen

Verwandte Blogs

Anwendungsfall

Unterstützt von