Phase 13: Tools & Protocols

Appels parallèles et diffusion en continu avec des outils

Type: Build

Languages: Python (stdlib, thread pool + streaming harness)

Prerequisites: Phase 13 · 02 (function calling deep dive)

Time: ~75 minutes

Objectifs d'apprentissage

• Expliquez pourquoi .parallel_tool_calls: trueIl existe et quand désactiver.
• Corréler les blocs d'arguments en streaming à l'identifiant de l'appel d'outil droit pendant le ventilateur parallèle.
• Rassembler partiellement argumentsles chaînes dans JSON complet sans parser tôt.
• Exécutez un indice météorologique de trois villes qui montre la latence séquentielle par rapport à la latence parallèle.

Le problème

Sans appels parallèles, un agent répondant à " quel est le temps à Bengaluru, Tokyo et Zurich " fait ceci:

Copyuser -> LLM
LLM -> call get_weather(Bengaluru)
host -> run executor, reply with result
LLM -> call get_weather(Tokyo)
host -> run executor, reply with result
LLM -> call get_weather(Zurich)
host -> run executor, reply with result
LLM -> final text answer

Trois voyages aller-retour de la LLM, chacun payant également la latence de l'exécuteur.

Avec des appels parallèles:

Copyuser -> LLM
LLM -> call get_weather(Bengaluru); call get_weather(Tokyo); call get_weather(Zurich)
host -> run all three executors concurrently, reply with three results
LLM -> final text answer

Un voyage aller-retour LLM. Le temps d'exécution est le maximum des trois, pas la somme. Les critères de référence de production sur OpenAI, Anthropic et Gemini montrent une réduction de 60 à 70% de la vitesse de la paroi sur les charges de travail de ventilateur.

Le prix est la complexité de la corrélation. Lorsque les trois appels sont terminés hors d'ordre, vos résultats doivent être conformes tool_call_idEn effet, les résultats de l'exécution de la requête sont en partie des éléments de l'argument JSON, et les résultats sont en cours de diffusion.

Le concept

Permetant le parallèle

• OpenAI. parallel_tool_calls: trueactivé par défaut.falsePour forcer la série.
• Anthropic.Parallèlement via disable_parallel_tool_use: false(par défaut sur Claude 3.5 et plus).truePour la série.
• Gemini.Toujours en parallèle .tool_config.function_calling_config.mode = "AUTO"Laissez le modèle décider.

Désactiver le parallèle lorsque les outils ont des dépendances d'ordre (create_fileAlors ...write_file), lorsque la sortie d'un appel informe la sortie d'un autre ou lorsque le limitateur de fréquence ne peut pas gérer le ventilateur.

Corrélation de l'ID

Chaque appel que le modèle émet a unidChaque résultat que l'hôte renvoie doit contenir le même identifiant.

• OpenAI. tool_call_idsur chaque message de rôle d'outil.
• Anthropic. tool_use_idsur chacun tool_resultLe bloc.
• Gemini. idsur chacun functionResponse(Gemini 3 et plus; Gemini 2 correspondant par nom qui a rompu pour les appels parallèles du même nom).

Exécution d'appels simultanément

L'hôte exécute l'exécuteur de chaque appel sur son propre fil, coroutine ou travailleur à distance.asyncio.gatherL'ordre de réalisation est imprévisible l'identifiant est l'identifiant.

Un bug commun: répondre avec des résultats dans l'ordre de la liste d'appels au lieu de l'ordre de finalisation.tool_call_id, mais si un résultat est abandonné ou dupliqué, la soumission hors ordre rend le débogage plus difficile.

Appels d' outils de diffusion

Quand le modèle est en streaming,argumentsTrois flux séparés de pièces pour trois appels parallèles interférent sur le fil.

Forme par fournisseur:

• OpenAI.Chaque morceau estchoices[0].delta.tool_calls[i].function.argumentsLa pièce porteindex(position dans la liste des appels). Vous accumulez par index, lire idquand il apparaît pour la première fois, et parser JSON quand finish_reason = "tool_calls"- Je suis désolé .
• Anthropic.Les événements de streaming sont message_start, puis un .content_block_startpar bloc avec type tool_use(contenant un identifiant, un nom, une entrée vide). content_block_deltaévénements de transport input_json_deltaDes morceaux.content_block_stopferme chaque bloc.
• Gemini. streamFunctionCallArguments(Gemini 3 et plus) émet des morceaux avec un functionCallIdAvant Gemini 3, le streaming renvoyait un appel complet à la fois.

JSON partiel et le piège de partage précoce

