Utwórz wielomodelowych agentów CrewAI za pomocą CometAPI:

Budowa systemu wieloagentowego z CrewAI staje się ciekawsza, gdy różne agenty mogą używać różnych modeli.

Badacz może skorzystać z szybkiego, ekonomicznego modelu, analityk może potrzebować modelu z silniejszym rozumowaniem, a autor może wymagać modelu zoptymalizowanego pod wysokiej jakości generowanie długich treści. Tradycyjnie, podłączanie tych agentów do różnych dostawców oznacza zarządzanie oddzielnymi poświadczeniami API, punktami końcowymi, SDK, systemami rozliczeń oraz konfiguracją specyficzną dla dostawcy.

Czystsza architektura polega na tym, aby CrewAI zarządzał agentami i przepłykiem pracy, a CometAPI zarządzał dostępem do modeli.

CometAPI udostępnia punkt końcowy kompatybilny z OpenAI pod https://api.cometapi.com/v1, dzięki czemu aplikacje mogą kierować żądania do modeli od wielu dostawców przez wspólny interfejs API. Aktualna dokumentacja szybkiego startu wspiera także użycie standardowego OpenAI Python SDK poprzez zmianę klucza API i bazowego URL.

W tym samouczku zbudujesz przepływ pracy CrewAI z trzema agentami:

• Gemini 3.7 Flash dla badań
• Claude Opus 5 dla analizy
• GPT-5.6 dla finalnego pisania
• Jeden klucz API CometAPI
• Jeden bazowy URL API
• Konfiguracja modelu per agent
• Ograniczony fallback dla przejściowych awarii
• Checkpointing CrewAI dla odtwarzania w produkcji
• Śledzenie użycia tokenów i wykonania
• Walidacja modelu po stronie serwera

Ważna granica architektoniczna jest prosta:

CrewAI obsługuje orkiestrację agentów. CometAPI obsługuje dostęp do modeli. Identyfikatory modeli definiują routing.

---

Czym jest trasowanie modeli w wieloagentowym CrewAI?

CrewAI to framework w Pythonie do tworzenia agentów, zadań, ekip (crews) i wieloagentowych przepływów pracy. Każdy agent może mieć własną konfigurację LLM, podczas gdy Crew koordynuje, jak agenty wykonują zadania i wymieniają kontekst.

Aktualna konfiguracja LLM w CrewAI wspiera jawne ustawienia model, api_key i base_url, w tym niestandardowe punkty końcowe kompatybilne z OpenAI.

To sprawia, że architektura wielomodelowa jest prosta:

                         CometAPI                            │              https://api.cometapi.com/v1                            │        ┌───────────────────┼───────────────────┐        │                   │                   │   Researcher            Analyst             Writer        │                   │                   │ Gemini 3.7 Flash      Claude Opus 5         GPT-5.6

Agenty pozostają logicznie odseparowane, ale dostęp do modeli jest scentralizowany.

To nie oznacza, że wszystkie modele są wymienne. API kompatybilne z OpenAI zapewnia wspólny interfejs żądania; nie gwarantuje identycznych limitów kontekstu, wsparcia narzędzi, sterowania rozumowaniem, zachowania wyjścia, opóźnienia ani cen.

To rozróżnienie ma znaczenie przy projektowaniu routingu produkcyjnego.

---

Dlaczego używać CometAPI z CrewAI?

Główną zaletą nie jest to, że CrewAI nagle staje się frameworkiem wielodostawcowym. CrewAI już wspiera wielu dostawców LLM.

Zaletą jest to, że dostęp do modeli można skonsolidować za jedną warstwą API.

Bez ujednoliconej warstwy API przepływ pracy z trzema agentami może wyglądać tak:

Z CometAPI:

Aktualna dokumentacja szybkiego startu CometAPI opisuje jego punkt końcowy jako drop-in replacement dla bazowego URL OpenAI API i listuje modele od wielu dostawców w ramach tej samej usługi.

Daje to aplikacji użyteczny podział odpowiedzialności:

CrewAI

• Definiuje role agentów
• Definiuje zadania
• Przekazuje kontekst
• Kontroluje wykonanie
• Zarządza iteracjami agentów
• Obsługuje orkiestrację na poziomie załogi

CometAPI

• Zapewnia wspólną warstwę dostępu do modeli
• Centralizuje uwierzytelnianie API
• Zapewnia routing modeli przez ID modeli
• Daje aplikacji jeden punkt końcowy API
• Zapewnia scentralizowaną widoczność użycia i rozliczeń

---

Co zbuduje ten przepływ pracy CrewAI?

Przykład tworzy trzech agentów wykonujących prace sekwencyjnie.

To jest przykładowa polityka routingu, a nie ranking wydajności.

Właściwy model dla Twojego agenta zależy od:

• złożoności zadania
• wymaganej długości kontekstu
• użycia narzędzi
• wymagań dotyczących struktury wyjścia
• opóźnienia
• niezawodności
• kosztu tokenów
• jakości wyjścia
• wyników oceny specyficznych dla aplikacji

