使用Python+rich库集成OpenAI API时终端ChatGPT流式响应换行问题

问题分析 #

你的ask_stream函数存在两个核心问题：

1. 逻辑错误：每次循环打印的是上一个响应片段，且每个片段单独渲染Markdown，结合console.print的默认换行行为，导致每个片段单独占一行。
2. 阻塞异步事件循环：在异步函数中使用time.sleep()会阻塞整个事件循环，影响程序响应性。

解决方案 #

使用Rich库的Live组件实现实时流式更新（类似ChatGPT的逐字/逐段显示效果），同时修正异步睡眠逻辑，累积响应内容后统一渲染Markdown，避免分段换行。

修改后的完整代码 #

import openai
from rich.markdown import Markdown
from rich.console import Console
from rich.live import Live  # 新增导入Live组件
from prompt_toolkit import PromptSession
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.history import InMemoryHistory
import argparse
import asyncio

openai.api_key = "MY API KEY"
model = "gpt-4"
delay_time = 0.01
max_response_length = 200
console = Console()


async def ask_stream(prompt):
    # 使用OpenAI异步客户端优化异步性能
    client = openai.AsyncOpenAI(api_key=openai.api_key)
    response = await client.chat.completions.create(
        model='gpt-4',
        messages=[{"role": "user", "content": f"{prompt}"}],
        max_tokens=8000,
        temperature=0.4,
        stream=True
    )
    
    full_response = ""
    # 初始化Live组件，实时更新渲染后的Markdown内容
    with Live(Markdown(full_response), auto_refresh=True) as live:
        async for event in response:
            # 提取当前响应片段的内容
            content = event.choices[0].delta.content
            if content:
                full_response += content
                # 更新Live组件的内容为最新渲染的Markdown
                live.update(Markdown(full_response))
            # 使用异步睡眠替代同步睡眠，避免阻塞事件循环
            await asyncio.sleep(delay_time)


async def ask(prompt) -> Markdown | str:
    if prompt:
        completion = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": f"{prompt}"}]
        )
        if completion:
            if 'error' in completion:
                return completion['error']['message']
            return Markdown(completion.choices[0].message.content)
        else:
            raise Exception("请求无响应")


def create_session() -> PromptSession:
    return PromptSession(history=InMemoryHistory())


async def get_input_async(
        session: PromptSession = None,
        completer: WordCompleter = None,
) -> str:
    """多行输入函数"""
    return await session.prompt_async(
        completer=completer,
        multiline=True,
        auto_suggest=AutoSuggestFromHistory(),
    )


async def main():
    console.print(f"Starting Chatgpt with model - {model}")
    session = create_session()
    while True:
        console.print("\nYou:")
        question = await get_input_async(session=session)
        console.print()
        if question == "!exit":
            break
        elif question == "!help":
            console.print(
                """
            !help - 显示帮助信息
            !exit - 退出程序
            """,
            )
            continue
        console.print("ChatGPT:")
        if args.no_stream:
            result = await ask(prompt=question)
            console.print(result)
        else:
            await ask_stream(prompt=question)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--no-stream", action="store_true")
    args = parser.parse_args()
    asyncio.run(main())

关键修改说明 #

1. 引入Live组件：通过Live上下文管理器，终端会实时刷新显示最新的完整响应内容，避免分段换行。
2. 累积响应内容：用full_response变量逐步拼接所有流式返回的内容片段，确保Markdown渲染的完整性。
3. 异步客户端优化：改用OpenAI的异步客户端AsyncOpenAI，配合async for遍历流式响应，避免阻塞异步事件循环。
4. 替换睡眠方式：用await asyncio.sleep()替代time.sleep()，保证异步程序的响应性。
5. 统一输出方式：将原代码中的print()替换为console.print()，保持Rich库的输出一致性。

内容的提问来源于stack exchange，提问作者sparker