Vous ne pouvez pas analyser .argumentsJSON partiel comme {"city": "BengLa bonne passerelle est le signal de fin d'appel du fournisseur: OpenAI finish_reason = "tool_calls", Anthropic's content_block_stopC'est seulement après que tu es arrivé.json.loadsUne approche plus robuste utilise un parseur JSON incrémentiel qui produit des événements à mesure que la structure est complétée; le guide de streaming d'OpenAI le recommande pour UX qui montre un indicateur de "pensation" en direct. Le comptage de brace est peu fiable en tant que test de complétude (les braces à l'intérieur des chaînes citées ou le contenu échappé provoquent des faux positifs) et ne doit être utilisé que comme un heuristique de débogage informel.

Résultats hors commande

Copycall_A: fast API, returns first
call_B: slow API, returns second
call_C: median API, returns third

La réponse de l'hôte doit toujours citer les identifiants:

Copy[{role: "tool", tool_call_id: "call_A", content: ...},
 {role: "tool", tool_call_id: "call_B", content: ...},
 {role: "tool", tool_call_id: "call_C", content: ...}]

L'ordre dans la réponse n'a pas d'importance pour la précision sur OpenAI ou Anthropic. Gemini accepte toute commande tant que les identifiants correspondent.

Indice de référence: séquentiel par rapport parallèle

Le harnais dans code/main.pyLa séquence le fonctionne en 1800 ms total. Le parallèle le fonctionne en max ((400, 600, 800) = 800 ms. La différence est constante, pas proportionnelle, donc les économies augmentent avec le nombre d'outils.

Avertissement dans le monde réel: les appels parallèles stressent les API en aval. Une extension de 10 voies vers un service limité de taux échouera. La phase 13 · 17 couvre la contrainte au niveau du gateway; une nouvelle sémantique d'essai est prévue pour une phase future.

Retour de ventilateur-horloge de mur

Si le modèle lui-même diffuse des flux, vous pouvez commencer à exécuter dès que les arguments d'un appel sont complets, plutôt que d'attendre que tous les appels soient terminés. Il s'agit d'une optimisation des documents OpenAI mais pas tous les SDK sont exposés.

Utilisez-le

code/main.pyLa première fonctionne trois appels météorologiques simulés en séquence et en parallèle en utilisantconcurrent.futures.ThreadPoolExecutorLa seconde moitié reproduit une fausse réponse de streaming argumentspour trois appels parallèles interleavés sur un flux et les réassemble par id avec StreamAccumulatorPas de Master, pas de réseau, juste la logique de réassemblage.

À quoi regarder:

• Le temporiseur séquentiel atteint 1,8 seconde. Le temporiseur parallèle atteint 0,8 seconde sur les mêmes latences fausses.
• L'accumulateur gère les morceaux arrivant hors ordre en tamponnant par ID et en analysant seulement lorsque le JSON de chaque appel est complet.
• L'exécuteur démarre dès que les arguments d'un ID sont définis, pas après la fin de tous les flux.

La faire partir

Cette leçon produit outputs/skill-parallel-call-safety-check.md. En raison d'un registre des outils, les audits des compétences permettent de comparer les outils qui sont sûrs, qui ont des dépendances de commande et qui dépasseraient les limites des taux en aval renvoyer un registre révisé avec un outil parallel_safeLes drapeaux.

Exercices

1. On court .code/main.pyConfirmer que le rapport parallèle-sequentiel est approximatif max/sum(les courbes réelles deviennent légèrement de l'idéal en raison de la planification des fils, de la sérialisation et du coût de l'harmonisation).

1. Élargir l'accumulateur pour gérer un cas "appel a été annulé en milieu de cours de route" en laissant tomber son tampon et émettant un cancelledQuel fournisseur a explicitement documenté cette affaire ?content_block_stopLa sémantique et les ouvertures d'AI finish_reason: "length"Le comportement.

1. Remplacez la piscine de fil par asyncio.gatherVous devriez voir de petites victoires sur asynchrony en raison du coût de commutation de contexte inférieur, mais seulement si les exécutants font de vraies opérations d'entrée/sortie.

1. Choisissez deux outils qui ne doivent pas être parallélisés (p. ex. create_fileAlors ...write_file ordering_dependencyIl s'agit de la machine minimale pour la planification consciente de la dépendance, que formalise une future phase d'ingénierie des agents.

1. Lisez la section d'appels parallèles à la fonction d'OpenAI et celle d'Anthropic disable_parallel_tool_useDocuments. Identifiez le type d'outil réel où Anthropic recommande de désactiver le parallélisme.

Les termes clés

Pour en savoir plus

• OpenAI — Parallel function calling comportement par défaut et le drapeau de désactivation
• Anthropic — Parallel tool use disable_parallel_tool_useet le partage des résultats
• Google — Gemini function calling parallel section Appels parallèles liés à l'id de Gémeaux 3
• OpenAI — Streaming responses with tools Rassemblement de arguments en morceaux pour les flux OpenAI
• Anthropic — Streaming messages content_block_deltaavec input_json_delta

This free lesson is part of the AI Engineering from Scratch curriculum. Read the full explanation, run the lesson code, and verify the result in the interactive reader or from the repository source.

Browse the complete course catalog or open this lesson on GitHub.