{T}

Ling 工作流框架

Ling 是一个基于流式 JSON 数据的异步工作流框架，通过 Adapter（模型适配）、Bot（推理节点）、Tube（数据管道）三大模块，实现多 AI 节点间的高效数据流转和实时分发。

设计思想

核心问题

在复杂 AI 应用中，多个大模型节点需要协同工作：

  • BotA 生成大纲 → BotB/C 并行展开各章节
  • 某字段输出完成 → 立即触发语音合成（不等待整体完成）
  • 所有节点的输出 → 通过单一 Stream 统一分发给客户端

架构概览

图表渲染中…

三大模块职责

模块职责核心能力
Adapter对接不同大模型 API统一接口、流式处理、错误重试
Bot工作流中的推理节点Prompt 管理、状态机、事件发射
Tube节点间数据管道数据路由、字段监听、并行分发

Adapter 模块

设计目标

屏蔽不同大模型 API 的差异，提供统一的调用接口：

typescript
// adapter/openai.ts
import type { ChatConfig, ChatOptions } from '../types';

export async function getChatCompletions(
  config: ChatConfig,
  options: ChatOptions = {}
): Promise<ReadableStream> {
  const response = await fetch(config.endpoint, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${config.apiKey}`,
    },
    body: JSON.stringify({
      model: config.model,
      messages: config.messages,
      stream: true,
      ...options,
    }),
  });

  if (!response.ok) {
    throw new Error(`API Error: ${response.status}`);
  }

  return response.body!;
}
typescript
// adapter/coze.ts
export async function getChatCompletions(
  config: ChatConfig,
  options: ChatOptions = {}
): Promise<ReadableStream> {
  const response = await fetch('https://api.coze.cn/v3/chat', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${config.apiKey}`,
    },
    body: JSON.stringify({
      bot_id: config.botId,
      user_id: config.userId,
      stream: true,
      additional_messages: config.messages,
    }),
  });

  return response.body!;
}

Bot 模块

状态机设计

每个 Bot 节点是一个状态机，管理推理生命周期：

typescript
enum WorkState {
  INIT = 'init',           // 初始化
  WORKING = 'chatting',    // 推理中
  INFERENCE_DONE = 'inference-done',  // 推理完成
  FINISHED = 'finished',   // 全部完成
  ERROR = 'error',         // 错误
}
图表渲染中…

ChatBot 实现

typescript
import EventEmitter from 'node:events';
import { Tube } from '../tube';
import nunjucks from 'nunjucks';
import { getChatCompletions } from '../adapter/openai';
import type { ChatConfig, ChatOptions } from '../types';

export class ChatBot extends EventEmitter {
  private prompts: Array<{ role: 'system'; content: string }> = [];
  private history: Array<{ role: string; content: string }> = [];
  private chatState = WorkState.INIT;
  private config: ChatConfig;
  private options: ChatOptions;

  constructor(
    private tube: Tube,
    config: ChatConfig,
    options: ChatOptions = {}
  ) {
    super();
    this.config = { ...config };
    this.options = { ...options };
  }

  get state(): WorkState {
    return this.chatState;
  }

  // 设置系统提示词（支持模板变量）
  setPrompt(template: string, data?: Record<string, any>) {
    const content = data
      ? nunjucks.renderString(template, data)
      : template;
    this.prompts = [{ role: 'system', content }];
    return this;
  }

  // 添加用户消息
  addUserMessage(content: string) {
    this.history.push({ role: 'user', content });
    return this;
  }

  // 启动推理
  async start() {
    this.chatState = WorkState.WORKING;
    this.emit('state-change', this.chatState);

    const messages = [...this.prompts, ...this.history];

    try {
      const stream = await getChatCompletions(
        { ...this.config, messages },
        this.options
      );

      // 将流数据送入 Tube 进行解析和分发
      await this.tube.process(stream);

      this.chatState = WorkState.INFERENCE_DONE;
      this.emit('state-change', this.chatState);
    } catch (error) {
      this.chatState = WorkState.ERROR;
      this.emit('error', error);
    }
  }
}

