AI Engineering Academy · Lektion

Tool-Aufrufe in gestreamten Antworten verarbeiten

Tool-Aufrufe in gestreamten Antworten verarbeiten ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Tool-Aufrufe treffen in Streams anders ein

Wenn ein LLM beschließt, eine Funktion aufzurufen, ändert sich die Struktur der Antwort. Anstelle eines content-Strings enthält das Delta ein tool_calls-Array. In einer gestreamten Antwort treffen die Argumente des Funktionsaufrufs jedoch Token für Token als partieller JSON-String ein – Sie erhalten kein vollständiges JSON-Objekt in einem einzigen Chunk. Sie müssen diese Fragmente puffern und das vollständige JSON wieder zusammensetzen, bevor Sie den Tool-Aufruf parsen und ausführen können.

# 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

Einen Tool-Aufruf im Stream erkennen

Prüfen Sie finish_reason jedes Chunks, um zu erkennen, wann Tool-Aufrufe zu erwarten sind. Wenn finish_reason 'tool_calls' lautet, hat das Modell beschlossen, eine Funktion aufzurufen, und der Stream endet. Wenn finish_reason 'stop' lautet, hat das Modell eine normale Textantwort erzeugt. Prüfen Sie während des Streamings, ob chunk.choices[0].delta.tool_calls nicht None ist, um Fragmente von Tool-Aufrufargumenten zu erkennen.

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

Argumente von Tool-Aufrufen puffern

Verwenden Sie ein nach dem Index des Tool-Aufrufs indiziertes Dictionary, um Argumentfragmente aus jedem Chunk zu sammeln. Der Index gibt an, welcher Tool-Aufruf gerade gestreamt wird – ein Modell kann mehrere Funktionen in einer Antwort aufrufen. Hängen Sie bei jedem Chunk mit einem nicht-null tool_calls-Delta das Argumentfragment an den passenden, durch den Tool-Aufruf-Index bestimmten Puffereintrag an.

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)

Tool-Aufrufe parsen und ausführen

Nachdem der Stream beendet ist und Sie vollständige Argument-Strings haben, parsen Sie jeden davon mit json.loads und leiten Sie ihn an die passende Python-Funktion weiter. Führen Sie die Tool-Aufrufe in der angeforderten Reihenfolge aus (oder parallel, wenn sie unabhängig voneinander sind) und formatieren Sie die Ergebnisse anschließend als Tool-Antwortnachrichten, die Sie im nachfolgenden API-Aufruf zurücksenden.

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

Der vollständige Ablauf für gestreamte Tool-Aufrufe

Der vollständige Ablauf für gestreamte Tool-Aufrufe erfordert eine mehrteilige Konversationsschleife. Die erste Anfrage kann Tool-Aufrufe zurückgeben; Sie führen diese aus und hängen die Ergebnisse an den Nachrichtenverlauf an. Eine zweite Anfrage liefert dann die endgültige Textantwort. Diese Schleife kann mehrmals durchlaufen werden, wenn das Modell aufgrund der vorherigen Ergebnisse weitere Tool-Aufrufe ausführen möchte.

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'

Text streamen und gleichzeitig Tool-Aufrufe puffern

In der Praxis möchten Sie Text sofort an den Client streamen und gleichzeitig etwaige Argumente von Tool-Aufrufen puffern. Dazu müssen Sie zwischen Chunks unterscheiden, die content enthalten (sofort streamen), und Chunks, die tool_calls enthalten (für die spätere Ausführung puffern). Erst nachdem der Stream beendet und die Tool-Aufrufe vollständig eingegangen sind, können Sie die Tools ausführen und fortfahren.

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

Parallele Ausführung von Tool-Aufrufen

Wenn das Modell mehrere Tool-Aufrufe gleichzeitig zurückgibt (eine Funktion namens parallele Funktionsaufrufe), führen Sie sie mit asyncio.gather parallel statt nacheinander aus. Eine sequenzielle Ausführung verursacht unnötige Latenz – wenn das Modell gleichzeitig eine Wetter-API und eine Datenbankabfrage aufruft, gibt es keinen Grund, auf den Abschluss des einen Aufrufs zu warten, bevor Sie den anderen starten.

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)

Die endgültige Antwort nach der Tool-Nutzung streamen