Użyteczna zasada:

Wybierz model odpowiedni do pracy, jaką wykonuje agent, a nie po prostu ze względu na dostawcę, od którego pochodzi.

---

Jaki model powinien używać każdy agent CrewAI?

Dla tego przykładu przypisanie modelu podąża za prostą strategią koszt kontra możliwości.

Researcher: Gemini 3.7 Flash

Badania często obejmują przetwarzanie stosunkowo dużej ilości informacji i tworzenie zwartego wyniku pośredniego.

Szybki model może być zatem użyteczny w zadaniach badawczych o dużej objętości.

"researcher": "gemini-3.7-flash"

Analyst: Claude Opus 5

Analityk ma rolę węższą, ale bardziej intensywną pod względem rozumowania. Otrzymuje wynik badań i zamienia go w rekomendację.

"analyst": "claude-opus-5"

Writer: GPT-5.6

Ostatni agent przekształca badania i analizę w techniczną notę decyzyjną skierowaną do deweloperów.

"writer": "gpt-5.6"

Ważny nie jest dokładnie taki dobór trzech modeli. Twoja aplikacja powinna ocenić kandydackie modele na reprezentatywnych zadaniach przed utrwaleniem polityki routingu.

---

Co potrzebujesz przed rozpoczęciem?

Potrzebujesz:

• Python 3.10+
• CrewAI
• Zgodności z OpenAI Python SDK
• python-dotenv
• Klucza API CometAPI
• Identyfikatorów modeli, których zamierzasz używać

Aktualna integracja Pythona w CometAPI wspiera API kompatybilne z OpenAI, a oficjalny pakiet CometAPI dla Pythona dokumentuje COMETAPI_KEY i COMETAPI_BASE_URL jako opcje konfiguracji oparte na środowisku.

Standardowy punkt końcowy to:

https://api.cometapi.com/v1

Przed wdrożeniem zweryfikuj, że wybrane identyfikatory modeli są obecnie dostępne i wspierają punkt końcowy oraz parametry wymagane przez Twój przepływ pracy CrewAI. Katalogi modeli i ceny mogą się zmieniać.

---

Jak zainstalować CrewAI i zależności?

Utwórz nowe środowisko Pythona:

python -m venv .venv

Aktywuj je:

source .venv/bin/activate

Na Windows:

.venv\Scripts\Activate.ps1

Następnie zainstaluj zależności:

pip install "crewai[openai]" openai python-dotenv

Jawne użycie openai jest zamierzone, ponieważ poniższa implementacja fallbacku importuje bezpośrednio klasy wyjątków OpenAI SDK.

W produkcji przypnij wersje, które testujesz, zamiast polegać bezterminowo na najnowszych wersjach.

Na przykład:

crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION

Warstwa LLM w CrewAI aktywnie ewoluuje, więc dokładny konstruktor i konfiguracja dostawcy powinny być sprawdzone względem wersji CrewAI używanej przez Twoją aplikację. Aktualna dokumentacja CrewAI wspiera konfigurację LLM z niestandardowym base_url i kluczem API.

---

Jak skonfigurować klucz API CometAPI?

Utwórz plik .env:

COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1

Załaduj te wartości w Pythonie:

import osfrom dotenv import load_dotenvload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv(    "COMETAPI_BASE_URL",    "https://api.cometapi.com/v1",)

Nigdy nie commituj .env do Git.

Dodaj go do .gitignore:

.env.venv/__pycache__/

Klucz API powinien pozostać poświadczeniem po stronie serwera. Aktualne wskazówki szybkiego startu CometAPI również zalecają przechowywanie klucza w zmiennych środowiskowych zamiast w kodzie źródłowym.

---

Jak podłączyć CrewAI do CometAPI?

Obiekt LLM w CrewAI może otrzymać nazwę modelu, klucz API i niestandardowy bazowy URL.

Utwórz pomocnika:

from crewai import LLMdef cometapi_llm(model_id: str) -> LLM:    return LLM(        model=model_id,        base_url=COMETAPI_BASE_URL,        api_key=COMETAPI_KEY,        timeout=60.0,        max_retries=0,    )

Jest to preferowane względem osadzania tej samej konfiguracji osobno w każdym agencie.

Każdy agent potrzebuje teraz tylko ID modelu:

research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")

Dlaczego ustawić max_retries=0?

Powodem jest kontrola fallbacku.

Jeśli klient LLM pod spodem automatycznie ponawia próby, a Twoja aplikacja również implementuje fallback, jedna awaria może stać się kilkoma ukrytymi żądaniami zanim logika fallbacku zadziała.

Dla samouczka z jawnym routingiem, czyściej jest pozwolić aplikacji decydować, kiedy ponowić próbę lub przełączyć model.

---

Jak zdefiniować politykę routingu modeli?

Trzymaj routing poza promptami:

PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",}FALLBACK_MODELS = {    "researcher": "gpt-5.6",    "analyst": "gpt-5.6",    "writer": "gemini-3.7-flash",}

