Streaming und Tools

Streaming mit Funktionsaufrufen

Mit Streaming erhalten Sie die Antwort des Modells inkrementell, Token für Token. Dies ist für responsive Benutzeroberflächen unerlässlich. Doch wie interagiert Streaming mit Funktionsaufrufen?

Die Grundregel

Funktionsaufrufe werden in einem einzelnen Chunk zurückgegeben, nicht schrittweise gestreamt. Dies unterscheidet sich grundlegend von normalem Text, der Token für Token ankommt.

In der Praxis:

• Antwort-Text wird schrittweise gestreamt
• tool_calls kommen in einem einzigen Ereignis alle auf einmal an

Diese Einschränkung besteht aus gutem Grund: Ein Funktionsaufruf muss vollständig sein (Name + gültige JSON-Argumente), damit Sie ihn ausführen können. Teilweise JSON wäre unbrauchbar.

Implementierung von Streaming mit Tools

stream = client.chat.completions.create(
    model="grok-4.5",
    messages=messages,
    tools=tools,
    stream=True
)

collected_tool_calls = []
collected_content = ""

for chunk in stream:
    delta = chunk.choices[0].delta

    # Gestreamter Text
    if delta.content:
        collected_content += delta.content
        print(delta.content, end="", flush=True)

    # Funktionsaufrufe (kommen in einem einzigen Block an)
    if delta.tool_calls:
        for tc in delta.tool_calls:
            collected_tool_calls.append(tc)

    # Prüfung auf Ende des Streams
    if chunk.choices[0].finish_reason == "tool_calls":
        # Modell erwartet Tool-Ergebnisse
        break
    elif chunk.choices[0].finish_reason == "stop":
        # Antwort beendet, kein Tool aufgerufen
        break

Rekonstruktion von Funktionsaufrufen beim Streaming

In einigen Fällen können Funktionsaufruf-Argumente in mehreren Teilen innerhalb des Streams ankommen. Sie müssen diese rekonstruieren:

tool_calls_buffer = {}

for chunk in stream:
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc_chunk in delta.tool_calls:
            idx = tc_chunk.index

            if idx not in tool_calls_buffer:
                tool_calls_buffer[idx] = {
                    "id": tc_chunk.id,
                    "function": {
                        "name": tc_chunk.function.name or "",
                        "arguments": ""
                    }
                }

            if tc_chunk.function.name:
                tool_calls_buffer[idx]["function"]["name"] = tc_chunk.function.name
            if tc_chunk.function.arguments:
                tool_calls_buffer[idx]["function"]["arguments"] += tc_chunk.function.arguments

Dieser Pufferungsansatz stellt sicher, dass Sie jeden Funktionsaufruf korrekt rekonstruieren, auch wenn die Daten fragmentiert ankommen.

Architektur für Streaming mit Tools

In einer echten Anwendung müssen Sie zwei gleichzeitige Streams verarbeiten:

1. Echtzeit-Anzeige: Text anzeigen, wie er ankommt
2. Sammlung von Funktionsaufrufen: Aufrufe sammeln zur Ausführung

async def handle_streaming_response(stream):
    content_parts = []
    tool_calls = []
    current_tool = None

    async for chunk in stream:
        delta = chunk.choices[0].delta

        if delta.content:
            content_parts.append(delta.content)
            yield {"type": "text", "data": delta.content}

        if delta.tool_calls:
            for tc in delta.tool_calls:
                tool_calls.append(tc)

        if chunk.choices[0].finish_reason == "tool_calls":
            # Tools ausführen und neu starten
            results = await execute_tools(tool_calls)
            yield {"type": "tool_results", "data": results}

Latenz und Benutzererfahrung

Streaming mit Tools beinhaltet natürliche Pausen im Ablauf:

1. Der Nutzer stellt seine Frage
2. Das Modell beginnt zu überlegen (keine sichtbaren Tokens)
3. Das Modell entscheidet sich für einen Funktionsaufruf (Sie erhalten den tool_call)
4. Pause – Sie führen die Funktion aus
5. Sie senden das Ergebnis zurück
6. Das Modell generiert seine Antwort (schrittweise gestreamt)

Zeigen Sie für die Benutzererfahrung während Schritt 4 einen Indikator an:

🔍 Suche nach Wetter in Paris...

Dies verhindert, dass der Nutzer denkt, die Anwendung sei eingefroren.

Server-Sent Events (SSE) für Webanwendungen

Wenn Sie eine Web-API erstellen, verwenden Sie SSE, um den Stream zu übermitteln:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(message: str):
    async def event_stream():
        stream = client.chat.completions.create(
            model="grok-4.5",
            messages=[{"role": "user", "content": message}],
            tools=tools,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Kernaussagen

• Funktionsaufrufe kommen in einem einzigen Chunk an, nicht schrittweise gestreamt
• Normaler Text wird weiterhin Token für Token gestreamt
• Fragmentierte Funktionsaufrufe mit einem Puffer rekonstruieren, der nach index indiziert ist
• Verarbeiten Sie zwei Streams parallel: Textanzeige + Sammlung von Funktionsaufrufen
• Zeigen Sie während der Tool-Ausführung einen Indikator an für die UX
• Verwenden Sie SSE für Webanwendungen