0Pricing
AI Engineering Academy · Ders

Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme

İşlev çağrısı bağımsız değişkenlerinin belirteç belirteç geldiği akış yanıtlarını ayrıştırın, JSON parçalarını arabelleğe alın ve araç çalıştırmayı yalnızca çağrı tamamlandığında başlatın.

Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme, CoddyKit'te ücretsiz bir AI Engineering Academy dersidir. Bu, 4 dersinin 4. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, AI Engineering Academy öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. AI Engineering Academy kursu toplamda 4 dersten oluşur.

Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.

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 work

Detecting 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_type

Buffering 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_messages

The 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 needed

Parallel 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 token

Displaying 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.

Sıkça Sorulan Sorular

“Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme” dersi ücretsiz mi?

Evet — “Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve AI Engineering Academy kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. AI Engineering Academy kursu toplamda 4 dersten oluşur.

“Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme” dersinde ne öğreneceğim?

İşlev çağrısı bağımsız değişkenlerinin belirteç belirteç geldiği akış yanıtlarını ayrıştırın, JSON parçalarını arabelleğe alın ve araç çalıştırmayı yalnızca çağrı tamamlandığında başlatın. AI Engineering Academy ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

AI Engineering Academy öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te AI Engineering Academy, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 4. dersidir.

“Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu AI Engineering Academy dersinde kod yazıp çalıştırabilir miyim?

Evet. Her AI Engineering Academy dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. Belirteç Akışını Anlama
  2. Python SDK ile Akışları Tüketme
  3. Server-Sent Events ile FastAPI'da Akış
  4. Akış Halindeki Yanıtlarda Araç Çağrılarını İşleme
← AI Engineering Academy Sayfasına Dön