AI Agents · 강의

CLI 에이전트의 스트리밍 출력

터미널 인터페이스에서 스트리밍 토큰을 문자 단위로 출력합니다.

레슨 4/413개 단계

CLI 에이전트의 스트리밍 출력은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

CLI 에이전트에서 스트리밍이 중요한 이유

스트리밍을 사용하지 않으면 전체 LLM 응답이 준비될 때까지 CLI 에이전트가 아무것도 출력하지 않습니다. 이 과정에는 5~30초가 걸릴 수 있어 사용자는 프로그램이 멈췄는지 걱정하며 빈 터미널을 바라보게 됩니다.

스트리밍을 사용하면 토큰이 생성되는 즉시 표시되므로 즉각적인 피드백을 제공하고 사용자 경험을 크게 향상할 수 있습니다.

OpenAI SDK에서 스트리밍 활성화하기

stream=True를 chat.completions.create()에 전달하십시오. 호출 결과로 완성된 응답 객체가 아니라 생성기가 반환됩니다. 생성기를 반복하면서 도착하는 청크를 처리하십시오.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain Python generators in 3 sentences.'}],
    stream=True  # <-- enable streaming
)

# Each chunk arrives as it is generated
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end='', flush=True)

print()  # newline after the response is complete

print()와 sys.stdout.write() 비교

스트리밍할 때는 print(text, end='', flush=True)를 사용하거나, sys.stdout.write(text) 다음에 sys.stdout.flush()를 사용하십시오. flush=True가 없으면 Python이 출력을 버퍼에 저장했다가 한 번에 모두 출력할 수 있어 스트리밍의 목적이 사라집니다.

import sys
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_to_terminal(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    full_response = ''
    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_response += token

        # Option 1: print with flush
        print(token, end='', flush=True)

        # Option 2: sys.stdout.write + flush
        # sys.stdout.write(token)
        # sys.stdout.flush()

    print()  # final newline
    return full_response

스트리밍 중 전체 응답 수집하기

스트리밍이 끝난 후 저장, 추가 처리 또는 표시를 위해 전체 응답 텍스트가 필요한 경우가 많습니다. 토큰을 출력하면서 문자열에 누적하십시오.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_and_collect(messages: list) -> str:
    full_text = ''

    print('Agent: ', end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()  # newline
    return full_text

# The return value contains the complete response for storage
# response_text = stream_and_collect(history)
# history.append({'role': 'assistant', 'content': response_text})

AsyncOpenAI를 사용한 비동기 스트리밍

비동기 에이전트 구조에서는 AsyncOpenAI와 async for를 사용하여 이벤트 반복을 차단하지 않고 스트리밍 청크를 순회하십시오.

import asyncio
import openai

async def async_stream_agent(query: str) -> str:
    client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')

    stream = await client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': query}],
        stream=True
    )

    full_text = ''
    print('Agent: ', end='', flush=True)

    async for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()
    return full_text

# asyncio.run(async_stream_agent('What is asyncio?'))

finish_reason으로 스트림 종료 감지하기

스트림의 마지막 청크에는 null이 아닌 finish_reason이 있습니다. 이를 확인하면 스트림이 종료된 이유를 알 수 있습니다: 'stop' = 정상 완료, 'length' = 잘림, 'tool_calls' = 함수 호출 필요.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_finish_detection(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    finish_reason = None
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)
        if chunk.choices[0].finish_reason:
            finish_reason = chunk.choices[0].finish_reason

    print()

    if finish_reason == 'length':
        print('[WARNING: Response was truncated. Try increasing max_tokens.]')
    elif finish_reason == 'stop':
        pass  # normal completion

    return finish_reason

터미널 출력의 ANSI 색상

ANSI 이스케이프 코드는 터미널 출력에 색상을 추가합니다. 이를 사용해 에이전트 접두사, 사용자 입력 프롬프트, 경고를 시각적으로 구분하십시오. colorama 라이브러리는 Windows를 포함한 여러 플랫폼을 지원합니다.

# pip install colorama
from colorama import Fore, Style, init
init(autoreset=True)  # reset color after each print

def print_colored_stream(messages: list, client):
    # Print agent prefix in cyan
    print(Fore.CYAN + 'Agent: ' + Style.RESET_ALL, end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        print(token, end='', flush=True)

    print()

# Also useful:
# print(Fore.GREEN + 'Success!') — green
# print(Fore.RED + 'Error!') — red
# print(Fore.YELLOW + 'Warning') — yellow

더 풍부한 터미널 출력을 위한 Rich 라이브러리

rich 라이브러리는 터미널에서 마크다운 렌더링, 구문 강조 코드 블록, 표, 스피너를 제공합니다. 스트리밍되는 에이전트 출력과 함께 사용하기 좋습니다.

# pip install rich
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown

console = Console()

def stream_with_rich(messages: list, client):
    full_text = ''

    with Live(console=console, refresh_per_second=10) as live:
        stream = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            stream=True
        )

        for chunk in stream:
            token = chunk.choices[0].delta.content or ''
            full_text += token
            # Render accumulated text as Markdown in real time
            live.update(Markdown(full_text))

    return full_text

도구 호출과 함께 스트리밍하기

스트리밍을 함수 호출과 함께 사용하면 tool_calls 필드도 여러 조각으로 스트리밍됩니다. 구문 분석하기 전에 청크 전체에서 JSON 문자열을 누적하십시오.

import json
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_tools(messages: list, tools: list) -> dict:
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        tools=tools,
        stream=True
    )

    tool_call_chunks = {}
    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_chunks:
                    tool_call_chunks[idx] = {'name': '', 'args': ''}
                if tc.function.name:
                    tool_call_chunks[idx]['name'] += tc.function.name
                if tc.function.arguments:
                    tool_call_chunks[idx]['args'] += tc.function.arguments

    # Parse accumulated tool calls
    return {v['name']: json.loads(v['args']) for v in tool_call_chunks.values()}

