处理流式响应中的工具调用
解析逐个令牌传入函数调用参数的流式响应，缓冲 JSON 片段，并仅在调用完成后触发工具执行。
处理流式响应中的工具调用 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
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.
常见问题解答
「处理流式响应中的工具调用」课时是免费的吗？
是的 — 「处理流式响应中的工具调用」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「处理流式响应中的工具调用」这节课中我会学到什么？
解析逐个令牌传入函数调用参数的流式响应，缓冲 JSON 片段，并仅在调用完成后触发工具执行。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「处理流式响应中的工具调用」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。