【第3376期】Ling（灵）：追求极致响应速度的LLM工作流优化

前端早读课 · 公众号 · 前端 · 2024-09-14 08:00

正文

前言

主要介绍了一个名为 Ling（灵）的工作流框架，它针对 LLM（大型语言模型）生成的结构化 JSON 数据流实时处理问题，通过实时解析 JSON 数据并以 jsonuri 格式分发，使得前端和后端能够立即处理流式输出的数据，优化了复杂内容生成的工作流程。今日前端早读课文章由 @月影授权分享。

正文从这开始～～

从今年年初开始，我自己创业，做儿童教育方向的 AI 应用产品。技术上来讲，最核心的其实是复杂的内容生成工作流的编排及内容输出，这里面有非常复杂的工作流。

举个例子，比如我们的产品波波熊学伴的核心工作流，涉及到非常多的 Agent，大概如下图所示：

图 1

可以看到，从用户输入话题，到最终生成内容，经过了问题改写、简答、大纲、子问题拆解和文章生成等好几个步骤，中间还穿插并行生成封面和口语化内容，这些内容都需要通过 LLM 的工作流编排来实现。

现在，一些工作流框架如 Dify、Coze 都可以比较方便地编排工作流，我们自己也可以选择 Langchain 或者其他开源工具来实现工作流，但是，具体实现时会遇到一些坑。

比如，一般情况下，类似于波波熊学伴这类内容应用的工作流，都是结构化数据。这种数据，一般来说，使用 JSON 作为数据格式是最方便的，LLM 也支持的最好，不论 openai 还是国内的大模型如 moonshot 或者 deepseek，都提供了 json-mode 能够方便地让模型生成结构化的 JSON 数据。

然而，JSON 数据有一个问题，那就是它的结构是一个完整的闭合结构。简单来说，也就是，一个 JSON 从第一个 “{” 到最后一个 “}“，才成为一个完整的合法 JSON，这样的数据才能进行后续处理。这种结构，显然对流式输出不友好，因为一般情况下，不论在前端还是后端，我们都很难立即处理输出了一半的 JSON 数据。

解决方案

为了解决这个问题，我们设计了一个轻量级的工作流框架 Ling（灵），它专注于解决结构化数据（JSON 格式）的流式输出。

在这个框架的底层，我实现了一个实时解析 JSON 数据的 Parser，它可以一个字符一个字符地解析 JSON 数据，也就是说，可以一边接收 LLM 的流式输出，一边解析 JSON 数据，然后将解析完的部分，比如当前数
据字段中的文本新增字符，以 jsonuri 的格式分发出去。