토큰 수 표시

스트리밍 중에 실시간 토큰 수를 표시하면 사용자가 사용량을 모니터링하고 비용을 이해하는 데 도움이 됩니다. stream_options={'include_usage': True}를 설정하면 OpenAI 스트림의 마지막 청크에 사용량 데이터가 포함됩니다.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_token_count(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True,
        stream_options={'include_usage': True}
    )

    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)

        # Last chunk includes usage
        if chunk.usage:
            print(f'\n[Tokens: prompt={chunk.usage.prompt_tokens}, '
                  f'completion={chunk.usage.completion_tokens}, '
                  f'total={chunk.usage.total_tokens}]')

스트리밍 모범 사례

CLI 에이전트의 스트리밍 출력 모범 사례 요약:

버퍼링을 방지하려면 항상 flush=True 또는 sys.stdout.flush()를 사용하십시오
스트리밍 후 저장할 수 있도록 토큰을 문자열에 누적하십시오
잘림을 감지하려면 finish_reason을 확인하십시오
시각적 명확성을 위해 ANSI 색상이나 rich를 사용하십시오
JSON 인수 청크를 누적하여 도구 호출 스트리밍을 처리하십시오

학습 확인: 스트리밍 출력

CLI 에이전트의 스트리밍 출력에 대한 이해도를 확인해 보십시오.

인터랙티브 레슨에서 답해보세요

복습: CLI 에이전트의 스트리밍 출력

이제 반응성이 뛰어나고 현대적인 느낌을 주는 스트리밍 CLI 에이전트를 구축할 수 있습니다.

stream=True를 전달하여 OpenAI SDK의 스트리밍을 활성화합니다
print(token, end='', flush=True)를 사용하여 토큰을 즉시 표시합니다
스트리밍 후 처리에 사용할 수 있도록 토큰을 문자열에 누적합니다
마지막 청크의 finish_reason을 확인하여 잘림이 발생했는지 감지합니다
비동기 에이전트에는 AsyncOpenAI와 함께 async for를 사용합니다
세련된 터미널 사용 경험을 위해 ANSI 색상이나 rich를 추가합니다

무료로 시작

AI 튜터와 함께 AI Agents을(를) 배우세요 — 무료

브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.

코스: 60
레슨: 239

무료로 학습 시작 전체 코스 보기

자주 묻는 질문

“CLI 에이전트의 스트리밍 출력” 강의는 무료인가요?

네 — “CLI 에이전트의 스트리밍 출력” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“CLI 에이전트의 스트리밍 출력”에서 뭘 배우나요?

터미널 인터페이스에서 스트리밍 토큰을 문자 단위로 출력합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.

“CLI 에이전트의 스트리밍 출력” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기