Membina sistem berbilang ejen dengan CrewAI menjadi lebih menarik apabila ejen berbeza boleh menggunakan model yang berbeza.
Seorang penyelidik mungkin mendapat manfaat daripada model yang pantas dan ekonomik, seorang penganalisis mungkin memerlukan model penaakulan yang lebih berkuasa, dan seorang penulis mungkin memerlukan model yang dioptimumkan untuk penjanaan panjang berkualiti tinggi. Secara tradisional, menyambungkan ejen-ejen ini kepada pembekal yang berbeza bermakna mengurus kelayakan API, titik akhir, SDK, sistem pengebilan, dan konfigurasi khusus pembekal secara berasingan.
Seni bina yang lebih bersih ialah membiarkan CrewAI mengurus ejen dan aliran kerja sementara CometAPI mengurus capaian model.
CometAPI menyediakan titik akhir serasi OpenAI di https://api.cometapi.com/v1, jadi aplikasi boleh menghala permintaan kepada model daripada pelbagai pembekal melalui antara muka API bersama. Dokumentasi mula pantas semasanya juga menyokong penggunaan SDK Python OpenAI standard dengan menukar kunci API dan URL asas.
Dalam tutorial ini, anda akan membina aliran kerja CrewAI tiga ejen dengan:
- Gemini 3.7 Flash untuk penyelidikan
- Claude Opus 5 untuk analisis
- GPT-5.6 untuk penulisan akhir
- Satu kunci API CometAPI
- Satu URL asas API
- Konfigurasi model per ejen
- Fallback terhad untuk kegagalan sementara
- Checkpointing CrewAI untuk pemulihan produksi
- Penjejakan penggunaan token dan pelaksanaan
- Pengesahan model di sisi pelayan
Sempadan seni bina penting adalah mudah:
CrewAI mengendalikan orkestrasi ejen. CometAPI mengendalikan capaian model. ID model menentukan perutean.
Apakah Perutean Model Berbilang Ejen CrewAI?
CrewAI ialah rangka kerja Python untuk mencipta ejen, tugas, kru, dan aliran kerja berbilang ejen. Setiap ejen boleh mempunyai konfigurasi LLMnya sendiri, manakala Crew menyelaras bagaimana ejen-ejen itu melaksanakan tugas dan bertukar konteks.
Konfigurasi LLM semasa CrewAI menyokong tetapan model, api_key, dan base_url yang jelas, termasuk titik akhir serasi OpenAI tersuai.
Itu menjadikan seni bina berbilang model mudah:
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Ejen kekal berasingan dari perspektif logik, tetapi capaian model mereka dipusatkan.
Ini berbeza daripada mengatakan semua model boleh saling gantikan. API serasi OpenAI menyediakan antara muka permintaan bersama; ia tidak menjamin had konteks, sokongan alat, kawalan penaakulan, tingkah laku output, kependaman, atau harga yang sama.
Perbezaan itu penting apabila mereka bentuk perutean produksi.
Mengapa Menggunakan CometAPI dengan CrewAI?
Kelebihan utama bukanlah CrewAI tiba-tiba menjadi rangka kerja pelbagai pembekal. CrewAI sudah menyokong pelbagai pembekal LLM.
Kelebihannya ialah capaian model boleh disatukan di belakang satu lapisan API.
Tanpa lapisan API bersatu, aliran kerja tiga ejen mungkin kelihatan seperti ini:
| Ejen | Pembekal | Kelayakan | Integrasi |
|---|---|---|---|
| Researcher | Kunci API Google | Khusus pembekal | |
| Analyst | Anthropic | Kunci API Anthropic | Khusus pembekal |
| Writer | OpenAI | Kunci API OpenAI | Khusus pembekal |
Dengan CometAPI:
| Ejen | Model | Kelayakan | Endpoint |
|---|---|---|---|
| Researcher | Gemini 3.7 Flash | Kunci CometAPI | CometAPI |
| Analyst | Claude Opus 5 | Kunci CometAPI | CometAPI |
| Writer | GPT-5.6 | Kunci CometAPI | CometAPI |
Dokumentasi mula pantas semasa CometAPI menerangkan titik akhirnya sebagai pengganti terus untuk URL asas API OpenAI dan menyenaraikan model daripada pelbagai pembekal melalui perkhidmatan yang sama.
Ini memberi aplikasi satu pemisahan yang berguna:
CrewAI
- Mendefinisikan peranan ejen
- Mendefinisikan tugas
- Menyampaikan konteks
- Mengawal pelaksanaan
- Mengurus iterasi ejen
- Mengendalikan orkestrasi peringkat kru
CometAPI
- Menyediakan lapisan capaian model bersama
- Memusatkan pengesahan API
- Menyediakan perutean model melalui ID model
- Memberi aplikasi satu titik akhir API
- Menyediakan keterlihatan penggunaan dan pengebilan berpusat
Aliran Kerja CrewAI Ini Akan Membina Apa?
Contoh ini mencipta tiga ejen berjujukan.
| Ejen CrewAI | Model Utama | Fallback | Peranan |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | Mengumpul fakta dan penyelidikan |
| Product Analyst | claude-opus-5 | gpt-5.6 | Mensintesis bukti dan kompromi |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | Menghasilkan memo keputusan akhir |
Ini ialah polisi perutean contoh, bukan penarafan penanda aras.
Model yang betul untuk ejen anda bergantung pada:
- kerumitan tugas
- panjang konteks yang diperlukan
- penggunaan alat
- keperluan output berstruktur
- kependaman
- kebolehpercayaan
- kos token
- kualiti output
- keputusan penilaian khusus aplikasi
Satu peraturan yang berguna ialah:
Pilih model untuk kerja yang dilakukan ejen, bukan hanya berdasarkan pembekal asalnya.
Model Manakah Yang Patut Digunakan Setiap Ejen CrewAI?
Untuk contoh ini, penugasan model mengikuti strategi mudah kos berbanding keupayaan.
Researcher: Gemini 3.7 Flash
Penyelidikan sering melibatkan memproses jumlah maklumat yang agak besar dan menghasilkan hasil perantaraan yang padat.
Model pantas oleh itu berguna untuk tugas penyelidikan berisipadu tinggi.
"researcher": "gemini-3.7-flash"
Analyst: Claude Opus 5
Penganalisis mempunyai peranan yang lebih sempit tetapi lebih intensif penaakulan. Ia menerima output penyelidikan dan menukarnya kepada cadangan.
"analyst": "claude-opus-5"
Writer: GPT-5.6
Ejen terakhir menukar penyelidikan dan analisis kepada memo keputusan untuk pembangun.
"writer": "gpt-5.6"
Bahagian penting bukanlah tepat tiga tugasan ini. Aplikasi anda harus menilai model calon terhadap tugas perwakilan sebelum menetapkan polisi perutean.
Apa Yang Diperlukan Sebelum Bermula?
Anda perlukan:
- Python 3.10+
- CrewAI
- Keserasian SDK Python OpenAI
python-dotenv- Kunci API CometAPI
- ID model yang anda ingin gunakan
Integrasi Python CometAPI semasa menyokong API serasi OpenAI, dan pakej Python CometAPI rasmi mendokumenkan COMETAPI_KEY dan COMETAPI_BASE_URL sebagai pilihan konfigurasi berasaskan persekitaran.
Titik akhir standard ialah:
https://api.cometapi.com/v1
Sebelum pengeluaran, sahkan bahawa ID model yang anda pilih sedang tersedia dan menyokong titik akhir serta parameter yang diperlukan oleh beban kerja CrewAI anda. Katalog model dan harga boleh berubah.
Bagaimana Memasang CrewAI dan Kebergantungan?
Cipta persekitaran Python baharu:
python -m venv .venv
Aktifkan:
source .venv/bin/activate
Pada Windows:
.venv\Scripts\Activate.ps1
Kemudian pasang kebergantungan:
pip install "crewai[openai]" openai python-dotenv
Menggunakan openai secara eksplisit adalah disengajakan kerana pelaksanaan fallback di bawah mengimport kelas pengecualian SDK OpenAI secara langsung.
Untuk produksi, kunci versi yang anda uji daripada bergantung tanpa batas pada versi terkini mengambang.
Sebagai contoh:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
Lapisan LLM CrewAI sedang berkembang aktif, jadi pembina dan konfigurasi pembekal yang tepat harus diperiksa terhadap versi CrewAI yang digunakan oleh aplikasi anda. Dokumentasi CrewAI semasa menyokong mengkonfigurasi LLM dengan base_url tersuai dan kunci API.
Bagaimana Mengkonfigurasi Kunci API CometAPI?
Cipta fail .env:
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Muatkan nilai ini dalam Python:
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",)
Jangan sekali-kali commit .env ke Git.
Tambahkan ke .gitignore:
.env.venv/__pycache__/
Kunci API harus kekal sebagai kelayakan sisi pelayan. Panduan mula pantas CometAPI semasa juga mengesyorkan menyimpan kunci dalam pembolehubah persekitaran dan bukannya dalam kod sumber.
Bagaimana Menyambungkan CrewAI ke CometAPI?
Objek LLM CrewAI boleh menerima nama model, kunci API, dan URL asas tersuai.
Cipta pembantu:
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, )
Ini lebih baik daripada memasukkan konfigurasi yang sama secara berasingan dalam setiap ejen.
Setiap ejen kini hanya memerlukan ID model:
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
Mengapa Tetapkan max_retries=0?
Sebabnya ialah kawalan fallback.
Jika klien LLM asas secara automatik cuba semula dan aplikasi anda juga melaksanakan fallback, satu kegagalan boleh menjadi beberapa permintaan tersembunyi sebelum logik fallback dilaksanakan.
Untuk tutorial dengan perutean eksplisit, lebih bersih untuk membiarkan aplikasi memutuskan bila untuk cuba semula atau menukar model.
Bagaimana Mendefinisikan Polisi Perutean Model?
Kekalkan perutean di luar prompt anda:
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",}
Ini mewujudkan sempadan konfigurasi yang jelas.
Anda kemudian boleh memindahkan pemetaan yang sama ke:
- konfigurasi persekitaran
- YAML
- JSON
- pangkalan data
- feature flag
- perkhidmatan perutean model dalaman
tanpa menulis semula prompt ejen.
Bagaimana Membina Tiga Ejen CrewAI?
Cipta satu objek LLM per ejen.
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
Penugasan model kini benar-benar bebas daripada definisi peranan ejen.
Itulah yang menjadikan perutean model praktikal.
Bagaimana Menghubungkan Ejen dengan Tugas Berjujukan?
Cipta tiga tugas:
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
Rantaian pergantungan ialah:
Topik ↓Penyelidikan ↓Analisis ↓Memo akhir
Penganalisis menerima output tugas penyelidikan, manakala penulis menerima kedua-dua konteks penyelidikan dan analisis.
Bagaimana Membina Crew?
Gabungkan ejen dan tugas:
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, )
Kini perutean model didorong sepenuhnya oleh konfigurasi.
Menukar:
"researcher": "gemini-3.7-flash"
kepada model lain yang disokong tidak memerlukan perubahan pada prompt atau definisi tugas penyelidikan.
Bagaimana Sepatutnya Fallback Model CrewAI Berfungsi?
Di sinilah pelaksanaan berorientasikan produksi memerlukan lebih perhatian.
Kesilapan biasa ialah:
Apa-apa ralat ↓Tukar model
Itu terlalu agresif.
Sebagai contoh, ralat ini secara umum tidak sepatutnya mencetuskan fallback model:
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Menukar model tidak akan membetulkan kunci API tidak sah atau permintaan yang cacat.
Fallback lebih sesuai untuk kegagalan sementara seperti:
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
Polisi fallback oleh itu sepatutnya:
Cuba semula atau tukar model hanya untuk kegagalan sementara yang terhad dan hanya apabila model fallback menyokong kontrak permintaan yang sama.
Bagaimana Mengesan Ralat Yang Boleh Dicuba Semula?
Anda boleh menggunakan kelas ralat SDK OpenAI:
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
Ini dengan sengaja mengecualikan ralat konfigurasi peringkat 400 selain 408 dan 429.
Patutkah Anda Mencuba Semula Seluruh Crew atau Hanya Ejen Yang Gagal?
Terdapat dua strategi fallback berbeza.
Fallback peringkat kru
Pelaksanaan paling ringkas ialah:
Mulakan crew ↓gagal ↓ubah perutean ↓jalankan crew lagi
Ini mudah difahami, tetapi ia boleh mengulangi tugas yang telah selesai.
Sebagai contoh:
Penyelidikan → selesaiAnalisis → selesaiPenulis → gagal
Cuba semula kickoff() penuh boleh melaksanakan:
Penyelidikan → lagiAnalisis → lagiPenulis → fallback
Itu meningkatkan:
- penggunaan token
- kependaman
- kos API
- potensi kesan sampingan
Pemulihan peringkat tugas
Aliran kerja produksi sepatutnya sebaliknya membuat checkpoint kerja yang telah selesai:
Penyelidikan ↓checkpoint ↓Analisis ↓checkpoint ↓Penulis gagal ↓cuba semula penulis dengan fallback
CrewAI pada masa ini menyediakan checkpointing yang menyimpan keadaan pelaksanaan dan membolehkan run disambung semula selepas kegagalan. Tingkah laku checkpoint yang didokumenkan melangkau tugas yang telah selesai dan meneruskan kerja hiliran dari keadaan yang disimpan.
Ini ialah seni bina yang lebih baik untuk aliran kerja yang mahal atau menghasilkan kesan sampingan.
Bagaimana Menambah Checkpointing CrewAI?
Untuk aliran kerja produksi, aktifkan checkpointing pada kru:
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
Sistem checkpointing CrewAI boleh mengekalkan keadaan pelaksanaan selepas penyiapan tugas dan memulihkan kru daripada checkpoint.
Sebagai contoh, run yang dipulihkan boleh menggunakan:
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
Konfigurasi checkpoint yang tepat harus mengikuti versi CrewAI yang digunakan oleh projek anda.
Titik seni bina penting ialah:
Checkpoint dahulu, fallback kemudian.
Ini menghalang kegagalan model sementara daripada memaksa kerja yang telah selesai yang mahal dijalankan semula.
Bagaimana Melaksanakan Fallback Terhad Yang Ringkas?
Untuk tutorial, anda masih boleh menunjukkan fallback peringkat kru yang ringkas.
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
Perhatikan perbezaan penting:
Ini tidak mendakwa bahawa ejen yang gagal telah dikenal pasti.
Ia ialah strategi fallback peringkat kru yang terhad.
Untuk aliran kerja kecil tanpa keadaan, ini mungkin boleh diterima. Untuk aliran kerja produksi dengan penyelidikan mahal, alat, atau kesan sampingan, gunakan pemulihan berasaskan checkpoint.
Bagaimana Menjejak Penggunaan Token CrewAI?
Penjejakan penggunaan harus menjadi sebahagian daripada lapisan perutean, bukan selepas fikiran.
Pada akhir run, periksa hasil 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)
Medan penggunaan yang tepat tersedia boleh bergantung pada versi CrewAI dan laluan pelaksanaan, jadi anggap objek hasil yang dikembalikan sebagai sumber kebenaran untuk versi yang anda gunakan.
Rekod penggunaan produksi sebaik-baiknya mengandungi:
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
Ini membolehkan anda menjawab soalan seperti:
Ejen mana yang menggunakan bajet paling banyak?
Seberapa kerap penganalisis fallback?
Model mana mempunyai kependaman tertinggi?
Berapa banyak kos setiap aliran kerja?
Bagaimana Mengawal Kos Pada Peringkat Ejen?
Perutean berbilang model paling berguna apabila ia mencerminkan perbezaan beban kerja sebenar.
Sebagai contoh:
Researcher→ isipadu tinggi→ model kos lebih rendahAnalyst→ isipadu rendah→ model penaakulan lebih kuatWriter→ isipadu sederhana→ model pengeluaran serba guna
Anda juga boleh mengehadkan kos melalui konfigurasi ejen.
Sebagai contoh:
max_iter=3
menghadkan gelung iterasi ejen. Ia bukan harus ditafsirkan sebagai had keras tepat tiga panggilan API atau tiga bajet token.
Kawalan tambahan termasuk:
- mengehadkan konteks tugas
- merumuskan output perantaraan
- menyimpan cache penyelidikan berulang
- mengehadkan saiz input maksimum
- mengehadkan output token maksimum di mana disokong
- menyekat panggilan alat
- menetapkan bajet per pengguna
- menetapkan bajet per aliran kerja
- menjejaki kekerapan fallback
Bagaimana Mengesahkan Model Sebelum Pengeluaran?
Jangan kekalkan ID model selama-lamanya.
Sebuah model boleh menjadi:
- tidak tersedia
- dinamakan semula
- ditamatkan
- dihadkan
- berubah keupayaan
- berubah harga
- tidak serasi dengan parameter yang digunakan aplikasi anda
CometAPI menyediakan titik akhir katalog model yang boleh ditanya secara programatik, manakala direktori model awamnya boleh digunakan untuk penemuan model oleh manusia.
Semakan pengeluaran boleh kelihatan seperti:
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Kemudian sahkan bahawa ID model yang anda konfigurasikan wujud sebelum pengeluaran.
Sebagai contoh, proses CI anda boleh mengesahkan:
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
Jangan jadikan pemeriksaan ketersediaan sebagai pengganti ujian aplikasi. Model tersenarai dalam katalog tidak bermakna setiap parameter, alat, atau format output yang digunakan oleh ejen CrewAI anda disokong.
Bagaimana Rupa Contoh CrewAI Lengkap?
Berikut ialah pelaksanaan disatukan:
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()
Penambahbaikan penting berbanding versi asal ialah kod tidak lagi menyiratkan secara salah bahawa pengecualian mengenal pasti ejen yang gagal dengan tepat.
Ia secara eksplisit ialah pelaksanaan fallback peringkat kru yang terhad.
Untuk produksi, gabungkan polisi perutean yang sama dengan checkpointing CrewAI.
Bagaimana Menjalankan Aliran Kerja CrewAI?
Simpan fail sebagai:
crewai_multi_model.py
Kemudian jalankan:
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Respons berjaya akan mengandungi maklumat seperti:
{ "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>"}
Respons dan nilai penggunaan yang tepat bergantung pada input, tingkah laku model, versi CrewAI, dan laluan pelaksanaan.
Jika ralat boleh dicuba semula mengaktifkan laluan fallback, objek selected_models menunjukkan laluan yang digunakan untuk pelaksanaan kru tersebut.
Bagaimana Mereka Bentuk Perutean Model Produksi?
Polisi perutean produksi harus mempertimbangkan lebih daripada kualiti model.
Fungsi keputusan yang berguna ialah:
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Anda boleh melaksanakannya pada beberapa peringkat.
Perutean berasaskan kos
Tugas ringkas → model ekonomikTugas kompleks → model premium
Perutean berasaskan kependaman
Permintaan interaktif → model pantasAliran kerja latar → model berkualiti lebih tinggi
Perutean berasaskan kebolehpercayaan
Model utama ↓kegagalan sementara ↓model fallback
Perutean berasaskan tugas
Penyelidikan → Model AAnalisis → Model BPenulisan → Model CKod → Model D
Pendekatan terakhir amat semula jadi untuk CrewAI kerana rangka kerja sudah memberikan setiap ejen peranan yang berbeza.
Bagaimana Menjadikan Fallback Selamat?
Sistem fallback yang mantap harus menguatkuasakan empat peraturan.
Jangan fallback pada ralat pengesahan
Jika kunci API tidak sah:
401
menukar model tidak akan membetulkan masalah.
Jangan fallback pada permintaan cacat
Jika permintaan tidak sah:
400422
betulkan permintaan sebaliknya.
Jangan fallback tanpa batas
Tetapkan had keras:
MAX_FALLBACK_ATTEMPTS = 2
Sistem fallback tanpa had boleh menjadi gelung cuba semula yang mahal.
Jadikan model fallback serasi permintaan
Model fallback mesti menyokong ciri yang diperlukan ejen anda.
Sebagai contoh, jika ejen utama memerlukan alat tertentu atau tingkah laku output berstruktur, fallback mesti menyokong kontrak yang sama.
Serasi OpenAI tidak bermakna serasi ciri.
Apakah Ralat Paling Biasa CrewAI + CometAPI?
| Gejala | Punca Mungkin | Baiki |
|---|---|---|
| 401 Unauthorized | Kunci API tidak sah/hilang | Semak COMETAPI_KEY; jangan fallback |
| 400 Bad Request | Parameter permintaan tidak sah | Betulkan permintaan |
| 404 Model Not Found | ID model lapuk | Semak katalog model semasa |
| 408 Timeout | Tamat masa permintaan sementara | Cuba semula dalam polisi terhad |
| 429 Rate Limited | Terlalu banyak permintaan | Undur dan cuba semula |
| 500–504 | Kegagalan pelayan/gateway | Gunakan fallback terhad |
| Ejen berulang kali cuba semula | Cuba semula tersembunyi SDK | Kawal max_retries |
| Tugas selesai berjalan lagi | Cuba semula kru penuh | Gunakan pemulihan berasaskan checkpoint |
| Model berbeza berkelakuan berbeza | Keupayaan model berbeza | Uji setiap model secara bebas |
| Ralat pembina CrewAI tidak dijangka | Ketidakpadanan versi | Pin dan sahkan versi CrewAI |
Bagaimana Memisahkan Ralat CrewAI daripada Ralat Model?
Perbezaan ini penting semasa penyahpepijatan.
Ralat konfigurasi
Kunci API hilangID model tidak sahURL asas tidak sahParameter tidak disokong
Ini harus gagal dengan cepat.
Ralat pembekal/API
401403404429500503
Ini memerlukan pengendalian yang berbeza bergantung pada status.
Ralat aplikasi
Output ejen tidak sahAlat memulangkan data cacatKonteks tugas hilangKesan sampingan gagal
Ini tidak semestinya diselesaikan dengan menukar model.
Sistem ejen matang oleh itu harus mempunyai pengendalian berasingan untuk:
konfigurasi ↓pengangkutan API ↓pelaksanaan model ↓logik ejen ↓pelaksanaan alat ↓kesan sampingan aplikasi
Ini jauh lebih selamat daripada:
except Exception: use_fallback()
Bagaimana Melindungi Kesan Sampingan Luaran?
Fallback menjadi jauh lebih rumit apabila ejen melakukan lebih daripada menjana teks.
Sebagai contoh, bayangkan ejen yang:
- mencipta rekod pangkalan data
- menghantar e-mel
- memanggil API luaran
- mengemas kini CRM
Jika model tamat masa selepas tindakan luaran berjaya, menjalankan semula seluruh kru boleh menduplikasi tindakan tersebut.
Gunakan:
- kunci idempotensi
- checkpoint tugas
- sempadan transaksi
- ID pelaksanaan
- keadaan tugas yang berterusan
- pengesahan kesan sampingan eksplisit
Sebagai contoh:
job_id = crew_run_123task_id = writer_456
Simpan pengecam ini dengan operasi luaran supaya cuba semula boleh menentukan sama ada operasi sudah berlaku.
Bagaimana Memantau Aliran Kerja CrewAI Berbilang Model?
Sekurang-kurangnya, logkan:
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
Jangan log:
Kunci APIkelayakan peribayadata sensitif penuhprompt sensitif peribayadata pengguna peribayakeluaran model tidak ditapis
Untuk setiap model, pantau:
Kebolehpercayaan
kadar kejayaankadar tamat masakadar 5xxkadar fallback
Prestasi
kependaman p50kependaman p95kependaman p99
Kos
token masuktoken keluarkos per tugaskos per aliran kerja lengkap
Kualiti
kadar kejayaan tugaspenilaian manusia kesahihan output berstrukturkejayaan panggilan alat
Ini menukar perutean model daripada keutamaan kod keras kepada sistem kejuruteraan yang boleh diperhatikan.
Bagaimana Memilih Antara API Pembekal Terus dan CometAPI?
Pilihan bergantung pada seni bina anda.
| Seni bina | Kelayakan | Pertukaran Model | Integrasi Pembekal | Perutean Berpusat |
|---|---|---|---|---|
| API pembekal terus | Pelbagai | Tersuai | Tinggi | Tidak |
| Satu pembekal | Satu | Terhad | Rendah | Terhad |
| CrewAI + CometAPI | Satu kelayakan CometAPI | Berasaskan ID | Lebih rendah | Ya |
Jika aplikasi anda hanya memerlukan satu pembekal dan ciri natifnya, integrasi terus boleh jadi wajar.
Jika aplikasi CrewAI anda memerlukan model daripada beberapa pembekal dan anda mahu satu lapisan capaian, CometAPI menjadi lebih menarik.
Titik penting ialah CometAPI tidak menggantikan CrewAI.
Sebaliknya:
CrewAIOrkestrasi ejen ↓CometAPICapaian model ↓Pelbagai model
Setiap lapisan mempunyai tanggungjawab berbeza.
Bagaimana Seni Bina Ini Berskala?
Setelah polisi perutean dipisahkan daripada definisi ejen, menambah model lain tidak memerlukan pembinaan semula keseluruhan aplikasi.
Sebagai contoh:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
Seni bina yang sama kemudian boleh menyokong:
Ejen penyelidikanEjen analisisEjen pengkodanEjen semakanEjen penulisanEjen semak fakta
Setiap ejen boleh mempunyai model berbeza sambil berkongsi lapisan capaian CometAPI yang sama.
Langkah seterusnya ialah menjadikan perutean dinamik.
Daripada:
"analyst": "claude-opus-5"
anda akhirnya boleh gunakan:
select_model( task="analysis", budget=budget, latency_target=latency_target,)
Sistem perutean kemudian boleh memilih daripada model yang diluluskan berdasarkan keperluan aplikasi.
Apakah Seni Bina Produksi Terbaik untuk CrewAI + CometAPI?
Untuk aliran kerja kecil:
Input pengguna ↓CrewAI ↓CometAPI ↓Model
Untuk produksi:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
Komponen produksi utama ialah:
- Senarai benarkan model
- Perutean per ejen
- Cuba semula terhad
- Checkpoint tugas
- Penjejakan penggunaan
- Kawalan kos
- Ujian keserasian model
- Kebolehcerapan
- Kesan sampingan idempoten
Seni bina itu jauh lebih mantap daripada sekadar menambah try/except di sekitar crew.kickoff().
Satu Kunci CometAPI, Model Berbeza, Peranan Ejen Lebih Jelas
Cara paling berguna untuk memikirkan CrewAI dan CometAPI bersama ialah sebagai dua lapisan saling melengkapi.
CrewAI mendefinisikan apa yang ejen lakukan.
CometAPI mendefinisikan bagaimana ejen tersebut mengakses model.
Pemisahan itu membolehkan anda menugaskan model pantas kepada ejen penyelidikan berisipadu tinggi, model penaakulan lebih kukuh kepada ejen analisis, dan model serba guna kepada ejen penulis akhir tanpa mengekalkan integrasi pembekal berasingan dalam aliran kerja.
Pelaksanaan paling ringkas menggunakan satu kunci CometAPI dan satu URL asas serasi OpenAI:
https://api.cometapi.com/v1
Untuk produksi, bawa seni bina selangkah lebih jauh: kekalkan perutean model dalam konfigurasi, sahkan ketersediaan model sebelum pengeluaran, gunakan fallback terhad hanya untuk kegagalan sementara, checkpoint tugas yang selesai, dan rekod metadata model serta penggunaan untuk setiap run.
Itu memberikan corak yang jauh lebih tahan lama daripada sekadar menyambungkan CrewAI kepada satu LLM:
CrewAI mengorkestrasi ejen. CometAPI memusatkan capaian model. ID model mengawal perutean. Checkpoint melindungi kerja selesai. Penjejakan penggunaan mengawal kos.
Soalan Lazim
Bolehkah CrewAI menggunakan pelbagai model AI dalam kru yang sama?
Ya. Tetapkan konfigurasi LLM berbeza kepada setiap ejen CrewAI. Setiap konfigurasi boleh menentukan modelnya sendiri sambil menggunakan kunci API dan URL asas CometAPI yang sama.
Bolehkah CrewAI menyambung ke API serasi OpenAI?
Ya. Konfigurasi LLM CrewAI menyokong base_url tersuai dan kunci API untuk titik akhir serasi OpenAI.
Dengan CometAPI, URL asas ialah:
https://api.cometapi.com/v1
Adakah saya memerlukan kunci API berasingan untuk GPT, Claude, dan Gemini?
Apabila mengakses model ini melalui CometAPI, aplikasi boleh menggunakan kelayakan dan titik akhir CometAPI dan bukannya melaksanakan kelayakan pembekal berasingan dalam setiap ejen CrewAI.
Adakah satu kunci API bermakna model mempunyai keupayaan yang sama?
Tidak. Antara muka API boleh disatukan sementara keupayaan model kekal berbeza. Tingkap konteks, sokongan alat, parameter, tingkah laku output, kependaman, dan harga boleh berbeza mengikut model.
Patutkah saya mencuba semula keseluruhan aliran kerja CrewAI apabila satu model gagal?
Hanya untuk aliran kerja ringkas tanpa keadaan. Cuba semula keseluruhan kru boleh mengulangi tugas selesai dan meningkatkan kos. Untuk aliran kerja produksi, checkpoint tugas selesai dan sambung dari bahagian yang gagal di mana praktikal.
Fungsi checkpointing CrewAI semasa direka untuk memelihara keadaan pelaksanaan dan menyambung semula selepas kegagalan.
Patutkah setiap pengecualian CrewAI mencetuskan fallback model?
Tidak. Pengesahan, permintaan cacat, ID model tidak sah, dan parameter tidak disokong lazimnya memerlukan perubahan konfigurasi dan bukannya model berbeza.
Fallback lebih baik dikhaskan untuk kegagalan sementara terhad seperti tamat masa, had kadar, dan respons 5xx sementara.
Bagaimana saya menjejak kos setiap ejen CrewAI?
Rekod nama ejen, ID model, penggunaan token, kependaman, status pelaksanaan, dan maklumat fallback untuk setiap tugas. Gunakan data tersebut untuk mengira kos per ejen dan aliran kerja.
Bolehkah saya menukar model yang ditugaskan kepada ejen secara dinamik?
Ya. Kekalkan ID model dalam konfigurasi perutean dan bukannya memasukkannya terus dalam definisi ejen. Aplikasi anda kemudian boleh memilih model berdasarkan kos, kependaman, jenis tugas, atau ketersediaan.
Adakah CometAPI pengganti CrewAI?
Tidak. Mereka beroperasi pada lapisan berbeza. CrewAI mengorkestrasi ejen dan tugas, sementara CometAPI menyediakan lapisan capaian model bersepadu.
Di mana saya boleh mencari model CometAPI semasa?
Gunakan CometAPI model directory untuk penemuan model oleh manusia dan API model untuk pengesahan programatik. Halaman mula pantas CometAPI semasa menyenaraikan 500+ model merentas kategori teks, imej, video, dan audio.