例如，当前的 Agent 正在输出如下数据：

 {
   "outline": [
     {
       "topic": "云朵是由什么构成的？"
     },
     {
       "topic": "为什么云朵看起来软软的？"
     }
 ...

Parser 可以根据接收到的数据立即解析并分发：

 data: {"uri": "outline/0/topic", "delta": "云"}
 data: {"uri": "outline/0/topic", "delta": "朵"}
 data: {"uri": "outline/0/topic", "delta": "是"}
 data: {"uri": "outline/0/topic", "delta": "由"}
 data: {"uri": "outline/0/topic", "delta": "什"}
 data: {"uri": "outline/0/topic", "delta": "么"}
 data: {"uri": "outline/0/topic", "delta": "构"}
 data: {"uri": "outline/0/topic", "delta": "成"}
 data: {"uri": "outline/0/topic", "delta": "的"}
 data: {"uri": "outline/0/topic", "delta": "？"}
 data: {"uri": "outline/1/topic", "delta": "为"}
 data: {"uri": "outline/1/topic", "delta": "什"}
 data: {"uri": "outline/1/topic", "delta": "么"}
 data: {"uri": "outline/1/topic", "delta": "云"}
 data: {"uri": "outline/1/topic", "delta": "朵"}
 data: {"uri": "outline/1/topic", "delta": "看"}
 data: {"uri": "outline/1/topic", "delta": "起"}
 data: {"uri": "outline/1/topic", "delta": "来"}
 data: {"uri": "outline/1/topic", "delta": "软"}
 data: {"uri": "outline/1/topic", "delta": "软"}
 data: {"uri": "outline/1/topic", "delta": "的"}
 data: {"uri": "outline/1/topic", "delta": "？"}

这样我们在前端就可以立即处理：

 const es = new EventSource('http://localhost:3000/?question=Can I laid on the cloud?');

 es.onmessage = (e) => {
   console.log(e.data); // {"uri": "outline/0/topic", "delta": "云"}
 }

 es.onopen = () => {
   console.log('Connecting');
 }

 es.onerror = (e) => {
   console.log(e);
 }

同样，在服务端，一个字段解析完成后，比如 outline/0/topic，会触发 string-response 事件，我们就可以立即交给下一个 Agent 开始处理，不用等待整个 JSON 内容生成完。

以下是一个典型的服务端工作流的示例代码：

 function workflow(question: string, sse: boolean = false) {
   const config: ChatConfig = {
     model_name,
     api_key: apiKey,
     endpoint: endpoint,
   };

   const ling = new Ling(config);
   ling.setSSE(sse);

   // 工作流
   const bot = ling.createBot(/*'bearbobo'*/);
   bot.addPrompt('你用JSON格式回答我，以{开头\n[Example]\n{"answer": "我的回答"}');
   bot.chat(question);
   bot.on('string-response', ({uri, delta}) => {
     // JSON中的字符串内容推理完成，将 anwser 字段里的内容发给第二个 bot
     console.log('bot string-response', uri, delta);

     const bot2 = ling.createBot(/*'bearbobo'*/);
     bot2.addPrompt('将我给你的内容扩写成更详细的内容，用JSON格式回答我，将解答内容的详细文字放在\'details\'字段里，将2-3条相关的其他知识点放在\'related_question\'字段里。\n[Example]\n{"details": "我的详细回答", 
 "related_question": ["相关知识内容",...]}');
     bot2.chat(delta);
     bot2.on('response', (content) => {
       // 流数据推送完成
       console.log('bot2 response finished', content);
     });

     const bot3 = ling.createBot();
     bot3.addPrompt('将我给你的内容**用英文**扩写成更详细的内容，用JSON格式回答我，将解答内容的详细英文放在\'details_eng\'字段里。\n[Example]\n{"details_eng": "my answer..."}');
     bot3.chat(delta);
     bot3.on('response', (content) => {
       // 流数据推送完成
       console.log('bot3 response finished', content);
     });
   });

   ling.close(); // 可以直接关闭，关闭时会检查所有bot的状态是否都完成了

   return ling;
 }

上面这个工作流，我们用了三个 Agent，首先根据用户提问生成简答，内容放在 answer 字段里。接着，我们根据 answer 字段里的内容，分别调用不同的 Agent，同时扩写成更详细的中、英文内容。

可以看到，我们直接在 bot 的 string-response 事件里拿到 answer 字段的内容，此时 bot 的内容还没生成完，但是我们就已经可以用 answer 内容启动后面的两个 bot 了，这样就节约了等待时间。

除了上面说的通过流式格式化内容节省响应时间以外，Ling 主要有以下一些核心特性：

支持 JSONL 协议的数据流输出
JSON 的 TOKEN 异常的自动修复
支持复杂的异步工作流
支持流式输出过程中的状态消息
支持 Server Sent Events

具体的使用方法，详见官方文档：https://ling.bearbobo.com/

ling：https://github.com/WeHomeBot/ling

关于本文
作者：@十年踪迹
原文：https://juejin.cn/post/7413228844403294249

这期前端早读课
对你有帮助，帮” 赞 “一下，
期待下一期，帮” 在看” 一下。