# Python 实现 LLM 流式输出实战指南
在大语言模型应用火遍全球的今天，响应速度直接决定了用户体验。同步调用 API 的痛点很明显：用户对着空白屏幕等半天，才能看到完整回复。10 秒的等待时间漫长得像一个世纪。流式输出技术就是为了解决这个问题诞生的——模型边生成边返回，用户看着字一个个蹦出来，感知上快多了。
流式输出的本质是利用 SSE 协议，通过持久连接持续传输数据块。技术上不难，但实际开发中很多人卡在各种细节上。本文会手把手教你用 Python 对接 OpenAI、Anthropic Claude 和本地 Ollama 三种主流场景，给出完整可运行的代码。
## 实际开发中的困惑
很多开发者想实现类似 ChatGPT 的打字机效果，但不知道从哪下手。不同 API 有什么坑？怎么处理网络断开的情况？流式响应怎么和现有代码集成？
传统 HTTP 请求是“发出去等结果”模式。LLM 生成一段 1000 字的文章可能要 10 秒甚至更久，用户只能对着转圈圈干着急。流式输出改变了这个流程——模型刚开始吐字，数据就通过建立的连接流向客户端。
下面直接上代码，覆盖三个最常见的场景。
## 环境准备
先装必要的包：
“`bash
pip install requests sseclient-py openapi anthropic
“`
## OpenAI GPT 流式输出
OpenAI 的接口设计最简洁，官方直接支持流式模式：
“`python
import requests
import json
def stream_openai_response(api_key: str, model: str = “gpt-4”,
message: str = “介绍一下 Python 的流式输出”) -> None:
“””
使用 OpenAI API 实现流式输出
“””
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {api_key}”,
“Content-Type”: “application/json”
}
payload = {
“model”: model,
“messages”: [{“role”: “user”, “content”: message}],
“stream”: True # 启用流式输出的关键参数
}
response = requests.post(url, headers=headers, json=payload, stream=True)
print(f”[OpenAI {model}] 流式响应：”)
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
data = line[6:]
if data == “[DONE]”:
break
try:
chunk = json.loads(data)
content = chunk.get(“choices”, [{}])[0].get(“delta”, {}).get(“content”, “”)
if content:
print(content, end=””, flush=True)
except json.JSONDecodeError:
continue
print()
if __name__ == “__main__”:
OPENAI_API_KEY = “your-api-key-here”
stream_openai_response(OPENAI_API_KEY)
“`
关键是 `stream: True` 这个参数。返回的数据是分块的 JSON，每块包含 `delta.content`，需要我们自己拼接。
## Anthropic Claude 流式输出
Claude 的接口和 OpenAI 差不多，但 header 和响应结构有些区别：
“`python
import requests
import json
def stream_claude_response(api_key: str, message: str = “解释什么是大语言模型”) -> None:
“””
使用 Anthropic Claude API 实现流式输出
“””
url = “https://api.anthropic.com/v1/messages”
headers = {
“x-api-key”: api_key,
“anthropic-version”: “2023-06-01”,
“Content-Type”: “application/json”
}
payload = {
“model”: “claude-3-5-sonnet-20241022”,
“max_tokens”: 1024,
“messages”: [{“role”: “user”, “content”: message}],
“stream”: True
}
response = requests.post(url, headers=headers, json=payload, stream=True)
print(“[Claude 3.5 Sonnet] 流式响应：”)
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
data = line[6:]
try:
chunk = json.loads(data)
delta = chunk.get(“delta”, {})
content = delta.get(“text”, “”)
if content:
print(content, end=””, flush=True)
except (json.JSONDecodeError, KeyError):
continue
print()
if __name__ == “__main__”:
CLAUDE_API_KEY = “sk-ant-api03-your-key-here”
stream_claude_response(CLAUDE_API_KEY)
“`
注意 Claude 用 `x-api-key` 而不是 Authorization header，而且需要指定 `anthropic-version`。
## Ollama 本地模型流式输出
Ollama 是目前最火的本地大模型运行工具，在自己电脑上就能跑 Llama、Mistral 这些开源模型：
“`python
import requests
import json
def stream_ollama_response(model: str = “llama3”,
message: str = “用一句话解释 Python”) -> None:
“””
使用 Ollama 本地模型实现流式输出
“””
url = “http://localhost:11434/api/chat”
payload = {
“model”: model,
“messages”: [{“role”: “user”, “content”: message}],
“stream”: True
}
response = requests.post(url, json=payload, stream=True)
print(f”[Ollama {model}] 流式响应：”)
for line in response.iter_lines():
if line:
try:
chunk = json.loads(line)
content = chunk.get(“message”, {}).get(“content”, “”)
if content:
print(content, end=””, flush=True)
except json.JSONDecodeError:
continue
print()
if __name__ == “__main__”:
stream_ollama_response()
“`
Ollama 的接口更简洁，不需要 API key，但需要先在本地运行 `ollama serve`。
## 封装一个通用类
实际项目中可能需要对接到不同模型，我封装了一个通用客户端：
“`python
import requests
import json
from typing import Callable, Optional
from enum import Enum
class LLMProvider(Enum):
OPENAI = “openai”
CLAUDE = “claude”
OLLAMA = “ollama”
class StreamingLLM:
“””通用流式 LLM 客户端”””
def __init__(self, provider: LLMProvider, **config):
self.provider = provider
self.config = config
def stream(self, prompt: str, on_chunk: Optional[Callable[[str], None]] = None) -> str:
“””
流式调用 LLM
Args:
prompt: 用户输入
on_chunk: 可选的回调函数，每收到一个数据块时调用
Returns:
完整的响应文本
“””
full_response = “”
if self.provider == LLMProvider.OPENAI:
full_response = self._stream_openai(prompt, on_chunk)
elif self.provider == LLMProvider.CLAUDE:
full_response = self._stream_claude(prompt, on_chunk)
elif self.provider == LLMProvider.OLLAMA:
full_response = self._stream_ollama(prompt, on_chunk)
else:
raise ValueError(f”Unsupported provider: {self.provider}”)
return full_response
def _stream_openai(self, prompt: str, on_chunk: Optional[Callable]) -> str:
url = “https://api.openai.com/v1/chat/completions”
headers = {
“Authorization”: f”Bearer {self.config[“api_key”]}”,
“Content-Type”: “application/json”
}
payload = {
“model”: self.config.get(“model”, “gpt-4”),
“messages”: [{“role”: “user”, “content”: prompt}],
“stream”: True
}
response = requests.post(url, headers=headers, json=payload, stream=True)
result = []
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
data = line[6:]
if data == “[DONE]”:
break
chunk = json.loads(data)
content = chunk.get(“choices”, [{}])[0].get(“delta”, {}).get(“content”, “”)
if content:
result.append(content)
if on_chunk:
on_chunk(content)
return “”.join(result)
def _stream_claude(self, prompt: str, on_chunk: Optional[Callable]) -> str:
url = “https://api.anthropic.com/v1/messages”
headers = {
“x-api-key”: self.config[“api_key”],
“anthropic-version”: “2023-06-01”,
“Content-Type”: “application/json”
}
payload = {
“model”: self.config.get(“model”, “claude-3-5-sonnet-20241022”),
“max_tokens”: 1024,
“messages”: [{“role”: “user”, “content”: prompt}],
“stream”: True
}
response = requests.post(url, headers=headers, json=payload, stream=True)
result = []
for line in response.iter_lines():
if line:
line = line.decode(“utf-8”)
if line.startswith(“data: “):
chunk = json.loads(line[6:])
content = chunk.get(“delta”, {}).get(“text”, “”)
if content:
result.append(content)
if on_chunk:
on_chunk(content)
return “”.join(result)
def _stream_ollama(self, prompt: str, on_chunk: Optional[Callable]) -> str:
url = “http://localhost:11434/api/chat”
payload = {
“model”: self.config.get(“model”, “llama3”),
“messages”: [{“role”: “user”, “content”: prompt}],
“stream”: True
}
response = requests.post(url, json=payload, stream=True)
result = []
for line in response.iter_lines():
if line:
chunk = json.loads(line)
content = chunk.get(“message”, {}).get(“content”, “”)
if content:
result.append(content)
if on_chunk:
on_chunk(content)
return “”.join(result)
if __name__ == “__main__”:
client = StreamingLLM(
provider=LLMProvider.OPENAI,
api_key=”your-openai-key”
)
def print_chunk(text: str):
print(text, end=””, flush=True)
print(“>>> “)
response = client.stream(“用三句话介绍Python语言”, on_chunk=print_chunk)
print(“\n”)
print(f”完整响应长度: {len(response)} 字符”)
“`
这个类支持切换不同模型，回调函数可以用于实时更新 UI。
## 跑起来看看效果
执行代码，你会看到字符一个个往外蹦：
“`
>>> [OpenAI gpt-4] 流式响应：
Python 是一种高级编程语言，由 Guido van Rossum 于 1991 年首次发布。它以简洁的语法和强大的功能而闻名，支持多种编程范式。Python 拥有丰富的标准库和第三方包，广泛应用于 Web 开发、数据分析、人工智能等领域。
完整响应长度: 98 字符
“`
本地 Ollama 也是一样的效果：
“`
>>> [Ollama llama3] 流式响应：
Python 是一种简洁优雅的编程语言，强调代码可读性，拥有强大的数据处理和科学计算能力。
完整响应长度: 45 字符
“`
## 踩过的坑
实际用起来有几点要留意：
第一，OpenAI 返回 `[DONE]` 就表示结束了，但 Claude 可能还有其他字段，需要做好容错。第二，网络断开是常有的事，最好加个重试机制。第三，本地模型跑流式输出对显卡有要求，显存不够会卡。第四，生产环境记得加请求限流，别一下子上太多并发。
流式输出已经是现代 LLM 应用的标配。把这套东西搞定了，做什么 AI 产品都不虚。