干等太心烦，教你怎么把TimechoAI的接口改成打字机效果

干等太心烦教你怎么把TimechoAI的接口改成打字机效果前面四篇文章我们把从造数据、查数据库到拼提示词的活儿都干完了。代码跑起来也能看到大模型给出的分析报告了。但是呢你如果真的自己跑过那些代码你肯定会有一种感觉。就是等得太心烦了。代码卡在requests.post那一行半天不动弹。你就盯着那个光标在闪烁心里没底不知道它是死了还是在算。如果数据量大一点等个一分钟两分钟那是常有的事。这种体验在你自己写脚本调试的时候还能忍一忍。但是如果你要做一个给别人用的产品比如一个Web页面用户是绝对受不了这种干等的。所以今天这篇我们要解决这个痛点。我们要把接口的调用方式改一改改成流式输出。让它想出一个字就给你吐一个字像打字机一样。一、 现在的调用方式有个致命缺点1.1 一等就是一分钟用户体验极差我们先来看看之前我们写的那个ask_timecho_ai函数。它里面核心的一句话是response requests.post(...)。这句话在底层是怎么干的呢它是把你的请求发过去然后就在那傻等着。服务器在后台开始算算完了一个字再算下一个字。等把所有的字都算完了服务器把它们打包成一个大的JSON字符串一次性扔回来。这时候requests才算拿到了结果程序才继续往下走。这个过程里你的程序是阻塞住的。什么也干不了。这就叫同步阻塞调用。对于短请求来说这没啥。但是大模型生成几百字的分析报告可能需要十几秒甚至几十秒。这几十秒对用户来说就是黑盒。他不知道系统是在努力干活还是已经卡死了。如果超过半分钟没反应大部分用户就会觉得这系统烂透了直接关掉页面走人。1.2 其实大模型是一个字一个字往外想的那为什么我们在网页上用ChatGPT或者别的模型时看到的是一个字一个字蹦出来的呢其实大模型的底层原理是预测下一个词的概率。它根据前面的上下文猜出第一个字。然后把这个字加到上下文里再猜第二个字。它是边猜边出的。也就是说服务器那边并不是等全想好了才发数据。它是想出一个字就有一个字可以往外发了。我们之前的调用方式是服务器那边把所有字攒齐了一锅端给我们。如果我们能改一下方式让服务器有一个字就发一个字我们收到一个字就显示一个字。那体验不就丝滑了吗这就是流式输出的核心逻辑。说白了就是把一锅端变成了流水席。二、 什么是流式输出Streaming / SSE2.1 别被SSE这个英文缩写吓到一提到流式输出很多教程就会甩出一个词叫SSE全称是 Server-Sent Events。听着特别高大上其实原理特别土。你就把它想象成食堂打饭。以前的方式是你点了个大份套餐厨师在后面把饭、菜、汤全装好放在一个托盘里一次性端给你。这就是普通的请求。SSE是什么呢是你坐在桌位上厨师做好了一道凉菜先给你端上来。你先吃着。过两分钟热菜炒好了再给你端上来。再过一会汤熬好了再端上来。厨师和你的连接一直没断他就是不停地把新做好的东西送过来。在这个过程中你不需要等所有菜都做好才能动筷子。你收到了一道菜就可以吃一道。这就是SSE。服务端一直占着连接有新数据产生就立马推过来。2.2 对比一下普通请求和流式请求的报文区别那在代码层面怎么告诉服务器“我要吃流水席”呢很简单。你在发请求的时候在参数里加一个stream: True。这就相当于你跟厨师说“我赶时间做好一道上一道”。服务器看到这个参数它就不会把结果打包成一个大JSON了。它会返回一个长连接。然后每想出一个词就往这个连接里塞一个小JSON。直到全部想完它发一个特殊的标记比如data: [DONE]告诉你菜上齐了。三、 代码怎么改才能接住流式数据3.1requests库的隐藏开关streamTrue我们之前用的requests.post其实也支持流式接收。只不过我们需要手动打开它。在发请求的时候你可以这样写responserequests.post(url,headersheaders,jsonpayload,streamTrue)注意这里多了一个streamTrue。你加上这个之后requests库的行为就变了。它不会再傻等服务器把所有数据都返回了。它只要一拿到响应头确认连接建立了就立刻把控制权还给你。这时候response这个对象就不包含完整的响应体了。它变成了一个管道。数据还在源源不断地从服务器流过来存在这个管道里。你需要自己去把数据抠出来。3.2 返回的东西变了不能再.json()了以前我们是response.json()一口气把所有数据转成字典。现在这招不管用了。因为数据还没流完呢你调.json()必须报错格式不对。那怎么抠数据呢requests对象提供了一个方法叫iter_lines()。这个方法可以一行一行地去读管道里的数据。服务器每发过来一个小JSON它就读出一行。我们写个骨架看看forlineinresponse.iter_lines():ifline:# 把字节转成字符串decoded_lineline.decode(utf-8)print(decoded_line)这段代码就能把你收到的每一行数据都打印出来。你运行一下就能看到服务器是怎么一点点把数据吐过来的了。四、 动手把之前的骨架函数重写一遍4.1 处理那些没用的空行和前缀上面那个骨架能跑但是打印出来的东西很乱。因为SSE协议规定每一条数据的前面都要加data:这个前缀。而且为了分隔服务器会发很多空行过来。我们要把这些没用的东西过滤掉。真正包含内容的是data: {json字符串}这种格式的行。我们先把之前的函数拿过来大改一版。importrequestsimportjsondefask_timecho_ai_stream(question,api_key):urlhttps://ai.timecho.com/v1/chat/completionsheaders{Content-Type:application/json,Authorization:fBearer{api_key}}payload{model:timecho-model,messages:[{role:user,content:question}],stream:True# 一定要加上这个告诉服务器我们要流式}try:# 记得这里也要加上streamTrueresponserequests.post(url,headersheaders,jsonpayload,streamTrue,timeout60)response.raise_for_status()# 开始逐行读取forlineinresponse.iter_lines():ifnotline:continue# 空行直接跳过decoded_lineline.decode(utf-8)# 如果不是以 data: 开头也跳过说明不是正经数据ifnotdecoded_line.startswith(data: ):continue# 把前缀 data: 剁掉拿到后面的JSON字符串json_strdecoded_line[6:]# 检查是不是结束标记ifjson_str.strip()[DONE]:print(\n)# 结束了换个行好看点break# 解析这个小JSONtry:data_dictjson.loads(json_str)# 抠出真正的文字内容deltadata_dict[choices][0][delta]ifcontentindelta:worddelta[content]# 打印的时候不换行并且立刻刷新缓冲区print(word,end,flushTrue)exceptjson.JSONDecodeError:# 有时候偶尔解析出错不要紧跳过等下一行就行continueexceptExceptionase: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, flushTrue)来打印。如果不加end每打印一个字就会换行那就成了竖着写了。加了end字就会横着往后排。那为什么还要加flushTrue呢因为Python的输出是有缓冲区的。有时候字到了缓冲区里它不立马显示在屏幕上要等攒够了一波再显示。加了flushTrue就是强逼着它只要有字就立刻显示出来。这样你才能看到打字机的效果。五、 跑一下看看效果感受丝滑5.1 真的像人在打字一样现在我们把主流程改一下调用这个新的流式函数。if__name____main__:my_keysk-你的真实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字你一看坏了方向不对。你可以直接按CtrlC把程序掐断或者在前端点一个“停止生成”的按钮。这时候连接断了模型就不会再往下生成了。你只消耗了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或者叫工具调用。也就是让大模型长出手脚来能自己调用外部的工具。如果这套能跑通我们写代码的工作量就能大幅度减少。我们只需要写好查数据的工具注册给模型就行。具体查什么怎么查全由模型自己定。这个话题稍微有点难度但是非常非常有价值。这是目前大模型落地的最热门的架构。所以下一期我们就来硬啃这块骨头。我们下篇见。