4. 实现流式输出

任务目标

本案例要求配置TGI服务以流式方式返回生成结果，并编写客户端代码逐步接收、实时显示生成的文本。最终验证方式是：客户端在完整生成结束前，就能看到已生成的部分文本持续追加输出。

环境与依赖

• 已安装并运行TGI服务（默认端口8080），模型已加载（如Qwen/Qwen2.5-7B-Instruct）。
• Python 3.9+，安装requests库（pip install requests）。

实现原理

TGI原生支持流式输出，只需在HTTP请求中设置stream=true，响应即为text/event-stream格式。服务端会按生成进度，以SSE（Server-Sent Events）事件逐段推送数据。客户端通过逐行读取响应流，解析data:前缀的JSON负载，即可获得增量生成的文本。

关键完整代码

import requests
import json

# TGI服务地址与请求参数
url = "http://localhost:8080/generate_stream"
payload = {
    "inputs": "请用三句话介绍深度学习。",
    "parameters": {
        "max_new_tokens": 200,
        "temperature": 0.7,
        "stream": True
    }
}

headers = {"Content-Type": "application/json"}

def stream_generate():
    """发送流式请求，逐块打印生成的文本"""
    with requests.post(url, json=payload, headers=headers, stream=True) as resp:
        resp.raise_for_status()
        # 检查响应类型是否为SSE
        if "text/event-stream" not in resp.headers.get("Content-Type", ""):
            print("警告：服务未返回流式格式，请确认TGI已启用stream")
        # 逐行读取SSE流
        for line in resp.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data:"):
                continue
            data_str = line[5:].strip()  # 去掉"data:"前缀
            if data_str == "[DONE]":
                break
            try:
                data = json.loads(data_str)
                _text = data.get("", {}).get("text", "")
                if _text:
                    print(_text, end="", flush=True)
            except json.JSONDecodeError:
                continue

if __name__ == "__main__":
    print("流式输出结果：\n")
    stream_generate()
    print("\n\n生成完成")

运行与验证

1. 确保TGI服务已启动且模型加载完成，可通过curl http://localhost:8080/health返回200确认。
2. 将上述代码保存为stream_client.py，在终端执行python stream_client.py。
3. 通过判定标准：观察终端输出，文本会以逐词或逐字的方式持续追加打印（而非一次性全部出现），且最终能完整输出三句关于深度学习的介绍。若在生成过程中看到部分文本先出现、后续继续增长，即表示流式功能正常。

常见问题

• 响应非SSE格式：确认请求体中的parameters.stream为true，且访问的是/generate_stream端点而非/generate。
• 中文乱码：确保终端编码为UTF-8，或在代码中设置sys.stdout.reconfigure(encoding='utf-8')。
• 连接中断：若生成时间过长，可适当调低max_new_tokens或检查网络代理设置。