Tube 模块

数据管道职责

Tube 是连接 Bot 和输出流的管道，核心职责：

  1. 消费 Adapter 返回的原始流
  2. 通过 JSONParser 动态解析 JSON
  3. 在字段完成时触发 string-resolve 事件
  4. 将增量数据路由到正确的输出位置
typescript
import { JSONParser } from '../parser';
import type { WritableStream } from 'stream/web';

export class Tube {
  private parser: JSONParser;
  private output: WritableStream;

  constructor(output: WritableStream) {
    this.parser = new JSONParser();
    this.output = output;
    this.setupEvents();
  }

  private setupEvents() {
    // 增量数据 → 转发给客户端
    this.parser.on('data', ({ uri, delta }) => {
      this.writeToOutput({ type: 'data', uri, delta });
    });

    // 字段完成 → 通知订阅者
    this.parser.on('string-resolve', ({ uri, value }) => {
      this.emit('field-complete', { uri, value });
    });
  }

  async process(stream: ReadableStream) {
    const reader = stream.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop() || '';

      for (const line of lines) {
        if (!line.startsWith('data: ')) continue;
        const data = line.slice(6).trim();
        if (data === '[DONE]') continue;

        try {
          const parsed = JSON.parse(data);
          const content = parsed.choices?.[0]?.delta?.content;
          if (content) this.parser.feed(content);
        } catch { /* skip incomplete */ }
      }
    }

    this.parser.end();
  }

  private writeToOutput(data: any) {
    const writer = this.output.getWriter();
    writer.write(`data: ${JSON.stringify(data)}\n\n`);
    writer.releaseLock();
  }
}

工作流编排实战

两级工作流：大纲 + 展开

图表渲染中…

服务端编排代码

typescript
import { ChatBot } from '@bearbobo/ling';
import { Tube } from '@bearbobo/ling';

app.post('/api/generate', async (req, res) => {
  const { question } = req.body;

  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');

  const tube = new Tube(res);

  // 第一级：生成大纲
  const outlineBot = new ChatBot(tube, {
    endpoint: process.env.LLM_ENDPOINT,
    apiKey: process.env.LLM_API_KEY,
    model: 'deepseek-chat',
  });

  outlineBot.setPrompt(`
    根据用户问题生成一篇儿童科普文章的大纲。
    输出 JSON 格式：
    {
      "title": "文章标题",
      "sections": [
        {"title": "章节标题", "keywords": ["关键词"]}
      ]
    }
  `);
  outlineBot.addUserMessage(question);

  // 监听大纲字段完成，触发第二级
  tube.on('field-complete', async ({ uri, value }) => {
    if (uri.match(/\/sections\/\d+\/title/)) {
      // 某个章节标题完成 → 启动展开 Bot
      const expandBot = new ChatBot(tube, { /* config */ });
      expandBot.setPrompt(`
        为儿童撰写关于"${value}"的科普段落，200字左右，语言生动有趣。
      `);
      expandBot.addUserMessage(`主题：${value}\n原始问题：${question}`);
      await expandBot.start();
    }
  });

  await outlineBot.start();
  res.end();
});

客户端集成

typescript
import { set } from 'jsonuri';

// 安装: pnpm i @bearbobo/ling jsonuri

const eventSource = new EventSource('/api/generate');
const state = reactive({});

eventSource.onmessage = (event) => {
  const { type, uri, delta } = JSON.parse(event.data);
  if (type === 'data') {
    const current = get(state, uri) || '';
    set(state, uri, current + delta);
  }
};

常见问题与陷阱

问题原因解决方案
并行 Bot 输出交错多个 Bot 共用一个 Tube使用 uri 前缀区分不同 Bot 的输出
内存泄漏事件监听器未清理工作流结束后 removeAllListeners
模板渲染错误nunjucks 变量未定义提供默认值 {% if var %}
流提前关闭客户端断开连接监听 close 事件，中止推理

参考资源