LLM应用开发与落地：流式响应

一、背景

最近智能客服产品给到一个游戏客户那边，客户那边的客服负责人体验后认为我们产品回答的准确率是还是比较高的。同时，他反馈了几个需要改进的地方，其中一个就是机器人回复慢。机器人回复慢有很多原因，也有优化方式，其中一个就是流式响应。

二、原理

我们在微信需要发送比较长一段文字的时候，我们需要花比较长的时间去写，跟你聊天的人那边的感触就是要有一段时间的等待。如果我们每写好一句话就先发送过去，对方的等待的感觉就会弱一点。推到极致就是，我们每写一个字就发送过去，这样对方的等待感是最弱的。当然，人不可能这么做 ，因为我们要检查和改动我们写的内容。但是LLM可以，因为LLM也是一个一个token生成，而且不需要检查和改动。

LLM应用要实现流式响应，其实需要三个点都支持流式响应。

首先，LLM生成响应的时候，每生成一部分要先提前返回。百度的ernie-bot-4 是支持流式响应的，只需要请求body中带上参数 stream=True。

其次，应用服务器跟LLM服务器间的数据通道需要支持流式响应，一般http接口的封装都有这个参数，比如python request包的post方法的stream参数。

最后，应用服务器返回给前端的数据通道，以及前端展示要支持流式响应。gradio的chatbot也是支持流式响应的。

三、实践

万事俱备，就等coding。

LLM和数据通道这块都只是加个参数，这个改动不大。我们来看下gradio的流式怎么实现。

如果我们要用gradio实现一个回显的demo（就是我们发送什么，服务器就返回什么），我们会这么实现

import random
import gradio as grdef echo_response(message, history):return "你输入：" + messagegr.ChatInterface(echo_response).launch()

输出效果如下：

实现一个流式响应的也很简单，区别就是把响应函数变成一个生成器，每次返回最新的消息：

import time
import gradio as grdef echo_response(message, history):for i in range(len(message)):time.sleep(0.3)yield "你输入: " + message[: i+1]gr.ChatInterface(echo_response).launch()

为了看到回显消息一个字一个字出来，故意每增加输出一个字延迟0.3秒。看的的效果就是“你输入：”后边的字一个一个显示出来。

接下来就是把LLM的流式响应参数和http通道的流式响应的参数设置为True，然后把gradio的响应函数改成生成器即可。

使用百度 ernie-bot 和 gradio 写了个demo，把全流程串起来验证了下，log中可以看到ernie-bot的分批返回：

{'id': 'as-gxfvpsrx35', 'object': 'chat.completion', 'created': 1707351331, 'sentence_id': 0, 'is_end': False, 'is_truncated': False, 'result': '我是百度公司', 'need_clear_history': False, 'finish_reason': 'normal', 'usage': {'prompt_tokens': 2, 'completion_tokens': 0, 'total_tokens': 2}}
stream result:我是百度公司
{'id': 'as-gxfvpsrx35', 'object': 'chat.completion', 'created': 1707351334, 'sentence_id': 1, 'is_end': False, 'is_truncated': False, 'result': '开发的人工智能语言模型，我的中文名是文心一言，英文名是ERNIE Bot，我可以为人类提供信息解决问题，比如回答问题，提供定义、解释', 'need_clear_history': False, 'finish_reason': 'normal', 'usage': {'prompt_tokens': 2, 'completion_tokens': 0, 'total_tokens': 2}}
stream result:开发的人工智能语言模型，我的中文名是文心一言，英文名是ERNIE Bot，我可以为人类提供信息解决问题，比如回答问题，提供定义、解释
{'id': 'as-gxfvpsrx35', 'object': 'chat.completion', 'created': 1707351336, 'sentence_id': 2, 'is_end': False, 'is_truncated': False, 'result': '和建议，也可以辅助人类进行创作产生新的内容，如文本生成与创作、文本改写等。', 'need_clear_history': False, 'finish_reason': 'normal', 'usage': {'prompt_tokens': 2, 'completion_tokens': 0, 'total_tokens': 2}}
stream result:和建议，也可以辅助人类进行创作产生新的内容，如文本生成与创作、文本改写等。
{'id': 'as-gxfvpsrx35', 'object': 'chat.completion', 'created': 1707351337, 'sentence_id': 3, 'is_end': False, 'is_truncated': False, 'result': '如果您有任何问题，请随时向我提问。', 'need_clear_history': False, 'finish_reason': 'normal', 'usage': {'prompt_tokens': 2, 'completion_tokens': 0, 'total_tokens': 2}}
stream result:如果您有任何问题，请随时向我提问。
{'id': 'as-gxfvpsrx35', 'object': 'chat.completion', 'created': 1707351337, 'sentence_id': 4, 'is_end': True, 'is_truncated': False, 'result': '', 'need_clear_history': False, 'finish_reason': 'normal', 'usage': {'prompt_tokens': 2, 'completion_tokens': 62, 'total_tokens': 64}}
stream result:

