Jeder LLM-Anbieter verfügt über ein eigenes SDK, ein eigenes Authentifizierungsformat und eine eigene Antwortstruktur. Wenn Sie Ihre Anwendung eng um die API eines KI-Anbieters herum aufbauen, kann ein späterer Wechsel zu einem anderen eine Überarbeitung großer Teile Ihrer Integration bedeuten.
Ein Gateway löst dieses Drawback, indem es zwischen Ihrer Anwendung und den Anbietern sitzt. Ihre Anwendung ruft dieselbe Funktion immer auf die gleiche Weise auf. Das Gateway verwaltet die darunter liegenden anbieterspezifischen Particulars. Der Anbieterwechsel wird zu einem Ein-Wort-Wechsel.
In diesem Tutorial erstellen wir dieses Gateway von Grund auf. Am Ende eine Single generate() Der Anruf funktioniert identisch, unabhängig davon, ob er mit TogetherAI oder Anthropic kommuniziert. Dank der gleichen Bausteine können Sie ganz einfach jeden anderen LLM-Anbieter anschließen, den Sie benötigen, und das mit konsistenter Ausgabe und sauberer Fehlerbehandlung.
Was Sie lernen werden
Am Ende dieses Tutorials wissen Sie, wie Sie:
- Entwerfen Sie eine anbieterunabhängige Schnittstelle mithilfe einer gemeinsamen Schnittstelle
generate()Funktion - Senden Sie mithilfe von Python direkte HTTP-Anfragen an TogetherAI und Anthropic
requestsBibliothek - Behandeln Sie anbieterspezifische Authentifizierung und Header
- Analysieren und normalisieren Sie verschiedene JSON-Antwortformate in eine konsistente Ausgabe
- Schreiben Sie benutzerdefinierte Ausnahmen für saubere, lesbare Fehlermeldungen
- Strukturieren Sie ein Python-Projekt mit mehreren Dateien, um Klarheit und Wartbarkeit zu gewährleisten
Bevor Sie beginnen
Sie benötigen Python 3.8+ requests Bibliothek (pip set up requests) und aktive API-Schlüssel für TogetherAI und Anthropic. Die Ausführung des gesamten Projekts kostet in der Regel deutlich weniger als 0,20 US-Greenback für die API-Nutzung.
Sie sollten mit Python-Funktionen, Wörterbüchern und der grundlegenden Fehlerbehandlung vertraut sein. Eine gewisse Vertrautheit mit HTTP-Anfragen und JSON wird hilfreich sein. Wenn Sie mit LLM-APIs noch nicht vertraut sind, können Sie Folgendes tun: KI-Engineering-Pfad deckt die grundlegenden Fähigkeiten ab, auf denen dieses Projekt aufbaut.
Greifen Sie auf das vollständige Projekt zu Dataquest-App und die Lösungsdateien auf GitHub.
Projektstruktur
Wir werden an drei Dateien arbeiten:
constants.py: API-URLs, Versionszeichenfolgen und Standardeinstellungenllm_gateway.py: Die Gateway-Logik, einschließlich dergenerate()funktions- und anbieterspezifische Request-Handlerdemo.py: Eine Testdatei, die das Gateway in Aktion demonstriert
Diese Trennung ist beabsichtigt. Konstanten sind die Werte, die sich am wahrscheinlichsten ändern, wenn Sie Anbieter hinzufügen oder Standardeinstellungen optimieren. Wenn Sie sie in einer eigenen Datei aufbewahren, müssen Sie sich nicht durch die Anwendungslogik wühlen, um eine URL oder ein Token-Restrict zu aktualisieren.
Teil 1: Konstanten
# constants.py
TOGETHER_URL = "https://api.collectively.xyz/v1/chat/completions"
ANTHROPIC_URL = "https://api.anthropic.com/v1/messages"
ANTHROPIC_VERSION = "2023-06-01"
DEFAULT_MAX_TOKENS = 300
DEFAULT_TIMEOUT = 30DEFAULT_MAX_TOKENS = 300 Hält die API-Kosten während der Entwicklung niedrig. Jeder von Ihnen generierte Token kostet Geld, daher ist die Beschränkung der Antworten auf 300 Token ein sinnvoller Commonplace für Assessments. Aus diesem Grund benötigt Anthropic bei jeder Anfrage einen API-Versionsheader ANTHROPIC_VERSION wird hier gespeichert und nicht in der Funktion fest codiert.
Teil 2: Benutzerdefinierte Fehlerbehandlung
Bevor wir eine Gateway-Logik schreiben, richten wir eine benutzerdefinierte Ausnahmeklasse ein. Dadurch erhalten wir saubere, lesbare Fehlermeldungen anstelle der ausführlichen Tracebacks, die Python standardmäßig generiert.
# llm_gateway.py
import os
import requests
import constants
class LLMGatewayError(Exception):
"""Raised when the gateway can not return a sound response."""
crossLLMGatewayError erbt von Pythons integrierter Funktion Exception Klasse. Der cross physique bedeutet, dass kein neues Verhalten hinzugefügt wird. Alles, was wir brauchen, ist der Klassenname, der in jeder Fehlerausgabe deutlich erscheint und uns genau sagt, wo etwas schief gelaufen ist.
Teil 3: Die generate() Funktion
Dies ist die öffentliche Schnittstelle des Gateways. Es ist die einzige Funktion, die Ihre Anwendung jemals aufrufen muss.
def generate(supplier, mannequin, messages, max_tokens=constants.DEFAULT_MAX_TOKENS):
"""
Ship a chat-style request to both Collectively or Anthropic
and return a normalized response.
Returns:
{
"supplier": "...",
"mannequin": "...",
"textual content": "..."
}
"""
supplier = supplier.strip().decrease()
if supplier == "collectively":
cross
elif supplier == "anthropic":
cross
else:
elevate LLMGatewayError("Unsupported supplier. Use 'collectively' or 'anthropic'.")supplier.strip().decrease() normalisiert die Eingabe, sodass „Anthropic“, „ANTHROPIC“ und „anthropic“ alle auf die gleiche Weise funktionieren. Die Funktion sendet an einen anbieterspezifischen Handler und gibt die gleiche Wörterbuchform zurück, unabhängig davon, welcher Anbieter verwendet wurde. Diese konsistente Ausgabestruktur ist der Kernwert des Gateways.
Sie werden auch feststellen, dass es noch keine Rückgabeerklärung gibt. Das ist Absicht. Die anbieterspezifischen Funktionen sind additionally noch nicht erstellt cross Hält die Funktion syntaktisch gültig und ausführbar, während wir den Relaxation des Codes erstellen. Die return-Anweisung wird hinzugefügt, sobald tatsächlich etwas zurückgegeben werden soll.
Lerneinblick: Der
generate()Funktion ist eine klassische Abstraktionsschicht. Es verbirgt Komplexität hinter einer stabilen Schnittstelle. Ihr Anwendungscode muss nichts darüber wissen, wie TogetherAI oder Anthropic ihre Anfragen formatieren. Es ruft einfachgenerate()und empfängt{"supplier": ..., "mannequin": ..., "textual content": ...}jedes Mal.
Teil 4: Der TogetherAI-Handler
Anbieterspezifische Funktionen werden mit einem Unterstrich () vorangestellt._call_together). Dies ist eine Python-Konvention, die signalisiert, dass die Funktion nur für den internen Gebrauch bestimmt ist. Es kann technisch gesehen extern aufgerufen werden, sollte jedoch nicht direkt von außerhalb des Moduls aufgerufen werden.
def _call_together(mannequin, messages, max_tokens):
api_key = os.getenv("TOGETHER_API_KEY")
if not api_key:
elevate LLMGatewayError("Lacking TOGETHER_API_KEY.")
headers = {
"Authorization": "Bearer " + api_key,
"Content material-Sort": "software/json",
}
payload = {
"mannequin": mannequin,
"messages": messages,
"max_tokens": max_tokens,
}
response = requests.submit(
constants.TOGETHER_URL,
headers=headers,
json=payload,
timeout=constants.DEFAULT_TIMEOUT,
)
if response.status_code != 200:
elevate LLMGatewayError("Collectively API error: " + response.textual content)
information = response.json()
selections = information.get("selections", ())
return selections(0).get("message", {}).get("content material") if selections else NoneDer _call_together() Die Funktion hat vier Aufgaben. Zunächst ruft es den API-Schlüssel aus Ihren Umgebungsvariablen ab os.getenv("TOGETHER_API_KEY"). API-Schlüssel sollten niemals im Quellcode erscheinen. Wenn Sie einen Schlüssel in ein öffentliches Repository übertragen, behandeln Sie ihn als kompromittiert und rotieren Sie ihn sofort.
Zweitens erstellt und sendet es eine authentifizierte POST-Anfrage. Das Header-Wörterbuch teilt TogetherAI mit, wer wir sind und welches Format wir als Antwort erwarten. Die Nutzlast enthält den Modellnamen, die Konversationsnachrichten und das Token-Restrict.
Drittens wird geprüft, ob die Anfrage erfolgreich conflict. HTTP-Standing 200 bedeutet, dass alles wie erwartet gelaufen ist. Jeder andere Statuscode bedeutet, dass etwas schief gelaufen ist und wir eine Fehlermeldung auslösen LLMGatewayError mit dem Antworttext, sodass die Fehlermeldung informativ und nicht allgemein ist.
Schließlich wird die Antwort analysiert. TogetherAI gibt eine tief verschachtelte JSON-Struktur zurück, der Aufrufer benötigt jedoch nur den Textinhalt. Der Parsing-Schritt extrahiert genau das, additionally generate() erhält eine saubere Zeichenfolge anstelle einer rohen API-Antwort.
Lerneinblick: Die Überprüfung des API-Schlüssels vor der Anfrage ist eine bewusste Entwurfsentscheidung. Fehlt der Schlüssel, erhalten Sie sofort eine eindeutige Fehlermeldung. Ohne diese Prüfung würden Sie die Anfrage ohne Autorisierung senden, eine 401-Antwort von der API erhalten und müssten das Drawback über die HTTP-Ebene zurückverfolgen. Scheitern Sie schnell mit einer hilfreichen Nachricht, anstatt spät mit einer verwirrenden Nachricht zu scheitern.
Teil 5: Der anthropische Handler
Die API von Anthropic weist einige strukturelle Unterschiede zu der von TogetherAI auf. Am wichtigsten ist die Artwork und Weise, wie Systemmeldungen behandelt werden.
def _call_anthropic(mannequin, messages, max_tokens):
api_key = os.getenv("ANTHROPIC_API_KEY")
if not api_key:
elevate LLMGatewayError("Lacking ANTHROPIC_API_KEY.")
headers = {
"x-api-key": api_key,
"anthropic-version": constants.ANTHROPIC_VERSION,
"Content material-Sort": "software/json",
}
system_text = None
non_system_messages = ()
for msg in messages:
if msg.get("position") == "system":
system_text = msg.get("content material")
else:
non_system_messages.append(msg)
payload = {
"mannequin": mannequin,
"messages": non_system_messages,
"max_tokens": max_tokens,
}
if system_text:
payload("system") = system_text
response = requests.submit(
constants.ANTHROPIC_URL,
headers=headers,
json=payload,
timeout=constants.DEFAULT_TIMEOUT,
)
if response.status_code != 200:
elevate LLMGatewayError("Anthropic API error: " + response.textual content)
information = response.json()
content material = information.get("content material", ())
text_block = subsequent((block for block in content material if block.get("sort") == "textual content"), None)
return text_block.get("textual content") if text_block else NoneTogetherAI akzeptiert Systemnachrichten inline innerhalb des messages Liste. Anthropic erwartet sie als eigenständige High-Ebene system Feld in der Nutzlast. Die Schleife, die Systemnachrichten von Nicht-Systemnachrichten trennt, berücksichtigt diesen Unterschied. Wenn keine Systemmeldung vorhanden ist, wird die system Der Schlüssel wird in der Nutzlast einfach weggelassen.
Anthropic verwendet auch einen anderen Authentifizierungsheader (x-api-key anstatt Authorization: Bearer) und erfordert die anthropic-version Header in jeder Anfrage.
Die Antwortanalyse verwendet subsequent() mit einem Generatorausdruck, um den ersten Inhaltsblock zu finden "sort": "textual content". Die Antwort von Anthropic kann mehrere Inhaltsblöcke unterschiedlichen Typs umfassen, daher müssen wir den Textblock gezielt lokalisieren.
Teil 6: Verkabelung
Wenn beide Anbieterfunktionen abgeschlossen sind, wird die generate() Die Funktion muss sie nur aufrufen und das Ergebnis zurückgeben.
def generate(supplier, mannequin, messages, max_tokens=constants.DEFAULT_MAX_TOKENS):
supplier = supplier.strip().decrease()
if supplier == "collectively":
textual content = _call_together(mannequin, messages, max_tokens)
elif supplier == "anthropic":
textual content = _call_anthropic(mannequin, messages, max_tokens)
else:
elevate LLMGatewayError("Unsupported supplier. Use 'collectively' or 'anthropic'.")
return {
"supplier": supplier,
"mannequin": mannequin,
"textual content": textual content,
}Die Funktion ist bewusst einfach gehalten. Die ganze anbieterspezifische Komplexität lebt darin _call_together Und _call_anthropic. generate() Es muss nur noch versendet und zurückgesendet werden.
Teil 7: Testen des Gateways
Die Demodatei zeigt das Gateway in Aktion.
# demo.py
from llm_gateway import generate
messages = (
{"position": "system", "content material": "You're concise."},
{"position": "consumer", "content material": "Clarify what an API is in a single sentence."},
)
print(
generate(
supplier="collectively",
mannequin="meta-llama/Llama-3.2-3B-Instruct-Turbo",
messages=messages,
)("textual content")
)
print(
generate(
supplier="anthropic",
mannequin="claude-haiku-4-5",
messages=messages,
)("textual content")
)Zwei generate() Anrufe. Unterschiedliche Anbieter, unterschiedliche Modelle. Identischer Aufbau. Das Ausführen dieser Datei erzeugt etwa Folgendes:
An API, or Utility Programming Interface, is a algorithm and protocols
that permits totally different software program purposes to speak with one another.
An API is a algorithm that lets one piece of software program ask one other piece
of software program to do one thing for it.Gleiche Frage, zwei Anbieter, eine Schnittstelle. Das ist das Gateway, das wie vorgesehen funktioniert.
Beachten Sie, dass messages ist eine Liste von Wörterbüchern, keine einfache Zeichenfolge oder ein einzelnes Wörterbuch. Die API erwartet ein Array von Nachrichtenobjekten in einem bestimmten Format. Die Übergabe einer Zeichenfolge oder eines bloßen Wörterbuchs führt zu einem Validierungsfehler des Anbieters.
Was wir gebaut haben
Ausgehend von drei leeren Dateien haben wir ein Multi-Supplier-LLM-Gateway mit folgendem Design erstellt:
constants.py: API-Endpunkte, Versionszeichenfolgen und konfigurierbare Standardeinstellungen, isoliert von der Anwendungslogik.
llm_gateway.py: A LLMGatewayError Klasse für saubere Fehlerberichterstattung, eine öffentliche generate() Funktion, die Anfragen basierend auf Anbieter und privat weiterleitet _call_together() Und _call_anthropic() Handler, die die Authentifizierung, Anforderungsformatierung und Antwortanalyse verwalten.
demo.py: Eine Demonstration, die identisch ist generate() Anrufe funktionieren bei beiden Anbietern.
Das Gateway verbirgt alles Providerspezifische. Das Hinzufügen eines neuen Anbieters bedeutet, einen neuen zu schreiben _call_* Funktion und Hinzufügen einer elif verzweigen nach generate(). Sonst muss sich nichts ändern.
Nächste Schritte
Wiederholungslogik hinzufügen. Wenn eine Ratenbegrenzung oder ein vorübergehender Fehler dazu führt, dass eine Anfrage fehlschlägt, würden automatische Wiederholungsversuche mit exponentiellem Backoff das Gateway in der Produktion robuster machen.
Token-Nutzung normalisieren. Verschiedene Anbieter zählen Token etwas unterschiedlich. Sie könnten ein hinzufügen token_usage -Feld in das zurückgegebene Wörterbuch ein, damit Anrufer den Verbrauch unabhängig vom Anbieter konsistent verfolgen können.
Fügen Sie einen neuen Anbieter hinzu. Gemini, Grok oder jeder andere Anbieter mit einer HTTP-API können durch Schreiben eines integriert werden call_<supplier>() Funktion und Hinzufügen des entsprechenden Zweigs zu generate(). Der TogetherAI-Handler ist eine gute Vorlage für den Anfang.
Erkunden Sie den Endpunkt „Antworten“. OpenAI hat eine neuere /responses Endpunkt, der einige Anwendungsfälle anders behandelt als /chat/completions. Es lohnt sich, einen Blick darauf zu werfen, wenn Sie die Funktionen des Gateways erweitern möchten.
Ressourcen
Teilen Sie Ihr erweitertes Gateway in der Group und taggen Sie @Anna_strahl. Das Hinzufügen eines dritten Anbieters, das Implementieren einer Wiederholungslogik oder das Verbinden des Gateways mit einer größeren Anwendung sind allesamt Erweiterungen, die es wert sind, geteilt zu werden.