0Pricing
AI Agents · Pelajaran

Mengalirkan Output dalam Agen CLI

Cetak token yang dialirkan satu karakter demi satu karakter pada antarmuka terminal.

Mengalirkan Output dalam Agen CLI adalah pelajaran AI Agents gratis di CoddyKit. Ini adalah pelajaran 4 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar AI Agents, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus AI Agents mencakup 4 pelajaran total.

Mengapa Pengaliran Penting bagi Agen CLI

Tanpa pengaliran, agen CLI Anda tidak mencetak apa pun sampai respons LLM lengkap siap — proses ini dapat memerlukan waktu 5-30 detik. Pengguna hanya melihat terminal kosong dan bertanya-tanya apakah program mengalami kerusakan.

Dengan pengaliran, token muncul saat dihasilkan sehingga pengguna segera mendapatkan umpan balik dan pengalaman yang jauh lebih baik.

Mengaktifkan Pengaliran di SDK OpenAI

Teruskan stream=True ke chat.completions.create(). Pemanggilan ini mengembalikan pembangkit, bukan objek respons lengkap. Lakukan iterasi terhadapnya untuk memproses potongan saat tiba.

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() vs sys.stdout.write()

Saat melakukan pengaliran, gunakan print(text, end='', flush=True) atau sys.stdout.write(text), lalu sys.stdout.flush(). Tanpa flush=True, Python mungkin menyangga keluaran dan mencetak semuanya sekaligus — sehingga tujuan pengaliran tidak tercapai.

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

Mengumpulkan Respons Lengkap Selama Pengaliran

Sering kali Anda memerlukan teks respons lengkap setelah pengaliran selesai — untuk penyimpanan, pemrosesan lebih lanjut, atau tampilan. Kumpulkan token ke dalam string sambil mencetaknya.

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})

Pengaliran Asinkron dengan AsyncOpenAI

Untuk arsitektur agen asinkron, gunakan AsyncOpenAI dan async for untuk melakukan iterasi terhadap potongan pengaliran tanpa memblokir perulangan peristiwa.

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?'))

Mendeteksi Akhir Aliran dengan finish_reason

Potongan terakhir dalam aliran memiliki finish_reason yang bukan null. Periksa nilai tersebut untuk mengetahui alasan aliran berakhir: 'stop' = selesai secara normal, 'length' = terpotong, 'tool_calls' = diperlukan pemanggilan fungsi.

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

Warna ANSI pada Keluaran Terminal

Kode pelolosan ANSI menambahkan warna pada keluaran terminal. Gunakan kode tersebut untuk membedakan awalan agen, permintaan masukan pengguna, dan peringatan secara visual. Pustaka colorama menyediakan dukungan lintas platform, termasuk 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

Pustaka Rich untuk Keluaran Terminal yang Lebih Kaya

Pustaka rich menyediakan penyajian markdown, blok kode dengan penyorotan sintaks, tabel, dan indikator putar di terminal. Pustaka ini cocok dipadukan dengan keluaran agen yang dialirkan.

# 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

Pengaliran dengan Pemanggilan Alat

Saat pengaliran digabungkan dengan pemanggilan fungsi, bidang tool_calls juga dialirkan dalam beberapa bagian. Kumpulkan string JSON di antara berbagai potongan sebelum menguraikannya.

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()}

Tampilan Penghitung Token

Tampilkan penghitung token secara langsung selama pengaliran untuk membantu pengguna memantau penggunaan dan memahami biaya. Aliran OpenAI menyertakan data penggunaan dalam potongan terakhir saat stream_options={'include_usage': True} ditetapkan.

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

Praktik Terbaik Pengaliran

Ringkasan praktik terbaik untuk keluaran yang dialirkan pada agen CLI:

  • Selalu gunakan flush=True atau sys.stdout.flush() untuk mencegah penyanggaan
  • Kumpulkan token ke dalam string untuk disimpan setelah pengaliran selesai
  • Periksa finish_reason untuk mendeteksi pemotongan
  • Gunakan warna ANSI atau rich untuk kejelasan visual
  • Tangani pengaliran pemanggilan alat dengan mengumpulkan potongan argumen JSON

Pemeriksaan Pemahaman: Keluaran yang Dialirkan

Uji pemahaman Anda tentang keluaran yang dialirkan pada agen CLI.

Ringkasan: Keluaran Bertahap pada Agen CLI

Sekarang Anda dapat membangun agen CLI dengan keluaran yang ditampilkan secara bertahap sehingga terasa responsif dan modern:

  • Teruskan stream=True untuk mengaktifkan keluaran bertahap dari SDK OpenAI
  • Gunakan print(token, end='', flush=True) untuk segera menampilkan token
  • Kumpulkan token menjadi teks untuk diproses setelah keluaran selesai
  • Periksa finish_reason pada potongan terakhir untuk mendeteksi keluaran yang terpotong
  • Gunakan async for dengan AsyncOpenAI untuk agen asinkron
  • Tambahkan warna ANSI atau rich untuk pengalaman terminal yang lebih rapi

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Mengalirkan Output dalam Agen CLI” gratis?

Ya — teks lengkap “Mengalirkan Output dalam Agen CLI” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus AI Agents, upgrade ke CoddyKit PRO. Kursus AI Agents mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Mengalirkan Output dalam Agen CLI”?

Cetak token yang dialirkan satu karakter demi satu karakter pada antarmuka terminal. Kamu berlatih AI Agents dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai AI Agents?

Tidak diperlukan pengalaman sebelumnya. AI Agents di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 4 dari 4.

Berapa lama pelajaran “Mengalirkan Output dalam Agen CLI” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran AI Agents ini?

Ya. Setiap pelajaran AI Agents menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

  1. Membangun Antarmuka Agen Baris Perintah
  2. Agen Interaktif Bergaya REPL
  3. Penguraian Argumen dan Teks Bantuan
  4. Mengalirkan Output dalam Agen CLI
← Kembali ke AI Agents