Tworzy to wyraźną granicę konfiguracji.

Później możesz przenieść ten sam mapping do:

• konfiguracji środowiskowej
• YAML
• JSON
• bazy danych
• flag funkcji
• wewnętrznej usługi routingu modeli

bez przepisywania promptów agentów.

---

Jak zbudować trzech agentów CrewAI?

Utwórz jeden obiekt LLM na agenta.

from crewai import Agentdef build_agents(model_map: dict[str, str]):    researcher = Agent(        role="Market Researcher",        goal="Collect the facts needed to answer the topic",        backstory=(            "You create concise, source-aware research briefs "            "and clearly separate facts from assumptions."        ),        llm=cometapi_llm(model_map["researcher"]),        max_iter=3,        allow_delegation=False,    )    analyst = Agent(        role="Product Analyst",        goal="Turn research into a defensible recommendation",        backstory=(            "You identify evidence, assumptions, risks, "            "and trade-offs before making recommendations."        ),        llm=cometapi_llm(model_map["analyst"]),        max_iter=3,        allow_delegation=False,    )    writer = Agent(        role="Technical Writer",        goal="Produce a concise technical decision memo",        backstory=(            "You write clear technical explanations "            "without unnecessary marketing language."        ),        llm=cometapi_llm(model_map["writer"]),        max_iter=3,        allow_delegation=False,    )    return researcher, analyst, writer

Przypisanie modeli jest teraz całkowicie niezależne od definicji roli agenta.

To właśnie czyni routing modeli praktycznym.

---

Jak połączyć agentów zadaniami sekwencyjnymi?

Utwórz trzy zadania:

from crewai import Taskdef build_tasks(researcher, analyst, writer):    research_task = Task(        description=(            "Research this topic: {topic}. "            "Return the key facts, uncertainties, "            "and relevant sources that the analyst should consider."        ),        expected_output=(            "A compact research brief containing facts, "            "uncertainties, and source references."        ),        agent=researcher,    )    analysis_task = Task(        description=(            "Using the research brief, analyze {topic}. "            "Identify the strongest conclusion and explain "            "the major trade-offs."        ),        expected_output=(            "A decision outline with evidence, "            "assumptions, risks, and trade-offs."        ),        agent=analyst,        context=[research_task],    )    writing_task = Task(        description=(            "Write a concise technical decision memo about {topic}. "            "State the recommendation early and preserve "            "important caveats."        ),        expected_output="A polished technical decision memo in Markdown.",        agent=writer,        context=[research_task, analysis_task],    )    return research_task, analysis_task, writing_task

Łańcuch zależności to:

Temat  ↓Badania  ↓Analiza  ↓Końcowa nota

Analityk otrzymuje wynik zadania badawczego, podczas gdy autor otrzymuje zarówno kontekst badań, jak i analizy.

---

Jak zbudować załogę (Crew)?

Połącz agentów i zadania:

from crewai import Crew, Processdef build_crew(model_map: dict[str, str]) -> Crew:    researcher, analyst, writer = build_agents(model_map)    research_task, analysis_task, writing_task = build_tasks(        researcher,        analyst,        writer,    )    return Crew(        agents=[researcher, analyst, writer],        tasks=[            research_task,            analysis_task,            writing_task,        ],        process=Process.sequential,        verbose=True,    )

Teraz routing modeli jest w pełni sterowany konfiguracją.

Zmiana:

"researcher": "gemini-3.7-flash"

na inny wspierany model nie wymaga zmiany promptu badawczego ani definicji zadania.

---

Jak powinien działać fallback modelu w CrewAI?

Tu implementacja zorientowana na produkcję wymaga większej uwagi.

Częstym błędem jest:

Dowolny błąd   ↓Przełącz model

To zbyt agresywne.

Na przykład poniższe błędy generalnie nie powinny uruchamiać fallbacku modelu:

400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error

Przełączenie modeli nie naprawi nieprawidłowego klucza API ani błędnego żądania.

Fallback jest bardziej odpowiedni dla tymczasowych awarii, takich jak:

408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout

Polityka fallbacku powinna zatem być:

Ponawiaj próbę lub przełączaj modele tylko dla ograniczonych, przejściowych awarii i tylko wtedy, gdy model fallback wspiera ten sam kontrakt żądania.

---

Jak wykryć błędy nadające się do retry?

Możesz użyć klas błędów z OpenAI SDK:

from collections.abc import Iteratorfrom openai import (    APIConnectionError,    APIStatusError,    APITimeoutError,)def exception_chain(error: BaseException) -> Iterator[BaseException]:    current: BaseException | None = error    seen: set[int] = set()    while current is not None and id(current) not in seen:        seen.add(id(current))        yield current        current = (            current.__cause__            or current.__context__        )def should_fallback(error: BaseException) -> bool:    for current in exception_chain(error):        if isinstance(            current,            (APIConnectionError, APITimeoutError),        ):            return True        if isinstance(current, APIStatusError):            return (                current.status_code in {408, 429}                or current.status_code >= 500            )    return False

