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

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

可以看到，从用户输入话题，到最终生成内容，经过了问题改写、简答、大纲、子问题拆解和文章生成等好几个步骤，中间还穿插并行生成封面和口语化内容，这些内容都需要通过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 主要有以下一些核心特性：

具体的使用方法，详见官方文档。

有任何问题，欢迎留言讨论。