Nachdem Sie die Tool-Aufrufe ausgeführt und die Ergebnisse an den Nachrichtenverlauf angehängt haben, stellen Sie eine zweite Streaming-Anfrage, um die endgültige Antwort des Modells abzurufen. Streamen Sie diese Antwort direkt an den Client. Dieses Muster aus zwei Anfragen (erste Anfrage mit Tool-Aufrufen + Folgeanfrage mit Tool-Ergebnissen) ist der standardmäßige Ablauf eines Agenten-Zyklus, und beide Anfragen können ihren Text an die Benutzeroberfläche streamen.

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

Fortschritt von Tool-Aufrufen in der Benutzeroberfläche anzeigen

Benutzer sollten sehen, woran der Agent gerade arbeitet, während sie auf den Abschluss der Tool-Aufrufe warten. Streamen Sie vor der Ausführung der Tools ein Statusereignis an den Client, das angibt, welche Funktion mit welchen Argumenten aufgerufen wird. Streamen Sie nach der Ausführung einen Abschlussstatus. Diese Transparenz verbessert die wahrgenommene Reaktionsfähigkeit erheblich und hilft Benutzern, unerwartete Tool-Nutzung zu untersuchen.

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

Fehlerbehandlung in Streams mit Tool-Aufrufen

Die Ausführung von Tools kann fehlschlagen – APIs geben Fehler zurück, Funktionen lösen Ausnahmen aus und das Parsen von JSON kann scheitern. Fangen Sie bei der Tool-Ausführung immer Ausnahmen ab und geben Sie eine strukturierte Fehlerantwort an das Modell zurück. Das Modell kann dann entscheiden, ob es es mit anderen Argumenten erneut versucht, ein Ersatz-Tool aufruft oder dem Benutzer erklärt, dass die angeforderte Aktion fehlgeschlagen ist. Lassen Sie niemals zu, dass eine nicht abgefangene Tool-Ausnahme die Streaming-Schleife zum Absturz bringt.

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'])),
}

Streaming und Nicht-Streaming für Agenten vergleichen

Bei agentenbasierten Anwendungen erhöht Streaming die Komplexität, bietet aber erhebliche Vorteile für die Benutzererfahrung. Ohne Streaming sieht der Benutzer während einer mehrstufigen Tool-Aufruf-Schleife, die 10–30 Sekunden dauern kann, nichts. Mit Streaming sieht er Zwischentext, Benachrichtigungen zu Tool-Aufrufen und die endgültige Antwort, die Token für Token erscheint. Die zusätzliche Codekomplexität lohnt sich für interaktive Anwendungen in der Regel, während Hintergrundagenten, die autonom ausgeführt werden, für einfacheren Code Nicht-Streaming verwenden können.

Schnelltest

Testen Sie Ihr Verständnis von Tool-Aufrufen in gestreamten Antworten aus dieser Lektion.

Zusammenfassung der Lektion

In dieser Lektion haben Sie gelernt: Die Argumente von Tool-Aufrufen treffen in gestreamten Antworten als JSON-Fragmente ein und müssen vor dem Parsen nach dem Tool-Aufruf-Index gepuffert werden; führen Sie Tools nach Abschluss des Streams aus und stellen Sie anschließend eine zweite Streaming-Anfrage für die endgültige Antwort; und die parallele Ausführung mit asyncio.gather minimiert die Latenz, wenn das Modell mehrere Funktionen gleichzeitig aufruft. Fangen Sie Fehler bei der Tool-Ausführung immer ab, um einen Absturz der Agentenschleife zu verhindern. Als Nächstes implementieren wir das Caching von Antworten, um API-Kosten zu senken.

Häufig gestellte Fragen

Ist die Lektion „Tool-Aufrufe in gestreamten Antworten verarbeiten“ kostenlos?

Ja — der vollständige Text von „Tool-Aufrufe in gestreamten Antworten verarbeiten“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Tool-Aufrufe in gestreamten Antworten verarbeiten“?

Parsen Sie gestreamte Antworten, in denen Argumente für Funktionsaufrufe Token für Token eintreffen, puffern Sie die JSON-Fragmente und starten Sie die Tool-Ausführung erst, wenn der Aufruf vollständ… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Engineering Academy zu starten?

Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Tool-Aufrufe in gestreamten Antworten verarbeiten“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?

Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

Token-Streaming verstehen
Streams mit dem Python-SDK verarbeiten
Streaming in FastAPI mit Server-Sent Events
Tool-Aufrufe in gestreamten Antworten verarbeiten

← Zurück zu AI Engineering Academy