To celowo wyklucza błędy konfiguracji z zakresu 400 z wyjątkiem 408 i 429.

---

Czy ponawiać cały Crew czy tylko nieudanego agenta?

Istnieją dwie różne strategie fallbacku.

Fallback na poziomie Crew

Najprostsza implementacja to:

Start crew   ↓awaria   ↓zmień routing   ↓uruchom crew ponownie

To łatwe do zrozumienia, ale może powtórzyć ukończone zadania.

Na przykład:

Badania → ukończoneAnaliza → ukończoneAutor → awaria

Pełny retry kickoff() może wykonać:

Badania → ponownieAnaliza → ponownieAutor → fallback

To zwiększa:

• zużycie tokenów
• opóźnienie
• koszt API
• potencjalne efekty uboczne

Odzyskiwanie na poziomie zadań

Przepływ pracy produkcyjny powinien zamiast tego zapisywać checkpointy ukończonej pracy:

Badania   ↓checkpoint   ↓Analiza   ↓checkpoint   ↓Autor zawodzi   ↓ponów autora z fallbackiem

CrewAI obecnie zapewnia checkpointing, który zapisuje stan wykonania i pozwala wznowić przebieg po awarii. Udokumentowane zachowanie checkpointów pomija ukończone zadania i kontynuuje dalszą pracę od zapisanego stanu.

To lepsza architektura dla kosztownych lub wywołujących skutki uboczne przepływów pracy.

---

Jak dodać checkpointing CrewAI?

Dla przepływów produkcyjnych włącz checkpointing w załodze:

crew = Crew(    agents=[researcher, analyst, writer],    tasks=[        research_task,        analysis_task,        writing_task,    ],    process=Process.sequential,    checkpoint=True,    verbose=True,)

System checkpointów CrewAI może utrwalić stan wykonania po ukończeniu zadania i odtworzyć załogę z checkpointu.

Na przykład, odtworzony przebieg może użyć:

from crewai import CheckpointConfigresult = crew.kickoff(    from_checkpoint=CheckpointConfig(        restore_from="./.checkpoints/checkpoint.json",    ))

Dokładna konfiguracja checkpointów powinna odpowiadać wersji CrewAI używanej w Twoim projekcie.

Ważny punkt architektoniczny:

Najpierw checkpoint, potem fallback.

Zapobiega to sytuacji, w której tymczasowa awaria modelu wymusza ponowne uruchamianie kosztownie ukończonej pracy.

---

Jak zaimplementować prosty ograniczony fallback?

Dla samouczka nadal możesz zademonstrować prosty fallback na poziomie crew.

def run_with_fallback(topic: str):    routes = [        PRIMARY_MODELS,        {            **PRIMARY_MODELS,            "writer": FALLBACK_MODELS["writer"],        },        {            **PRIMARY_MODELS,            "analyst": FALLBACK_MODELS["analyst"],            "writer": FALLBACK_MODELS["writer"],        },    ]    last_error = None    for attempt, model_map in enumerate(routes, start=1):        try:            crew = build_crew(model_map)            result = crew.kickoff(                inputs={"topic": topic}            )            return result, model_map        except Exception as error:            last_error = error            if not should_fallback(error):                raise            if attempt == len(routes):                raise            print(                f"Transient failure on attempt {attempt}. "                f"Trying bounded fallback route.",                flush=True,            )    raise RuntimeError(        "Crew execution failed after all fallback routes."    ) from last_error

Zwróć uwagę na ważne rozróżnienie:

To nie twierdzi, że zidentyfikowano dokładnie nieudanego agenta.

To ograniczona strategia fallbacku na poziomie crew.

Dla małych bezstanowych przepływów pracy może to być akceptowalne. Dla przepływów produkcyjnych z kosztownymi badaniami, narzędziami lub efektami ubocznymi użyj odzyskiwania opartego na checkpointach.

---

Jak śledzić zużycie tokenów w CrewAI?

Śledzenie użycia powinno być częścią warstwy routingu, a nie dodatkiem.

Na końcu przebiegu sprawdź wynik CrewAI:

result, selected_models = run_with_fallback(topic)print("Selected models:")print(selected_models)print("Final result:")print(result.raw)print("Usage:")print(result.token_usage)

Dokładne dostępne pola użycia mogą zależeć od wersji CrewAI i ścieżki wykonania, więc traktuj zwrócony obiekt wyniku jako źródło prawdy dla wersji, którą wdrażasz.

Produkcja powinna idealnie zawierać rekord użycia:

job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at

Pozwala to odpowiedzieć na pytania takie jak:

Który agent konsumuje najwięcej budżetu?

Jak często analityk stosuje fallback?

Który model ma najwyższe opóźnienie?

Ile kosztuje każdy przepływ pracy?

---

Jak kontrolować koszt na poziomie agenta?

