Gérer les appels d’outils dans les réponses diffusées
Analysez les réponses diffusées contenant des arguments d’appel de fonction reçus jeton par jeton, mettez les fragments JSON en mémoire tampon et ne déclenchez l’exécution de l’outil que lorsque l’appel est complet.
Gérer les appels d’outils dans les réponses diffusées est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.
Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.
Tool Calls Arrive Differently in Streams
When an LLM decides to call a function, the response structure changes. Instead of a content string, the delta contains a tool_calls array. But in a streamed response, the function call arguments arrive token by token as a partial JSON string — you do not receive a complete JSON object in a single chunk. You must buffer these fragments and reassemble the complete JSON before you can parse and execute the tool call.
# In a non-streaming response, tool call is complete:
# choice.message.tool_calls[0].function.arguments = '{"city": "Paris"}'
# In a streaming response, arguments arrive in pieces:
# chunk 1: delta.tool_calls[0].function.arguments = '{'
# chunk 2: delta.tool_calls[0].function.arguments = '"city"'
# chunk 3: delta.tool_calls[0].function.arguments = ': "'
# chunk 4: delta.tool_calls[0].function.arguments = 'Paris'
# chunk 5: delta.tool_calls[0].function.arguments = '"}'
# You must concatenate these before JSON.parse can workDetecting a Tool Call in the Stream
Check each chunk's finish_reason to know when to expect tool calls. When finish_reason is 'tool_calls', the model has decided to call a function and the stream is ending. When finish_reason is 'stop', the model produced a normal text response. While streaming, check whether chunk.choices[0].delta.tool_calls is non-None to identify tool call argument fragments.
async def detect_stream_type(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=tools,
stream=True,
)
response_type = 'text'
async for chunk in stream:
choice = chunk.choices[0]
if choice.delta.tool_calls: # tool call fragment arriving
response_type = 'tool_call'
if choice.finish_reason == 'tool_calls':
print('Model wants to call a function')
elif choice.finish_reason == 'stop':
print('Normal text response')
return response_typeBuffering Tool Call Arguments
Use a dictionary keyed by tool call index to accumulate argument fragments from each chunk. The index identifies which tool call is being streamed — a model may call multiple functions in one response. For each chunk with a non-null tool_calls delta, append the argument fragment to the appropriate buffer entry keyed by the tool call index.
from collections import defaultdict
async def collect_streamed_tool_calls(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=tools,
stream=True,
)
tool_call_buffers = defaultdict(lambda: {'name': '', 'id': '', 'arguments': ''})
text_buffer = ''
async for chunk in stream:
delta = chunk.choices[0].delta
if delta.content: # text content
text_buffer += delta.content
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if tc.id:
tool_call_buffers[idx]['id'] = tc.id
if tc.function.name:
tool_call_buffers[idx]['name'] += tc.function.name
if tc.function.arguments:
tool_call_buffers[idx]['arguments'] += tc.function.arguments
return text_buffer, dict(tool_call_buffers)Parsing and Executing Tool Calls
After the stream ends and you have complete argument strings, parse each one with json.loads and dispatch to the appropriate Python function. Execute tool calls in the order they were requested (or in parallel if they are independent), then format the results as tool response messages to send back in the follow-up API call.
import json
# Example tool registry
tools_registry = {
'get_weather': lambda city, unit='celsius': {'temp': 22, 'desc': 'sunny', 'city': city},
'search_docs': lambda query, top_k=3: [{'title': 'Doc 1', 'snippet': 'Relevant info...'}],
}
def execute_tool_calls(tool_call_buffers: dict) -> list[dict]:
tool_messages = []
for idx in sorted(tool_call_buffers.keys()):
tc = tool_call_buffers[idx]
func_name = tc['name']
args = json.loads(tc['arguments'])
if func_name in tools_registry:
result = tools_registry[func_name](**args)
else:
result = {'error': f'Unknown function: {func_name}'}
tool_messages.append({
'role': 'tool',
'tool_call_id': tc['id'],
'content': json.dumps(result),
})
return tool_messagesThe Full Streaming Tool Call Loop
The complete streaming tool call flow requires a multi-turn conversation loop. The first request may return tool calls; you execute them and append the results to the message history; a second request returns the final text answer. This loop may iterate multiple times if the model chooses to make additional tool calls based on the results of earlier ones.
async def streaming_agent_loop(initial_messages, tools):
messages = list(initial_messages)
max_iterations = 5
for iteration in range(max_iterations):
text, tool_calls = await collect_streamed_tool_calls(messages, tools)
if tool_calls:
# Append assistant message with tool calls
assistant_msg = {
'role': 'assistant',
'content': text or None,
'tool_calls': [
{'id': tc['id'], 'type': 'function',
'function': {'name': tc['name'], 'arguments': tc['arguments']}}
for tc in tool_calls.values()
]
}
messages.append(assistant_msg)
# Execute tools and append results
tool_results = execute_tool_calls(tool_calls)
messages.extend(tool_results)
else:
# No more tool calls — final text response
print('Final answer:', text)
return text
return 'Max iterations reached'Streaming Text While Buffering Tool Calls
In practice, you want to stream text to the client immediately while simultaneously buffering any tool call arguments. This requires distinguishing between chunks that carry content (stream immediately) and chunks that carry tool_calls (buffer for later execution). Only after the stream ends and tool calls are complete can you execute tools and continue.
async def stream_with_tools(messages, tools):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini', messages=messages, tools=tools, stream=True
)
tool_buffers = defaultdict(lambda: {'name': '', 'id': '', 'arguments': ''})
text_parts = []
async for chunk in stream:
delta = chunk.choices[0].delta
finish = chunk.choices[0].finish_reason
if delta.content:
text_parts.append(delta.content)
yield ('text', delta.content) # stream to client immediately
if delta.tool_calls:
for tc in delta.tool_calls:
if tc.id: tool_buffers[tc.index]['id'] = tc.id
if tc.function.name: tool_buffers[tc.index]['name'] += tc.function.name
if tc.function.arguments: tool_buffers[tc.index]['arguments'] += tc.function.arguments
if finish == 'tool_calls':
yield ('tool_calls', dict(tool_buffers)) # signal tool execution neededParallel Tool Call Execution
When the model returns multiple tool calls simultaneously (a feature called parallel function calling), execute them in parallel with asyncio.gather rather than sequentially. Sequential execution adds unnecessary latency — if the model calls a weather API and a database lookup simultaneously, there is no reason to wait for one to finish before starting the other.
import asyncio
async def execute_tool_calls_parallel(tool_call_buffers: dict) -> list[dict]:
async def execute_one(idx, tc):
func_name = tc['name']
args = json.loads(tc['arguments'])
if func_name in async_tools_registry:
result = await async_tools_registry[func_name](**args)
else:
result = {'error': f'Unknown function: {func_name}'}
return {
'role': 'tool',
'tool_call_id': tc['id'],
'content': json.dumps(result),
}
tasks = [execute_one(idx, tc) for idx, tc in sorted(tool_call_buffers.items())]
return await asyncio.gather(*tasks)Streaming the Final Answer After Tool Use
After executing tool calls and appending results to the message history, make a second streaming request to get the model's final answer. Stream this response directly to the client. This two-request pattern (initial request with tool calls + follow-up request with tool results) is the standard agent turn cycle, and both requests can stream their text to the UI.
async def full_tool_calling_stream(question: str, tools: list):
messages = [{'role': 'user', 'content': question}]
# First request: may produce tool calls
tool_buffers = {}
text1 = ''
async for event_type, data in stream_with_tools(messages, tools):
if event_type == 'text':
text1 += data
yield data # stream partial text if any
elif event_type == 'tool_calls':
tool_buffers = data
if tool_buffers:
# Execute tools, then get final streaming answer
tool_results = await execute_tool_calls_parallel(tool_buffers)
messages += [{ # assistant tool call message
'role': 'assistant',
'tool_calls': [
{'id': tc['id'], 'type': 'function',
'function': {'name': tc['name'], 'arguments': tc['arguments']}}
for tc in tool_buffers.values()
]
}] + tool_results
# Second request: final answer streams directly
async for token in token_stream(messages): # from earlier lesson
yield tokenDisplaying Tool Call Progress in the UI
Users should see what the agent is doing while waiting for tool calls to complete. Before executing tools, stream a status event to the client indicating which function is being called and with what arguments. After execution, stream a completion status. This transparency greatly improves perceived responsiveness and helps users debug unexpected tool use.
import json
async def stream_with_progress(question, tools):
messages = [{'role': 'user', 'content': question}]
tool_buffers = {}
async for event_type, data in stream_with_tools(messages, tools):
if event_type == 'tool_calls':
tool_buffers = data
for tc in tool_buffers.values():
args = json.loads(tc['arguments'])
yield f'data: {json.dumps({"type": "tool_start", "function": tc["name"], "args": args})}\n\n'
result = tools_registry.get(tc['name'], lambda **kw: {})(** args)
yield f'data: {json.dumps({"type": "tool_done", "function": tc["name"]})}\n\n'
# Then stream final answer...Error Handling in Tool Call Streams
Tool execution can fail — APIs return errors, functions raise exceptions, JSON parsing fails. Always catch exceptions in tool execution and return a structured error response to the model. The model can then decide to retry with different arguments, call a fallback tool, or explain to the user that the requested action failed. Never let an uncaught tool exception crash the streaming loop.
def safe_execute_tool(func_name: str, args: dict) -> str:
try:
if func_name not in tools_registry:
return json.dumps({'error': f'Function {func_name!r} not found'})
result = tools_registry[func_name](**args)
return json.dumps(result)
except TypeError as e:
return json.dumps({'error': f'Invalid arguments: {str(e)}'})
except Exception as e:
return json.dumps({'error': f'Execution failed: {str(e)}'})
# Tool result message with error handled
tool_message = {
'role': 'tool',
'tool_call_id': tc['id'],
'content': safe_execute_tool(tc['name'], json.loads(tc['arguments'])),
}Comparing Streaming vs Non-Streaming for Agents
For agentic applications, streaming adds complexity but significant UX value. Without streaming, the user sees nothing during a multi-step tool-calling loop that might take 10-30 seconds. With streaming, they see intermediate text, tool call notifications, and the final answer appearing token by token. The added code complexity is usually worth it for interactive applications, but background agents that run autonomously can use non-streaming for simpler code.
Quick Check
Test your understanding of tool calls in streamed responses from this lesson.
Lesson Recap
In this lesson you learned: tool call arguments arrive as JSON fragments in streamed responses and must be buffered by tool call index before parsing, execute tools after stream completion then make a second streaming request for the final answer, and parallel execution with asyncio.gather minimizes latency when the model calls multiple functions simultaneously. Always catch tool execution errors to prevent crashing the agent loop. Next up we implement response caching to reduce API costs.
Questions Fréquemment Posées
La leçon « Gérer les appels d’outils dans les réponses diffusées » est-elle gratuite ?
Oui — le texte complet de « Gérer les appels d’outils dans les réponses diffusées » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Gérer les appels d’outils dans les réponses diffusées » ?
Analysez les réponses diffusées contenant des arguments d’appel de fonction reçus jeton par jeton, mettez les fragments JSON en mémoire tampon et ne déclenchez l’exécution de l’outil que lorsque l’ap… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?
Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Gérer les appels d’outils dans les réponses diffusées » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?
Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Comprendre la diffusion des jetons
- Consommer des flux avec le SDK Python
- Diffusion dans FastAPI avec les événements envoyés par le serveur
- Gérer les appels d’outils dans les réponses diffusées