大家好，我是技术博客作者。上周有个做独立开发的朋友问我："我想在网页上实时显示 AI 的回复，就像 ChatGPT 那样打字效果，怎么实现啊？"今天我就来手把手教大家用 LangChain 的 StreamingCallbackHandler 实现这个效果。
读完这篇教程，你会学到：什么是流式输出、CallbackHandler 是什么、怎么在 LangChain 里用流式输出、以及怎么避开新手常犯的错误。全文代码都可以直接复制运行，建议收藏！
一、流式输出是什么？为什么要用它？
先科普一下概念。普通模式下，你问 AI 一个问题，AI 要把整段回答生成完才返回给你，可能要等好几秒。流式输出（Streaming）则是 AI 边生成边返回，一个字一个字地"吐"出来，前端实时显示，用户体验好很多。
用 HolySheep AI（立即注册）的话，国内直连延迟可以控制在 50ms 以内，流式输出体验非常丝滑，完全感觉不到延迟。
二、环境准备
2.1 安装依赖
首先我们安装必要的包，在命令行运行：
pip install langchain langchain-openai langchain-core2.2 配置 API Key
工欲善其事，必先利其器。我强烈推荐用 HolySheep AI 的接口，原因很简单：汇率 ¥1=$1 无损（官方是 ¥7.3=$1），对我们国内开发者来说省了超过 85% 的成本！支持微信和支付宝充值，实时到账。
先获取你的 API Key，登录 HolySheep AI 后台，在个人中心可以看到。推荐使用 DeepSeek V3.2 或者 GPT-4o-mini 这类性价比极高的模型。
然后设置环境变量：
import os
设置 HolySheep API Key
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
重要！必须设置这个 base_url，否则会请求 OpenAI 官方接口
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"三、StreamingCallbackHandler 核心原理
3.1 什么是 CallbackHandler？
CallbackHandler 就像一个"监听器"。LangChain 在执行过程中会触发各种事件（比如开始生成、生成新 token、结束生成），CallbackHandler 可以捕捉这些事件，让我们做一些自定义操作。
我们主要用 on_llm_new_token 这个方法——每当 AI 生成一个新的 token（可以理解为一个字/词），这个方法就会被调用，我们就在这里把新内容打印出来或者更新界面。
3.2 第一个流式输出示例
先来一个最简单的例子，感受一下流式输出是怎么工作的：
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import StreamingStdOutCallbackHandler
创建带流式输出的模型
llm = ChatOpenAI(
model="gpt-4o-mini", # 也可以用 deepseek-chat 等模型
temperature=0.7,
streaming=True, # 必须设置为 True 才能开启流式输出
callbacks=[StreamingStdOutCallbackHandler()] # 这个处理器会把内容打印到终端
)
简单测试一下
print("AI 正在回答：")
response = llm.invoke("请用三句话介绍自己")
print("\n回答完毕！")运行这段代码，你应该能看到 AI 的回答是逐字逐句"打"出来的，而不是一次性显示完整内容。非常神奇对不对！
四、自定义 CallbackHandler
4.1 为什么需要自定义？
StreamingStdOutCallbackHandler 只能打印到终端，实际项目中我们通常需要：
- 把内容显示到网页上
- 保存到变量里后续处理
- 统计 token 消耗
- 显示打字光标动画
这时候就要自定义 CallbackHandler 了。
4.2 自定义流式处理器
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Optional, List
class MyStreamHandler(BaseCallbackHandler):
"""自定义流式输出处理器"""
def __init__(self, container, initial_text=""):
self.container = container # 前端容器（Streamlit 中用 st.empty()）
self.text = initial_text
self.token_count = 0
def on_llm_new_token(self, token: str, **kwargs) -> None:
"""每当生成新 token 时调用"""
self.text += token
self.token_count += 1
# 这里你可以替换成任何 UI 更新逻辑
print(token, end="", flush=True)
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
"""LLM 调用结束时调用"""
print(f"\n\n[统计] 共生成 {self.token_count} 个 token")
def get_text(self) -> str:
"""获取完整文本"""
return self.text
使用自定义处理器
llm = ChatOpenAI(
model="gpt-4o-mini",
streaming=True,
callbacks=[MyStreamHandler(None)] # 传入容器
)
print("开始流式输出测试：")
response = llm.invoke("写一首关于春天的诗")我在实际项目中发现，自定义 Handler 最实用的场景是统计 token 消耗。上面代码里的 on_llm_end 方法可以获取完整的 LLMResult，里面包含 token 用量信息，方便做成本控制。
五、完整 Web 应用示例（Streamlit）
现在来做个真正能用的网页应用。用户输入问题，AI 实时流式回复，就像 ChatGPT 那样：
import streamlit as st
import os
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler
配置页面
st.set_page_config(page_title="AI 流式对话助手", page_icon="🤖")
st.title("🤖 AI 流式对话助手")
侧边栏设置
with st.sidebar:
st.header("⚙️ 设置")
# 输入 API Key（也可以用环境变量）
api_key = st.text_input("HolySheep API Key", type="password")
if api_key:
os.environ["OPENAI_API_KEY"] = api_key
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
# 选择模型
model_options = {
"GPT-4o Mini": "gpt-4o-mini",
"DeepSeek V3.2": "deepseek-chat",
"Claude 3.5 Sonnet": "claude-3-5-sonnet-20240620"
}
selected_model = st.selectbox("选择模型", list(model_options.keys()))
# 费用说明
st.markdown("---")
st.markdown("**💰 模型价格参考（/百万输出Token）**")
st.markdown("- GPT-4o Mini: $8")
st.markdown("- DeepSeek V3.2: **$0.42**（强烈推荐！）")
st.markdown("- Claude 3.5 Sonnet: $15")
自定义流式 Handler
class StreamHandler(BaseCallbackHandler):
def __init__(self, msg):
self.msg = msg
self.text = ""
def on_llm_new_token(self, token, **kwargs):
self.text += token
self.msg.markdown(self.text + "▌")
初始化聊天历史
if "messages" not in st.session_state:
st.session_state.messages = []
显示历史消息
for msg in st.session_state.messages:
with st.chat_message(msg["role"]):
st.markdown(msg["content"])
聊天输入
if prompt := st.chat_input("问我任何问题..."):
# 显示用户消息
with st.chat_message("user"):
st.markdown(prompt)
st.session_state.messages.append({"role": "user", "content": prompt})
# 生成 AI 回复
with st.chat_message("assistant"):
msg = st.empty()
stream_handler = StreamHandler(msg)
try:
llm = ChatOpenAI(
model=model_options[selected_model],
streaming=True,
callbacks=[stream_handler]
)
response = llm.invoke(prompt)
final_text = stream_handler.text
msg.markdown(final_text) # 去掉光标
st.session_state.messages.append({"role": "assistant", "content": final_text})
except Exception as e:
st.error(f"出错了: {str(e)}")
底部提示
st.markdown("---")
st.markdown("👉 使用 HolySheep API 国内直连延迟 <50ms，支持微信/支付宝充值")运行命令：streamlit run app.py
部署这个应用后，你就有了一个私人版的 ChatGPT！我用 DeepSeek V3.2 测试，响应速度非常快，而且成本极低——每百万 token 才 $0.42，性价比之王。
六、常见报错排查
报错1：AuthenticationError: Incorrect API key provided
原因： API Key 填写错误，或者没有设置 OPENAI_API_BASE 导致请求发到了错误的地址。
解决代码：
# 正确配置示例
import os
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY" # 替换成你的真实 Key
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1" # 必须是这个地址！
验证配置是否正确
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
print(llm.model) # 如果没有报错说明配置正确报错2：Stream is not iterable 或 TypeError
原因： 创建 ChatOpenAI 时忘记设置 streaming=True，但使用了流式 CallbackHandler。
解决代码：
# ❌ 错误写法
llm = ChatOpenAI(callbacks=[StreamingStdOutCallbackHandler()])
✅ 正确写法
llm = ChatOpenAI(
streaming=True, # 必须显式设置为 True
callbacks=[StreamingStdOutCallbackHandler()]
)报错3：RateLimitError 或 ConnectionError
原因： 请求被限流或者网络连接问题。如果是国内访问 OpenAI 官方接口，很可能是网络问题。
解决代码：
# 方案1：使用 HolySheep API（国内直连，延迟 <50ms）
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
方案2：添加重试机制
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_with_retry(llm, prompt):
return llm.invoke(prompt)
方案3：增加超时时间
llm = ChatOpenAI(
timeout=120, # 120秒超时
max_retries=3
)报错4：AttributeError: 'NoneType' object has no attribute 'xxx'
原因： 自定义 CallbackHandler 中容器对象为 None，或者在异步环境下使用不当。
解决代码：
# 确保容器不为 None
class SafeStreamHandler(BaseCallbackHandler):
def __init__(self, container=None):
self.container = container
self.text = ""
def on_llm_new_token(self, token, **kwargs):
self.text += token
if self.container:
self.container.markdown(self.text + "▌")
else:
print(token, end="", flush=True)
def on_llm_end(self, response, **kwargs):
# 安全获取 token 用量
if hasattr(response, 'llm_output') and response.llm_output:
token_usage = response.llm_output.get('token_usage', {})
print(f"\nToken 消耗: {token_usage}")
七、进阶技巧
7.1 同时支持多个输出目标
有时候你可能需要同时输出到终端和网页，可以创建多个 Handler：
# 同时支持终端和 Streamlit
from langchain_core.callbacks import StreamingStdOutCallbackHandler
llm = ChatOpenAI(
model="gpt-4o-mini",
streaming=True,
callbacks=[
StreamingStdOutCallbackHandler(), # 打印到终端
StreamHandler(msg) # 更新网页
]
)7.2 异步流式处理
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import AsyncCallbackHandler
class AsyncStreamHandler(AsyncCallbackHandler):
async def on_llm_new_token(self, token, **kwargs):
# 可以在这里 await 任何异步操作
print(token, end="", flush=True)
await asyncio.sleep(0) # 让出控制权
async def main():
llm = ChatOpenAI(
model="gpt-4o-mini",
streaming=True,
callbacks=[AsyncStreamHandler()]
)
await llm.ainvoke("介绍一下人工智能的发展历史")
asyncio.run(main())八、总结
今天我们学了：
- 什么是流式输出以及为什么重要
- LangChain StreamingCallbackHandler 的基本用法
- 如何自定义 Handler 实现各种场景需求
- 用 Streamlit 做了一个完整的 AI 对话应用
- 4 个最常见错误的排查方法
用 HolySheep AI 的流式接口，体验非常流畅。国内直连延迟低至 50ms，汇率更是逆天——¥1=$1 比官方省了 85% 以上。DeepSeek V3.2 模型每百万输出 token 才 $0.42，性价比拉满，非常适合做 AI 应用开发。
如果这篇文章对你有帮助，欢迎转发给需要的朋友！有其他问题可以在评论区留言，我会尽量解答。