AI Agents · 课时

CLI 代理中的流式输出

Q: 「CLI 代理中的流式输出」这节课中我会学到什么？

在终端界面中逐字符打印流式传输的词元。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

在终端界面中逐字符打印流式传输的词元。

CLI 代理中的流式输出 是 CoddyKit 上的免费 AI Agents 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

流式处理为何对 CLI 代理很重要

不使用流式处理时，CLI 代理要等到完整的 LLM 响应准备好后才会打印内容，这可能需要 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 检测流结束

流中的最后一个数据块具有非空的 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 库可以在终端中提供 Markdown 渲染、语法高亮的代码块、表格和旋转指示器。它非常适合与代理的流式输出搭配使用。

# 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，以检测输出是否被截断
对于异步代理，将 async for 与 AsyncOpenAI 结合使用
添加 ANSI 颜色或使用 rich，打造精致的终端体验

常见问题解答

「CLI 代理中的流式输出」课时是免费的吗？

是的 — 「CLI 代理中的流式输出」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「CLI 代理中的流式输出」这节课中我会学到什么？

在终端界面中逐字符打印流式传输的词元。你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「CLI 代理中的流式输出」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents