본문 바로가기
C.W.K.
Stream
Lesson 04 of 06 · published

Function Call Streaming — fragment 조립

~22 min · streaming, function-calls

Level 0Tokenizer
0 XP0/54 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

function call 의 arguments 는 streaming 중에 JSON fragment 로 나뉘어 와. {"loc, ation":", Seoul"} 처럼 각 조각만으로는 올바른 JSON 이 아니야. 모두 이어 붙인 뒤 한 번만 json.loads 를 호출해.

모델이 token 단위로 만들기 때문이야

모델은 arguments JSON 도 token 단위로 생성하므로 전송 chunk 가 JSON 경계와 맞지 않아. 중간 fragment 를 곧바로 parsing 하면 실패하는 게 정상이고, 끝날 때까지 기다려야 해.

tool_call_id 별 accumulator 를 둬

각 tool_call_id 에 arguments_str 를 하나씩 두고 fragment 가 올 때마다 뒤에 붙여. tool call 완료 event 나 stream 종료를 확인한 뒤에만 parsing 해. SDK 가 대신 처리하는 경우도 있지만 raw httpx 에서는 직접 구현해야 해.

여러 tool call 을 섞지 마

한 turn 에 여러 도구가 동시에 호출될 수 있어. tool_calls[i].function.arguments 의 인덱스마다 별도 accumulator 를 사용해 각 JSON 문자열이 뒤섞이지 않게 해.

Code

Streamed tool arguments 재조립·python
import json

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "What's the weather in Paris?"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string"}},
                "required": ["location"],
            }
        }
    }],
    stream=True,
)

tool_call_accumulator = {}
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.tool_calls:
        for tc in delta.tool_calls:
            idx = tc.index
            if idx not in tool_call_accumulator:
                tool_call_accumulator[idx] = {"id": "", "name": "", "arguments": ""}
            if tc.id:
                tool_call_accumulator[idx]["id"] = tc.id
            if tc.function.name:
                tool_call_accumulator[idx]["name"] = tc.function.name
            if tc.function.arguments:
                tool_call_accumulator[idx]["arguments"] += tc.function.arguments

# Process accumulated tool calls
for tc in tool_call_accumulator.values():
    args = json.loads(tc["arguments"])
    print(f"Tool: {tc['name']}({args})")

External links

Exercise

도구를 쓰는 stream 을 실행하고 chunk 마다 tool_calls fragment 를 기록해. 모두 이어 붙여 완성된 arguments JSON 을 재현하고 json.loads 가 마지막에만 성공하는지 확인해.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.