文章的最后有demo代码，感兴趣的可以自己验证下。

四、感悟

demo好写，但是发现在智能机器人项目支持这个还是有点改动，从agent framework 到具体agent的实现，整个返回的线路都要改成生成器模式。

最近感悟就是，简单的RAG也好，让人惊喜的AGI也好，都非常容易实现，但是真正要落地到企业中，非常多的坑和需要探索解决的东西，无论是准确率、响应速度、成本，以及使用人的接受度等等。这也是很大一部分人短暂接触LLM后放弃的原因。

最近跟一些在做LLM应用的人聊，大家其实真的还是需要信仰去支持调prompt，去探索新的方法。毕竟，这真的是没有现成的技术方案可以参考，就跟20年前怎么实现服务器高并发一样。那时候，C10K 问题都是会专门讨论的（C10K problem: http://www.kegel.com/c10k.html）。要做开拓者，先行者，这些都是必然会遇到的。

我正在开发一款基于自研的LLM agent framework 的智能客服产品，它具有私有知识问答，意图引导、信息收集、情绪安抚、内部系统融合、LUI与GUI 融合、人工接管、数据分析与洞察、异常监控等功能。

欢迎对prompt编写、LLM应用开发与落地、智能客服产品等等感兴趣的朋友加我微信，一起交流，共同前行。

今天是2024年的除夕了，在这里顺祝大家新年快乐！2023年大环境不好，很多人不容易，但是要相信这都是暂时的，只要保持前行，总有希望！

# LLM 流式响应demo
import gradio as gr
import random
import time
import requests
import json
from extension.llm import llm_baidu
API_KEY = "your_ai_key"
SECRET_KEY = "your_secret_key"def baidu_llm_respond():url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions_pro?access_token=" + get_access_token()playload_obj = llm_baidu.LlmBaiduMsg(role="user", content="你是谁?").__dict__msgs = [playload_obj]payload = json.dumps({"messages": msgs,"temperature": 0.99,# "system": "this is system","disable_search": False,"enable_citation": False,"stream": True,})print(payload)headers = {'Content-Type': 'application/json'}response = requests.request("POST", url, headers=headers, data=payload, stream=True)obj_list = fetch_stream(response)for obj in obj_list:print(obj)print(f'stream result:{obj["result"]}')yield obj["result"]def fetch_stream(response):for line in response.iter_lines():if not line.startswith(b"data: "):continuejson_data = line[6:] # 只要jsondict_obj = json.loads(json_data)yield dict_objdef get_access_token():url = "https://aip.baidubce.com/oauth/2.0/token"params = {"grant_type": "client_credentials", "client_id": API_KEY, "client_secret": SECRET_KEY}return str(requests.post(url, params=params).json().get("access_token"))with gr.Blocks() as demo:chatbot = gr.Chatbot()msg = gr.Textbox()clear = gr.Button("Clear")def user_input_handler(user_message, history):print(f"user() user_message:{user_message}")return "", history + [[user_message, None]]def respond(history):print(f"bot() history={history}")bot_message_list = baidu_llm_respond()history[-1][1] = ""for msg in bot_message_list:for character in msg:history[-1][1] += charactertime.sleep(0.2)yield historymsg.submit(user_input_handler, [msg, chatbot], [msg, chatbot], queue=False).then(respond, chatbot, chatbot)clear.click(lambda: None, None, chatbot, queue=False)demo.queue()
demo.launch()