Routowanie wielomodelowe jest najbardziej użyteczne, gdy odzwierciedla rzeczywiste różnice w obciążeniu.

Na przykład:

Researcher→ duża objętość→ tańszy modelAnalyst→ mała objętość→ silniejszy model rozumowaniaWriter→ średnia objętość→ model produkcyjny ogólnego przeznaczenia

Możesz także ograniczać koszt poprzez konfigurację agenta.

Na przykład:

max_iter=3

ogranicza pętlę iteracji agenta. Nie należy interpretować tego jako twardego limitu dokładnie trzech wywołań API lub trzech budżetów tokenów.

Dodatkowe kontrolki obejmują:

• ograniczenie kontekstu zadania
• podsumowanie wyników pośrednich
• cache’owanie powtarzalnych badań
• ograniczenie maksymalnego rozmiaru wejścia
• ograniczenie maksymalnych tokenów wyjściowych tam, gdzie jest to wspierane
• ograniczenie wywołań narzędzi
• ustawianie budżetów per użytkownik
• ustawianie budżetów per przepływ pracy
• śledzenie częstotliwości fallbacku

---

Jak walidować modele przed wdrożeniem?

Nie hardcoduj ID modeli na zawsze.

Model może stać się:

• niedostępny
• przemianowany
• wycofany
• ograniczony
• zmieniony pod względem możliwości
• zmieniony pod względem cen
• niekompatybilny z parametrem używanym przez Twoją aplikację

CometAPI udostępnia punkt końcowy katalogu modeli, który można odpytywać programowo, a jego publiczny katalog modeli może służyć do odkrywania modeli przez ludzi.

Kontrola wdrożeniowa może wyglądać tak:

curl -s \  https://api.cometapi.com/api/models \  -H "Authorization: Bearer $COMETAPI_KEY"

Następnie zweryfikuj, że skonfigurowane ID modeli istnieją przed wdrożeniem.

Na przykład Twój proces CI może sprawdzić:

gemini-3.7-flash → availableclaude-opus-5    → availablegpt-5.6          → available

Nie zastępuj testów aplikacji kontrolami dostępności. Obecność modelu w katalogu nie oznacza, że każdy parametr, narzędzie czy format wyjścia używany przez Twojego agenta CrewAI jest wspierany.

---

Jak wygląda pełny przykład CrewAI?

Oto skonsolidowana implementacja:

import jsonimport osimport sysfrom collections.abc import Iteratorfrom dotenv import load_dotenvfrom openai import (    APIConnectionError,    APIStatusError,    APITimeoutError,)from crewai import Agent, Crew, LLM, Process, Taskload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv(    "COMETAPI_BASE_URL",    "https://api.cometapi.com/v1",)PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",}FALLBACK_MODELS = {    "researcher": "gpt-5.6",    "analyst": "gpt-5.6",    "writer": "gemini-3.7-flash",}def cometapi_llm(model_id: str) -> LLM:    return LLM(        model=model_id,        base_url=COMETAPI_BASE_URL,        api_key=COMETAPI_KEY,        timeout=60.0,        max_retries=0,    )def build_crew(model_map: dict[str, str]) -> Crew:    researcher = Agent(        role="Market Researcher",        goal="Collect the facts needed to answer the topic",        backstory=(            "You create concise, source-aware research briefs "            "and distinguish facts from assumptions."        ),        llm=cometapi_llm(model_map["researcher"]),        max_iter=3,        allow_delegation=False,    )    analyst = Agent(        role="Product Analyst",        goal="Turn research into a defensible recommendation",        backstory=(            "You evaluate evidence, assumptions, risks, "            "and trade-offs."        ),        llm=cometapi_llm(model_map["analyst"]),        max_iter=3,        allow_delegation=False,    )    writer = Agent(        role="Technical Writer",        goal="Produce a concise technical decision memo",        backstory=(            "You write clear technical explanations "            "without unnecessary hype."        ),        llm=cometapi_llm(model_map["writer"]),        max_iter=3,        allow_delegation=False,    )    research_task = Task(        description=(            "Research this topic: {topic}. "            "Return the key facts, uncertainties, "            "and relevant sources."        ),        expected_output=(            "A concise research brief with facts "            "and open questions."        ),        agent=researcher,    )    analysis_task = Task(        description=(            "Using the research brief, analyze {topic}. "            "Identify the strongest conclusion and "            "explain the major trade-offs."        ),        expected_output=(            "A decision outline with evidence, "            "assumptions, risks, and trade-offs."        ),        agent=analyst,        context=[research_task],    )    writing_task = Task(        description=(            "Write a concise technical decision memo "            "about {topic}. State the recommendation early "            "and preserve important caveats."        ),        expected_output=(            "A polished technical decision memo in Markdown."        ),        agent=writer,        context=[            research_task,            analysis_task,        ],    )    return Crew(        agents=[            researcher,            analyst,            writer,        ],        tasks=[            research_task,            analysis_task,            writing_task,        ],        process=Process.sequential,        verbose=True,    )def exception_chain(    error: BaseException,) -> Iterator[BaseException]:    current = error    seen: set[int] = set()    while current is not None and id(current) not in seen:        seen.add(id(current))        yield current        current = (            current.__cause__            or current.__context__        )def should_fallback(error: BaseException) -> bool:    for current in exception_chain(error):        if isinstance(            current,            (                APIConnectionError,                APITimeoutError,            ),        ):            return True        if isinstance(current, APIStatusError):            return (                current.status_code in {408, 429}                or current.status_code >= 500            )    return Falsedef run_with_fallback(topic: str):    routes = [        PRIMARY_MODELS,        {            **PRIMARY_MODELS,            "writer": FALLBACK_MODELS["writer"],        },        {            **PRIMARY_MODELS,            "analyst": FALLBACK_MODELS["analyst"],            "writer": FALLBACK_MODELS["writer"],        },    ]    last_error = None    for attempt, model_map in enumerate(        routes,        start=1,    ):        try:            crew = build_crew(model_map)            result = crew.kickoff(                inputs={                    "topic": topic,                }            )            return result, model_map        except Exception as error:            last_error = error            if not should_fallback(error):                raise            if attempt == len(routes):                raise            print(                f"Transient failure on attempt "                f"{attempt}; trying fallback.",                file=sys.stderr,            )    raise RuntimeError(        "No model route completed the crew."    ) from last_errordef main():    topic = (        sys.argv[1]        if len(sys.argv) > 1        else (            "Should a small SaaS add "            "AI-generated meeting summaries?"        )    )    result, selected_models = (        run_with_fallback(topic)    )    output = {        "selected_models": selected_models,        "raw": result.raw,        "tasks_output": [            task.raw            for task in result.tasks_output        ],        "token_usage": str(            result.token_usage        ),    }    print(        json.dumps(            output,            indent=2,            default=str,        )    )if __name__ == "__main__":    main()

Ważną poprawą względem wersji oryginalnej jest to, że kod nie sugeruje fałszywie, iż wyjątek identyfikuje dokładnie nieudanego agenta.

To jawnie ograniczona implementacja fallbacku na poziomie crew.

W produkcji połącz tę samą politykę routingu z checkpointingiem CrewAI.

---

Jak uruchomić przepływ pracy CrewAI?

Zapisz plik jako:

crewai_multi_model.py

Następnie uruchom:

python crewai_multi_model.py \  "Should a small SaaS add AI-generated meeting summaries?"

Pomyślna odpowiedź będzie zawierać informacje podobne do:

{  "selected_models": {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6"  },  "raw": "<final decision memo>",  "tasks_output": [    "<research output>",    "<analysis output>",    "<writing output>"  ],  "token_usage": "<usage information>"}

Dokładna odpowiedź i wartości użycia zależą od wejścia, zachowania modelu, wersji CrewAI i ścieżki wykonania.

Jeśli błąd podlegający ponowieniu aktywuje trasę fallback, obiekt selected_models pokazuje trasę używaną dla tego wykonania crew.

---

Jak zaprojektować routing modeli w produkcji?

Polityka routingu produkcyjnego powinna uwzględniać więcej niż jakość modelu.

Użyteczna funkcja decyzyjna to:

Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency

Możesz zaimplementować to na kilku poziomach.

Routing oparty na koszcie

Proste zadanie → ekonomiczny modelZłożone zadanie → model premium

Routing oparty na opóźnieniu

Żądanie interaktywne → szybki modelPrzepływ w tle → model wyższej jakości

Routing oparty na niezawodności

Model podstawowy     ↓przejściowa awaria     ↓model fallback

Routing oparty na zadaniu

Badania → Model AAnaliza → Model BPisanie → Model CKod → Model D

Ostatnie podejście jest szczególnie naturalne dla CrewAI, ponieważ framework już nadaje każdemu agentowi odrębną rolę.

---

Jak uczynić fallback bezpiecznym?

Solidny system fallbacku powinien egzekwować cztery zasady.

Nie wykonuj fallbacku przy błędach uwierzytelniania

Jeśli klucz API jest nieprawidłowy:

401

zmiana modeli nie rozwiąże problemu.

Nie wykonuj fallbacku przy błędnych żądaniach

Jeśli żądanie jest nieprawidłowe:

400422

popraw żądanie zamiast przełączać model.

Nie wykonuj fallbacku bez końca

Ustaw twardy limit:

MAX_FALLBACK_ATTEMPTS = 2

System fallback bez limitu może stać się kosztowną pętlą ponowień.

Spraw, aby modele fallback były zgodne z żądaniem

Model fallback musi wspierać funkcje wymagane przez Twojego agenta.

Na przykład, jeśli podstawowy agent wymaga konkretnego narzędzia lub zachowania związanego ze strukturą wyjścia, fallback musi wspierać ten sam kontrakt.

OpenAI-kompatybilny nie znaczy funkcjonalnie kompatybilny.

---

Najczęstsze błędy CrewAI + CometAPI

---

Jak oddzielić błędy CrewAI od błędów modeli?

To rozróżnienie jest ważne przy debugowaniu.

Błędy konfiguracji

Missing API keyInvalid model IDInvalid base URLUnsupported parameter

Te powinny szybko kończyć się niepowodzeniem.

Błędy dostawcy/API

401403404429500503

Wymagają różnego postępowania w zależności od statusu.

Błędy aplikacji

Agent output invalidTool returned malformed dataTask context missingSide effect failed

Nie są one koniecznie rozwiązywane przez zmianę modeli.

Dojrzały system agentów powinien zatem mieć osobną obsługę dla:

configuration      ↓API transport      ↓model execution      ↓agent logic      ↓tool execution      ↓application side effects

To znacznie bezpieczniejsze niż ogólny:

except Exception:    use_fallback()

---

Jak chronić zewnętrzne skutki uboczne?

Fallback staje się znacznie bardziej złożony, gdy agenty robią coś więcej niż generowanie tekstu.

Na przykład wyobraź sobie agenta, który:

1. tworzy rekord w bazie danych
2. wysyła e-mail
3. wywołuje zewnętrzne API
4. aktualizuje CRM

Jeśli model przekroczy czas po tym, jak zewnętrzna akcja się powiedzie, ponowne uruchomienie całej załogi może zdublować tę akcję.

Używaj:

• kluczy idempotencji
• checkpointów zadań
• granic transakcji
• ID wykonania
• trwałego stanu zadań
• jawnego potwierdzenia skutków ubocznych

Na przykład:

job_id = crew_run_123task_id = writer_456

Przechowuj te identyfikatory wraz z operacjami zewnętrznymi, aby ponowienie mogło określić, czy operacja już miała miejsce.

---

Jak monitorować wielomodelowe przepływy pracy CrewAI?

Minimum, loguj:

workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens

Nie loguj:

API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output

Dla każdego modelu monitoruj:

Niezawodność

success ratetimeout rate5xx ratefallback rate

Wydajność

p50 latencyp95 latencyp99 latency

Koszt

input tokensoutput tokenscost per taskcost per completed workflow

Jakość

task success ratehuman evaluationstructured-output validitytool-call success

To zamienia routing modeli z zakodowanej na stałe preferencji w obserwowalny system inżynieryjny.

---

Jak wybrać między bezpośrednimi API dostawców a CometAPI?

Wybór zależy od Twojej architektury.

Jeśli Twoja aplikacja potrzebuje tylko jednego dostawcy i jego natywnych możliwości, bezpośrednia integracja może być zupełnie rozsądnym wyborem.

Jeśli Twoja aplikacja CrewAI potrzebuje modeli od kilku dostawców i chcesz jedną warstwę dostępu, CometAPI staje się bardziej atrakcyjne.

Ważne jest to, że CometAPI nie zastępuje CrewAI.

Zamiast tego:

CrewAIOrkiestracja agentów       ↓CometAPIDostęp do modeli       ↓Wiele modeli

Każda warstwa ma inną odpowiedzialność.

---

Jak ta architektura się skaluje?

Gdy polityka routingu jest odseparowana od definicji agentów, dodanie kolejnego modelu nie wymaga przebudowy całej aplikacji.

Na przykład:

PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",    "coder": "YOUR_CODE_MODEL",}

Ta sama architektura może wtedy wspierać:

Agent badawczyAgent analitycznyAgent kodującyAgent recenzującyAgent piszącyAgent fact-checking

Każdy agent może mieć inny model, korzystając ze wspólnej warstwy dostępu CometAPI.

Kolejny krok to uczynienie routingu dynamicznym.

Zamiast:

"analyst": "claude-opus-5"

możesz ostatecznie użyć:

select_model(    task="analysis",    budget=budget,    latency_target=latency_target,)

System routingu może wtedy wybierać spośród zatwierdzonych modeli na podstawie wymagań aplikacji.

---

Jaka jest najlepsza architektura produkcyjna dla CrewAI + CometAPI?

Dla małego przepływu:

Wejście użytkownika   ↓CrewAI   ↓CometAPI   ↓Modele

Dla produkcji:

                     ┌───────────────┐                     │ Model Catalog │                     └───────┬───────┘                             │                             ▼User → CrewAI → Routing Policy → CometAPI           │          │             │           │          │             ├── Gemini           │          │             ├── Claude           │          │             └── GPT           │          │           │          ▼           │     Cost / Quality /           │     Latency / Policy           │           ▼      Checkpoints           │           ▼      Usage Tracking

Kluczowe komponenty produkcyjne to:

1. Lista dozwolonych modeli (allowlist)
2. Routowanie per agent
3. Ograniczone retry
4. Checkpointing zadań
5. Śledzenie użycia
6. Kontrola kosztów
7. Testy kompatybilności modeli
8. Obserwowalność
9. Idempotentne skutki uboczne

Ta architektura jest znacznie bardziej odporna niż po prostu dodanie try/except wokół crew.kickoff().

---

Jeden klucz CometAPI, różne modele, klarowne role agentów

Najbardziej użyteczny sposób myślenia o CrewAI i CometAPI razem to jako dwie warstwy uzupełniające się.

CrewAI definiuje, co robią agenty.

CometAPI definiuje, jak te agenty uzyskują dostęp do modeli.

To rozdzielenie pozwala przypisać szybki model do agenta badawczego o wysokiej objętości, silniejszy model rozumowania do agenta analitycznego i model ogólnego przeznaczenia do finalnego autora bez utrzymywania osobnych integracji dostawców wewnątrz przepływu pracy.

Najprostsza implementacja używa jednego klucza CometAPI i jednego bazowego URL kompatybilnego z OpenAI:

https://api.cometapi.com/v1

W produkcji zrób kolejny krok: utrzymuj routing modeli w konfiguracji, waliduj dostępność modeli przed wdrożeniem, używaj ograniczonego fallbacku tylko dla przejściowych awarii, zapisuj checkpointy ukończonych zadań i rejestruj metadane modelu oraz użycia dla każdego przebiegu.

Daje to znacznie bardziej trwały wzorzec niż samo podłączenie CrewAI do jednego LLM:

CrewAI orkiestruje agenty. CometAPI centralizuje dostęp do modeli. ID modeli sterują routingiem. Checkpointy chronią ukończoną pracę. Śledzenie użycia kontroluje koszt.

---

Często zadawane pytania

Czy CrewAI może używać wielu modeli AI w tej samej załodze?

Tak. Przypisz inną konfigurację LLM do każdego agenta CrewAI. Każda konfiguracja może określić własny model, używając jednego klucza API CometAPI i jednego bazowego URL.

Czy CrewAI może łączyć się z API kompatybilnym z OpenAI?

Tak. Konfiguracja LLM w CrewAI wspiera niestandardowy base_url i klucz API dla punktów końcowych kompatybilnych z OpenAI.

Z CometAPI bazowy URL to:

https://api.cometapi.com/v1

Czy potrzebuję osobnych kluczy API dla GPT, Claude i Gemini?

Uzyskując dostęp do tych modeli przez CometAPI, aplikacja może użyć poświadczenia CometAPI i punktu końcowego zamiast implementowania osobnych poświadczeń dostawców w każdym agencie CrewAI.

Czy jeden klucz API oznacza identyczne możliwości modeli?

Nie. Interfejs API może być ujednolicony, podczas gdy możliwości modeli pozostają różne. Okna kontekstu, wsparcie narzędzi, parametry, zachowanie wyjścia, opóźnienie i ceny mogą różnić się między modelami.

Czy powinienem ponawiać cały przepływ pracy CrewAI, gdy jeden model zawiedzie?

Tylko dla prostych, bezstanowych przepływów. Ponowienie całej załogi może powtórzyć ukończone zadania i zwiększyć koszt. W produkcji zapisuj checkpointy ukończonych zadań i wznawiaj od nieudanego fragmentu, gdzie to możliwe.

Aktualna funkcjonalność checkpointów w CrewAI jest zaprojektowana tak, aby zachować stan wykonania i wznowić po awariach.

Czy każdy wyjątek CrewAI powinien uruchamiać fallback?

Nie. Uwierzytelnianie, błędne żądania, nieprawidłowe ID modelu i niewspierane parametry zazwyczaj wymagają zmian konfiguracji, a nie innego modelu.

Fallback lepiej rezerwować dla ograniczonych awarii przejściowych, takich jak przekroczenia czasu, limity zapytań i tymczasowe odpowiedzi 5xx.

Jak śledzić koszt każdego agenta CrewAI?

Rejestruj nazwę agenta, ID modelu, użycie tokenów, opóźnienie, status wykonania i informacje o fallbacku dla każdego zadania. Użyj wynikowych danych do obliczenia kosztu per agent i per przepływ pracy.

Czy mogę dynamicznie zmieniać model przypisany do agenta?

Tak. Przechowuj ID modeli w konfiguracji routingu zamiast osadzania ich bezpośrednio w definicjach agentów. Twoja aplikacja może wtedy wybierać modele na podstawie kosztu, opóźnienia, typu zadania lub dostępności.

Czy CometAPI zastępuje CrewAI?

Nie. Działają na różnych warstwach. CrewAI orkiestruje agentów i zadania, podczas gdy CometAPI zapewnia ujednoliconą warstwę dostępu do modeli.

Gdzie znajdę aktualne modele CometAPI?

Użyj katalogu modeli CometAPI do odkrywania modeli przez ludzi oraz API modeli do programowej walidacji. Aktualna strona szybkiego startu CometAPI listuje 500+ modeli w kategoriach tekst, obraz, wideo i audio.

---

Źródła

• CometAPI Quick Start
• CometAPI Model Directory
• CometAPI API Documentation
• CometAPI Python SDK
• CrewAI Documentation
• CrewAI LLM Configuration
• CrewAI Checkpointing