Jeśli budujesz backend do integracji wielu modeli, prawdopodobnie doświadczyłeś tego rodzaju załamania: kod biznesowy jest wypełniony if (vendor === 'openai') ... else if (vendor === 'claude') ..., interfejs strumieniowego czatu ma osiemset linii logiki adaptacyjnej, a dodanie nowego dostawcy wymaga modyfikacji całego kluczowego łańcucha. Problem nie leży w tobie — chodzi o to, że protokoły API różnych dużych modeli w ogóle nie zostały zaprojektowane jako jednolite.
Istotą bramy modeli jest warstwa tłumaczenia protokołów: zebranie heterogenicznych żądań/odpowiedzi od N dostawców do 1 wewnętrznego standardowego protokołu. Ten artykuł rozwija temat w trzech etapach — zasady, praktyka, porównanie — i wyjaśnia, jak naprawdę powinna być zbudowana warstwa tłumaczenia.
I. Cztery typy heterogeniczności: dlaczego bezpośrednie połączenie z wieloma dostawcami musi wymknąć się spod kontroli
Najpierw spójrzmy na rzeczywiste różnice, każda z liczbami.
Metoda uwierzytelniania: OpenAI używa Authorization: Bearer sk-xxx, długość pola nagłówka to około 51 znaków; Claude używa x-api-key; Gemini wstawia klucz do parametru zapytania ?key= w URL; Qwen i ERNIE używają Authorization, ale z różnymi algorytmami podpisu — access_token ERNIE domyślnie wygasa po 30 dniach, a po wygaśnięciu zwraca kod błędu 110. Czterech dostawców, cztery różne sposoby.
Format fragmentacji strumieniowania SSE: chunk OpenAI to data: {"choices":[{"delta":{"content":"x"}}]}, znacznik końca to data: [DONE]; typ zdarzenia Claude to event: content_block_delta z data: {"delta":{"text":"x"}}, nazwa pola zmienia się z content na text; Gemini umieszcza przyrost w candidates[0].content.parts[0].text, zagnieżdżenie głębsze o 3 poziomy. Aby warstwa biznesowa mogła sparsować to samo zdanie, trzeba napisać cztery zestawy parserów.
System kodów błędów: to jest najbardziej irytujące. OpenAI używa kodów statusu HTTP + error.type, ograniczenie szybkości to 429; Claude przy ograniczeniu szybkości zwraca 429, ale type w body to rate_limit_error; Qwen przy ograniczeniu szybkości może zwrócić kod statusu 200, ale w body z code: "Throttling"; ERNIE używa error_code: 17 do oznaczenia przekroczenia limitu. Ta sama rzecz, cztery wyrażenia.
Metryka statystyki tokenów: usage.total_tokens = prompt + completion w OpenAI, zarówno chiński, jak i angielski dzielone według BPE; Claude osobno zwraca input_tokens i output_tokens, a ziarnistość dzielenia znaków chińskich jest drobniejsza — ten sam fragment 500 chińskich znaków może dać różnicę w liczbie tokenów między dwoma dostawcami od 20% do 35%. Przy rozliczaniu według zużycia ta różnica bezpośrednio przekłada się na błąd na rachunku.
Przy bezpośrednim połączeniu z czterema dostawcami ilość kodu warstwy biznesowej rozrośnie się do 3-4 razy większej niż przy jednym dostawcy, a koszt utrzymania rośnie liniowo wraz z liczbą dostawców.
II. Projekt warstwy tłumaczenia: ujednolicony format, normalizacja strumieniowania, mapowanie kodów błędów, agregacja zużycia
Warstwa tłumaczenia musi robić cztery rzeczy i żadnej nie można pominąć.
Ujednolicony format żądania/odpowiedzi: wewnętrznie definiujemy zestaw standardowego Schematu, pola wyrównane do stylu OpenAI (ponieważ ma najszerszy ekosystem). Żądanie zewnętrzne najpierw jest konwertowane do formatu wewnętrznego, a następnie przez adapter do formatu każdego dostawcy. Wzorzec adaptera, jedna klasa na dostawcę, implementująca dwie metody buildRequest() i parseResponse().
Normalizacja fragmentacji strumieniowania: sedno polega na konwersji chunków SSE każdego dostawcy do jednolitego {delta: string, finish_reason: string|null}. Sposób: napisać maszynę stanów, która dysponuje według typu dostawcy do różnych parserów i wyprowadza jednolitą strukturę. Kluczowy punkt to obsługa granic: separator \n\n w SSE, prefiks data:, łączenie wielowierszowych danych — wszystko to musi być pochłonięte w parserze. Pułapką, na którą się natknąłem, są zdarzenia ping i message_stop w Claude — jeśli nie zostaną jawnie zignorowane, puste delty zostaną wypchnięte do frontendu, powodując migotanie UI.
Tabela mapowania kodów błędów: utrzymujemy tabelę mapowania vendorCode -> unifiedCode, ujednolicając do 6 kategorii: AUTH_FAILED, RATE_LIMITED, QUOTA_EXCEEDED, INVALID_REQUEST, UPSTREAM_ERROR, TIMEOUT. Cokolwiek zwróci upstream, warstwa tłumaczenia sprawdza tabelę i konwertuje na ujednolicony kod; warstwa biznesowa rozpoznaje tylko te 6.
Agregacja zużycia: po każdym wywołaniu metryki tokenów każdego dostawcy są ujednolicane do wewnętrznych standardowych tokenów, a przy zapisie do bazy dołączane są vendor, model, request_id do późniejszej weryfikacji i analizy kosztów. Należy tu uważać na różnice metryk — output_tokens Claude i completion_tokens OpenAI nie mogą być bezpośrednio sumowane, trzeba je znormalizować do tej samej podstawy według współczynnika przeliczeniowego.
W architekturze warstwowej zwykle dzielę na cztery warstwy: warstwa dostępu (uwierzytelnianie + ograniczanie szybkości), warstwa routingu (wybór modelu), warstwa tłumaczenia (konwersja protokołów), warstwa adaptera (wywołanie SDK dostawcy). Warstwa tłumaczenia to czysta funkcja, nie dotyka IO, co ułatwia testy jednostkowe.
III. Praktyczny kod: zmień jedną linię base_url, aby podłączyć bramę agregującą
Jeśli nie chcesz pisać warstwy tłumaczenia od zera, użycie gotowej platformy agregującej AI API jest bardziej pragmatycznym wyborem. W naszym projekcie używaliśmy token8341 — jest kompatybilna z OpenAI SDK, wystarczy zmienić jedną linię base_url, aby przełączyć, a pracę warstwy tłumaczenia wykonuje za ciebie.
from openai import OpenAI
client = OpenAI(
api_key="your-key",
base_url="https://api.token8341.com/v1" # zmień tylko tę linię
)
resp = client.chat.completions.create(
model="deepseek-v3", # można też zmienić na gpt-4o / qwen-max / claude-4-sonnet
messages=[{"role": "user", "content": "Wyjaśnij SSE"}],
stream=True
)
for chunk in resp:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="")W tym kodzie normalizacja fragmentacji SSE, mapowanie kodów błędów i agregacja zużycia są wykonane po stronie bramy, a warstwa biznesowa otrzymuje standardowy format OpenAI. Routing wielu modeli też jest zbędny — brama może automatycznie wybrać optymalny model według zadania, np. długi tekst chiński przez Qwen, zadania wnioskowania przez DeepSeek, scenariusze wrażliwe na koszty przez Doubao.
Ostrzeżenie o pułapce: po zmianie base_url nie przekazuj bezpośrednio parametrów specyficznych dla dostawcy. Na przykład logit_bias OpenAI, top_k Claude — brama niekoniecznie obsługuje je wszystkie; przekazane mogą zostać po cichu zignorowane, przez co będziesz myśleć, że działają, gdy w rzeczywistości nie. Przed wdrożeniem uruchom skrypt testów obciążeniowych, aby sprawdzić kompatybilność parametrów.
IV. Porównanie: routing wielu modeli vs wiązanie z jednym modelem
Podaję zestaw naszych zmierzonych liczb. Wiązanie z jednym modelem (bezpośrednie połączenie z OpenAI): czas integracji około 2 osobodni, przełączenie awaryjne wymaga zmiany kodu i ponownego wdrożenia, średni czas przywrócenia 15 minut, brak przestrzeni na optymalizację kosztów. Routing wielu modeli (przez bramę): czas integracji około 0,5 osobodnia, przełączenie awaryjne odbywa się automatycznie, czas przywrócenia poniżej 3 sekund, koszt można obniżyć o około 30% do 40% dzięki rozdzieleniu według zadań.
W złożoności utrzymania: kod adaptacyjny bezpośredniego połączenia z 4 dostawcami to około 2000 linii; po przejściu na bramę kod adaptacyjny warstwy biznesowej spada poniżej 200 linii, redukcja o 90%. Pod względem opóźnienia brama dodaje jedno przeskok, P99 wzrasta o około 20 do 40 ms — w porównaniu z zaoszczędzonym kosztem utrzymania i elastycznością przełączania ten narzut jest akceptowalny.
Oczywiście brama nie jest srebrną kulą. Jeśli twoja firma używa tylko jednego modelu i nigdy go nie zmieni, bezpośrednie połączenie jest prostsze. Ale gdy tylko w grę wchodzi wiele modeli, wielu dostawców, optymalizacja kosztów, warstwa tłumaczenia jest niezbędną infrastrukturą. Brama modeli, API dużych modeli, agregacja AI API — te trzy rzeczy to w istocie trzy sposoby nazwania tego samego problemu na tej samej warstwie.
Na koniec jedna uwaga: pokrycie testami jednostkowymi warstwy tłumaczenia musi być maksymalne, zwłaszcza przypadki graniczne fragmentacji strumieniowania — to obszar o wysokiej częstotliwości incydentów produkcyjnych.
Autor: Liu Zhiyuan
Data publikacji: 28 października 2026