干等太心烦，教你怎么把TimechoAI的接口改成打字机效果
前面四篇文章，我们把从造数据、查数据库到拼提示词的活儿都干完了。代码跑起来，也能看到大模型给出的分析报告了。
但是呢，你如果真的自己跑过那些代码，你肯定会有一种感觉。就是等得太心烦了。代码卡在 requests.post 那一行，半天不动弹。你就盯着那个光标在闪烁，心里没底，不知道它是死了还是在算。如果数据量大一点，等个一分钟两分钟那是常有的事。
这种体验，在你自己写脚本调试的时候还能忍一忍。但是如果你要做一个给别人用的产品，比如一个Web页面，用户是绝对受不了这种干等的。所以今天这篇，我们要解决这个痛点。我们要把接口的调用方式改一改，改成流式输出。让它想出一个字，就给你吐一个字，像打字机一样。
一、 现在的调用方式有个致命缺点
1.1 一等就是一分钟，用户体验极差
我们先来看看之前我们写的那个 ask_timecho_ai 函数。它里面核心的一句话是 response = requests.post(…)。
这句话在底层是怎么干的呢？它是把你的请求发过去，然后就在那傻等着。服务器在后台开始算，算完了一个字，再算下一个字。等把所有的字都算完了，服务器把它们打包成一个大的JSON字符串，一次性扔回来。这时候 requests 才算拿到了结果，程序才继续往下走。
这个过程里，你的程序是阻塞住的。什么也干不了。这就叫同步阻塞调用。
对于短请求来说，这没啥。但是大模型生成几百字的分析报告，可能需要十几秒甚至几十秒。这几十秒对用户来说就是黑盒。他不知道系统是在努力干活，还是已经卡死了。如果超过半分钟没反应，大部分用户就会觉得这系统烂透了，直接关掉页面走人。
1.2 其实大模型是一个字一个字往外想的
那为什么我们在网页上用ChatGPT或者别的模型时，看到的是一个字一个字蹦出来的呢？
其实大模型的底层原理，是预测下一个词的概率。它根据前面的上下文，猜出第一个字。然后把这个字加到上下文里，再猜第二个字。它是边猜边出的。
也就是说，服务器那边并不是等全想好了才发数据。它是想出一个字，就有一个字可以往外发了。我们之前的调用方式，是服务器那边把所有字攒齐了一锅端给我们。如果我们能改一下方式，让服务器有一个字就发一个字，我们收到一个字就显示一个字。那体验不就丝滑了吗？
这就是流式输出的核心逻辑。说白了，就是把一锅端变成了流水席。
二、 什么是流式输出（Streaming / SSE）
2.1 别被SSE这个英文缩写吓到
一提到流式输出，很多教程就会甩出一个词叫SSE，全称是 Server-Sent Events。听着特别高大上，其实原理特别土。
你就把它想象成食堂打饭。以前的方式是，你点了个大份套餐，厨师在后面把饭、菜、汤全装好，放在一个托盘里，一次性端给你。这就是普通的请求。
SSE是什么呢？是你坐在桌位上，厨师做好了一道凉菜，先给你端上来。你先吃着。过两分钟，热菜炒好了，再给你端上来。再过一会，汤熬好了，再端上来。厨师和你的连接一直没断，他就是不停地把新做好的东西送过来。
在这个过程中，你不需要等所有菜都做好才能动筷子。你收到了一道菜，就可以吃一道。这就是SSE。服务端一直占着连接，有新数据产生就立马推过来。
2.2 对比一下普通请求和流式请求的报文区别
那在代码层面，怎么告诉服务器“我要吃流水席”呢？
很简单。你在发请求的时候，在参数里加一个 "stream": True。这就相当于你跟厨师说，“我赶时间，做好一道上一道”。
服务器看到这个参数，它就不会把结果打包成一个大JSON了。它会返回一个长连接。然后每想出一个词，就往这个连接里塞一个小JSON。直到全部想完，它发一个特殊的标记，比如 data: [DONE]，告诉你菜上齐了。
三、 代码怎么改才能接住流式数据
3.1 requests 库的隐藏开关：stream=True
我们之前用的 requests.post，其实也支持流式接收。只不过我们需要手动打开它。
在发请求的时候，你可以这样写：
response = requests.post(url, headers=headers, json=payload, stream=True)
注意这里多了一个 stream=True。你加上这个之后，requests 库的行为就变了。它不会再傻等服务器把所有数据都返回了。它只要一拿到响应头，确认连接建立了，就立刻把控制权还给你。
这时候 response 这个对象，就不包含完整的响应体了。它变成了一个管道。数据还在源源不断地从服务器流过来，存在这个管道里。你需要自己去把数据抠出来。
3.2 返回的东西变了，不能再 .json() 了
以前我们是 response.json()，一口气把所有数据转成字典。现在这招不管用了。因为数据还没流完呢，你调 .json() 必须报错，格式不对。
那怎么抠数据呢？requests 对象提供了一个方法，叫 iter_lines()。这个方法可以一行一行地去读管道里的数据。服务器每发过来一个小JSON，它就读出一行。
我们写个骨架看看：
for line in response.iter_lines():
if line:
# 把字节转成字符串
decoded_line = line.decode('utf-8')
print(decoded_line)
这段代码就能把你收到的每一行数据都打印出来。你运行一下，就能看到服务器是怎么一点点把数据吐过来的了。
四、 动手把之前的骨架函数重写一遍
4.1 处理那些没用的空行和前缀
上面那个骨架能跑，但是打印出来的东西很乱。因为SSE协议规定，每一条数据的前面都要加 data: 这个前缀。而且为了分隔，服务器会发很多空行过来。
我们要把这些没用的东西过滤掉。真正包含内容的是 data: {json字符串} 这种格式的行。
我们先把之前的函数拿过来，大改一版。
import requests
import json
def ask_timecho_ai_stream(question, api_key):
url = "https://ai.timecho.com/v1/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
payload = {
"model": "timecho-model",
"messages": [
{
"role": "user",
"content": question
}
],
"stream": True # 一定要加上这个，告诉服务器我们要流式
}
try:
# 记得这里也要加上stream=True
response = requests.post(url, headers=headers, json=payload, stream=True, timeout=60)
response.raise_for_status()
# 开始逐行读取
for line in response.iter_lines():
if not line:
continue # 空行直接跳过
decoded_line = line.decode('utf-8')
# 如果不是以 'data: ' 开头，也跳过，说明不是正经数据
if not decoded_line.startswith('data: '):
continue
# 把前缀 'data: ' 剁掉，拿到后面的JSON字符串
json_str = decoded_line[6:]
# 检查是不是结束标记
if json_str.strip() == '[DONE]':
print("\\n") # 结束了换个行，好看点
break
# 解析这个小JSON
try:
data_dict = json.loads(json_str)
# 抠出真正的文字内容
delta = data_dict['choices'][0]['delta']
if 'content' in delta:
word = delta['content']
# 打印的时候不换行，并且立刻刷新缓冲区
print(word, end="", flush=True)
except json.JSONDecodeError:
# 有时候偶尔解析出错不要紧，跳过等下一行就行
continue
except Exception as e:
print(f"\\n发生错误了：{e}")
这段代码看着比以前复杂多了。我一点点给你拆解。
首先看 payload 里，我们加了 "stream": True。这是跟服务器打申请的关键。
然后看那个 for 循环。我们在循环里先过滤空行。再检查是不是 data: 打头。如果是，我们就用切片 decoded_line[6:] 把前面那6个字符剁掉。剩下就是个JSON了。
接着我们要处理 [DONE] 这个标记。服务器把话说完了，会发一行 data: [DONE]。我们碰到这个，就知道活干完了，可以跳出循环了。
4.2 抠出内容并实现打字机打印
最核心的打印逻辑在最底下。那个小JSON的结构，跟以前一次性返回的大JSON有点不一样。流式返回的数据，内容藏在 delta 这个字段里。
为什么要叫 delta 呢？其实说白了，就是增量的意思。因为每次只吐一个字或者几个字，所以叫增量。
我们用 data_dict['choices'][0]['delta'] 把这个小字典拿出来。然后判断里面有没有 content 字段。如果有的话，就把里面的字拿出来。
最后一步特别要紧。我们用 print(word, end="", flush=True) 来打印。
如果不加 end=""，每打印一个字就会换行，那就成了竖着写了。加了 end=""，字就会横着往后排。那为什么还要加 flush=True 呢？因为Python的输出是有缓冲区的。有时候字到了缓冲区里，它不立马显示在屏幕上，要等攒够了一波再显示。加了 flush=True，就是强逼着它只要有字就立刻显示出来。这样你才能看到打字机的效果。
五、 跑一下看看效果，感受丝滑
5.1 真的像人在打字一样
现在我们把主流程改一下，调用这个新的流式函数。
if __name__ == "__main__":
my_key = "sk-你的真实KEY粘贴在这里"
my_question = "帮我分析一下，如果时序数据中存在周期性的毛刺，通常可能是什么原因导致的？请详细说明。"
print("=== 流式输出开始 ===")
ask_timecho_ai_stream(my_question, my_key)
你运行一下这段代码。你会发现，程序刚跑起来没一秒钟，第一个字就蹦出来了。接着就是匀速地往外吐字。整个过程非常跟手。
你再也不用盯着空白屏幕发呆了。你可以立刻知道模型在说什么，如果它跑偏了，你也能第一时间发现。
5.2 如果中途断了怎么办
流式输出有个小缺点，就是网络必须一直保持稳定。因为它是长连接。
如果你在打字的过程中，网络突然抖了一下，连接断了。你的代码会抛出一个连接异常。这时代码就跳到 except 那里去了。
这时候你怎么办呢？其实已经打出来的字还是在屏幕上的。你虽然没拿到完整的结果，但是至少不用像以前那样，等了一分钟最后啥也没落着。你可以在异常处理里，把已经收到的部分结果存下来，或者给用户提示说“网络中断，只生成了部分内容”。这在容灾处理上，其实是比一次性返回要更友好的情况。
六、 流式输出在真实业务里的用处
6.1 做Web聊天界面必须得用它
如果你以后想用TimechoAI做一个内部的运维助手网页。那你绝对必须肯定要用流式输出。
你想想看，如果用普通的接口。用户点一下“分析”，页面上那个圈圈转了30秒。这30秒里啥反馈也没有。用户的耐心早就磨没了。
换成流式输出，用户点一下，一秒钟内文字就出来了。虽然后面也是一直在出，但是用户感觉系统是在“说话”，是在干活。这种心理暗示的差别是非常巨大的。现在所有的大模型产品，网页端全都是流式的。没有哪家敢用阻塞请求做聊天界面。
如果你用Python的后端框架，比如FastAPI，配合前端的EventSource或者fetch的stream，就能很轻松地把这个打字机效果搬到网页上去。这个等我们专栏后面讲到实战项目的时候会细说。
6.2 提前中断省钱机制
流式输出还有一个隐藏的好处，就是能省钱。
大模型的计费是按输入和输出的Token算的。输出的Token越少，越便宜。
如果你用的是普通请求，你让模型写一个1000字的报告。它写到500字的时候，其实已经开始跑题胡说八道了。但是你管不了它，你必须等它写完1000字，返回给你了，你才知道它跑题了。这时候500个输出Token的钱已经扣了。
但是如果是流式输出，你看着它打字。打到第200字，你一看，坏了，方向不对。你可以直接按 Ctrl+C 把程序掐断，或者在前端点一个“停止生成”的按钮。这时候连接断了，模型就不会再往下生成了。你只消耗了200个Token的钱。这不就省下一大半吗？
这种及时止损的能力，在实际高频调用的时候，能帮你省下不少费用。你可以去后台 https://ai.timecho.com/settings/keys 看看账单，如果发现很多无效的长输出，就可以考虑用这种掐断机制来控制成本。
七、 别忘了去示例页面感受一下原汁原味的效果
7.1 打开官方页面看看
纸上得来终觉浅。我强烈建议你现在就打开浏览器，去这个地址：https://ai.timecho.com/realtime
你在输入框里随便打个复杂点的问题，比如“详细解释一下时间序列数据库的存储引擎原理”。
然后你仔细观察它回答的过程。你会发现，它就是一个字一个字往外蹦的。中间偶尔会停顿零点几秒，那是模型在算下一个词的概率。
这就是标准的流式输出效果。我们在代码里搞出来的效果，跟这个是一模一样的。只不过它是在网页上，我们是在终端控制台里而已。
7.2 去文档里确认流式参数怎么传
另外，养成好习惯，去翻一翻官方的开发文档：https://ai.timecho.com/docs/
你在里面搜一下 stream 这个关键词。看看官方有没有什么特殊的说明。比如有些接口可能不光要在 payload 里加 "stream": True，可能还要你改一下请求的URL，比如变成 /v1/chat/completions-stream 之类的（当然TimechoAI目前不需要改URL，按我们写的这么加就行）。
文档里一般也会把SSE返回的数据格式写得很清楚。告诉你 delta 字段长什么样，结束标记 [DONE] 是什么格式。我上面写的代码就是按照标准的OpenAI兼容格式来的，大部分大模型厂商都遵循这个规范。但是保不齐哪天官方改了，所以你自己去确认一遍最稳妥。
八、 总结与下期预告
8.1 今天的核心就一个词：体验
今天这篇，我们没有引入什么新的业务逻辑。我们还是在调同一个接口，查同样的数据，问同样的问题。我们只做了一件事情，就是把请求方式从一锅端改成了流水席。
但是就这一步改动，带来的体验提升是天壤之别。从黑盒一样的傻等，变成了看得见摸得着的实时反馈。这其实也给我们一个启示：在做AI应用的时候，技术参数固然重要，但是用户的心智感受往往仅仅只需要一个简单的流式输出就能极大改善。
我们今天写的那个 ask_timecho_ai_stream 函数，你一定要保存好。以后我们会把它当成标准组件来用。
8.2 下期我们讲讲：能不能让模型自己去查数据库
到目前为止，我们的流程都是我们自己写代码查数据库，然后把数据拼成文本喂给模型。这其实有点累。我们在代码里既要搞数据库连接，又要搞大模型连接，两头忙。
你有没有想过一个问题：既然大模型那么聪明，我们能不能直接把数据库的连接信息告诉它，让它自己去写SQL，自己去查数据，自己分析？
这在技术上是有名字的，叫 Function Calling，或者叫工具调用。也就是让大模型长出手脚来，能自己调用外部的工具。
如果这套能跑通，我们写代码的工作量就能大幅度减少。我们只需要写好查数据的工具，注册给模型就行。具体查什么，怎么查，全由模型自己定。
这个话题稍微有点难度，但是非常非常有价值。这是目前大模型落地的最热门的架构。所以下一期，我们就来硬啃这块骨头。我们下篇见。
