[RFC] 050 - 文心一言 Provider 接入

[arvinxx]

背景

文心一言作为呼声非常高的 Model Provider ，我们是需要支持的。但是百度的 SDK 设计的复杂度挺高，因此没法很轻易地接入。但通过之前 #1310 的实现改造，现在已经具备了处理各异流式请求的 provider 的能力。因此计划在本篇 RFC 中记录相应的实现思路。

同时顺带改造优化前置 UI 环节的处理流程，简化一个 provider 的接入实现。

设计思路

Provider 接入初始化

核心是实现一个 LobeRuntimeAI 类的初始化和 chat 方法。

初始化：

import { ChatCompletion } from '@baiducloud/qianfan';


export class LobeWenxinAI implements LobeRuntimeAI {

constructor({
    accessKey,
    baseURL,
    secretKey,
  }: {
    accessKey?: string;
    baseURL?: string;
    secretKey?: string;
  }) {
   // 1. 确保 key 存在
    if (!accessKey || secretKey)
      throw AgentRuntimeError.createError(AgentRuntimeErrorType.InvalidProviderAPIKey);

   // 2. 初始化实例
    this.client = new ChatCompletion({
      QIANFAN_ACCESS_KEY: accessKey,
      QIANFAN_SECRET_KEY: secretKey,
    });
    this.baseURL = baseURL;
  }

chat 接口调用：

  async chat(payload: ChatStreamPayload, options?: ChatCompetitionOptions) {
    const stream: AsyncIterable<ChatResp> = await this.client.chat(
      { messages: payload.messages, stream: true },
      payload.model,
    );
  }

这里其实就是百度的 sdk 的实现和 openai 等比起来非常不友好的地方，像 AsyncIterable 这种东西 90% 的前端估计都没接触过。而接入复杂度也在这里。

wenxin Stream 处理

在我们的流处理框架中，核心思路就两条：

1. 将所有 流/AsyncIterable 都转换为 ReadableStream；
2. 利用流的 pipeThrough 能力，将流按需转换成目标协议的内容，并添加回调；

在 Wenxin 接入的过程中，Stream 处理流程如下：

1. 将 AsyncIterable 转为 ReadableStream。 利用 ai 方法中的 readableFromAsyncIterable 和自行实现的 chatStreamable ，将流转换为 ReadableStream

import { readableFromAsyncIterable } from 'ai';

// 将原始迭代器转换为另外一个迭代器
// 要这么做的原因是，原始迭代器的每个 chunk 可能是二进制，也可能是 json 对象
// 通过这一层以后，都会统一成 json 对象，便于后续
const chatStreamable = async function* <T>(stream: AsyncIterable<T>) {
  for await (const response of stream) {
    yield response;
  }
};

// 将原始流转为 ReadableStream
export const WenxinResultToStream = (stream: AsyncIterable<ChatResp>) => {
  // make the response to the streamable format
  return readableFromAsyncIterable(chatStreamable(stream));
};

2. 转换每个 chunk 的内容： ReadableStream 可以通过它的 pipeThrough 中传入一个 TransformStream ，来实现每个 chunk 的数据转换。

// 将原始流转成最终的消费流
// rawStream 消费 `WenxinResultToStream` 后得到的结果
const WenxinStream = (
  rawStream: ReadableStream<ChatResp>,
  callbacks?: ChatStreamCallbacks,
) => {
  const streamStack: StreamStack = { id: 'chat_' + nanoid() };

  return rawStream
    // 此处传入一个 TransformStream 对象，针对每个 stream 过来的 chunk，在这里进行处理，转成我们定义的流式协议
    .pipeThrough(createSSEProtocolTransformer(transformWenxinStream, streamStack))
    // 为这个流添加回调能力
    .pipeThrough(createCallbacksTransformer(callbacks));
};

createSSEProtocolTransformer 核心就是一个转换方法，用于创建我们定义的 SSE 的传输协议

const createSSEProtocolTransformer = (
  transformer: (chunk: any, stack: StreamStack) => StreamProtocolChunk,
  streamStack?: StreamStack,
) =>
  new TransformStream({
    transform: (chunk, controller) => {
      const { type, id, data } = transformer(chunk, streamStack || { id: '' });

      controller.enqueue(`id: ${id}\n`);
      controller.enqueue(`event: ${type}\n`);
      controller.enqueue(`data: ${JSON.stringify(data)}\n\n`);
    },
  });

具体转换实现

首先看下 wenxin 的接口返回数据结构：

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089502,"sentence_id":0,"is_end":false,"is_truncated":false,"result":"当然可以，","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089504,"sentence_id":1,"is_end":false,"is_truncated":false,"result":"以下是一些建议的自驾游路线，它们涵盖了各种不同的风景和文化体验：\n\n1. **西安-敦煌历史文化之旅**：\n\n\n\t* 路线：西安","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089506,"sentence_id":2,"is_end":false,"is_truncated":false,"result":" - 天水 - 兰州 - 嘉峪关 - 敦煌\n\t* 特点：此路线让您领略到中国西北的丰富历史文化。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089508,"sentence_id":3,"is_end":false,"is_truncated":false,"result":"您可以参观西安的兵马俑、大雁塔，体验兰州的黄河风情，以及在敦煌欣赏壮丽的莫高窟。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089511,"sentence_id":4,"is_end":false,"is_truncated":false,"result":"\n2. **海南环岛热带风情游**：\n\n\n\t* 路线：海口 - 三亚 - 陵水 - 万宁 - 文昌 - 海","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089512,"sentence_id":5,"is_end":false,"is_truncated":false,"result":"口\n\t* 特点：海南岛是中国唯一的黎族聚居区，这里有独特的热带风情、美丽的海滩和丰富的水果。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":153,"total_tokens":158}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089513,"sentence_id":6,"is_end":false,"is_truncated":false,"result":"您可以在三亚享受阳光沙滩，品尝当地美食，感受海南的悠闲生活。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":153,"total_tokens":158}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089516,"sentence_id":7,"is_end":false,"is_truncated":false,"result":"\n3. **穿越阿里大北线**：\n\n\n\t* 路线：成都 - 广元 - 汉中 - 西安 - 延安 - 银川 -","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":153,"total_tokens":158}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089518,"sentence_id":8,"is_end":false,"is_truncated":false,"result":" 阿拉善左旗 - 额济纳旗 - 嘉峪关 - 敦煌\n\t* 特点：这是一条充满挑战的自驾路线，穿越了中国","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":153,"total_tokens":158}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089519,"sentence_id":9,"is_end":false,"is_truncated":false,"result":"的西部。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089519,"sentence_id":10,"is_end":false,"is_truncated":false,"result":"您将经过壮观的沙漠、神秘的戈壁和古老的丝绸之路遗址。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089520,"sentence_id":11,"is_end":false,"is_truncated":false,"result":"此路线适合喜欢探险和寻求不同体验的旅行者。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089523,"sentence_id":12,"is_end":false,"is_truncated":false,"result":"\n4. **寻找北方净土 - 阿尔山自驾之旅**：\n\n\n\t* 路线：北京 - 张家口 - 张北 - 太仆寺旗","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089525,"sentence_id":13,"is_end":false,"is_truncated":false,"result":" - 锡林浩特 - 东乌珠穆沁旗 - 满都湖宝拉格 - 宝格达林场 - 五岔沟 - 阿尔山 -","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089527,"sentence_id":14,"is_end":false,"is_truncated":false,"result":" 伊尔施 - 新巴尔虎右旗 - 满洲里 - 北京\n\t* 特点：此路线带您穿越中国北方的草原和森林，抵达","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089527,"sentence_id":15,"is_end":false,"is_truncated":false,"result":"风景如画的阿尔山。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089528,"sentence_id":16,"is_end":false,"is_truncated":false,"result":"您可以在这里欣赏壮丽的自然风光，体验当地的民俗文化，享受宁静的乡村生活。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089529,"sentence_id":17,"is_end":false,"is_truncated":false,"result":"\n\n以上路线仅供参考，您可以根据自己的兴趣和时间安排进行调整。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089531,"sentence_id":18,"is_end":false,"is_truncated":false,"result":"在规划自驾游时，请务必注意道路安全、车辆保养以及当地的天气和交通状况。","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089531,"sentence_id":19,"is_end":false,"is_truncated":false,"result":"祝您旅途愉快！","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":239,"total_tokens":244}}

data: {"id":"as-vb0m37ti8y","object":"chat.completion","created":1709089531,"sentence_id":20,"is_end":true,"is_truncated":false,"result":"","need_clear_history":false,"finish_reason":"normal","usage":{"prompt_tokens":5,"completion_tokens":420,"total_tokens":425}}

核心就是提取 result 字段，因此实现就很简单：

const transformERNIEBotStream = (chunk: ChatResp): StreamProtocolChunk => {
  const finished = chunk.is_end;
  // 判断下结束，返回 stop 标记
  if (finished) {
    return { data: chunk.finish_reason || 'stop', id: chunk.id, type: 'stop' };
  }

  // 有 result ，返回 result 结果，类型为 text
  if (chunk.result) {
    return { data: chunk.result, id: chunk.id, type: 'text' };
  }
};

以上就完成了 wenxin 的流式协议转换

chat 方法中接入

有了上述方法以后，在chat层实现就很简单了，

 async chat(payload: ChatStreamPayload, options?: ChatCompetitionOptions) {
    const result: AsyncIterable<ChatResp> = await this.client.chat(
      { messages: payload.messages, stream: true },
      payload.model,
    );

    // 1. 获取 ReadableStream 实例
    const wenxinStream = WenxinResultToStream(result);

    // 2. 利用 tee 实现分流，这样消费 prod 和 debug 流就不会互相影响
    const [prod, useForDebug] = wenxinStream.tee();

    // 3. 加一个 debug 标记，这样在终端就能看到原始流的信息
    if (process.env.DEBUG_WENXIN_CHAT_COMPLETION === '1') {
      debugStream(useForDebug).catch();
    }

    // 4. 将原始的 ReadableStream 转为我们处理过的流
    const stream = WenxinStream(prod, options?.callback);

    // 5. 将流以 Response 的方式进行返回
    return StreamingResponse(stream, { headers: options?.headers });
  }

补齐测试

主要做下协议转换的测试：

import { describe, expect, it, vi } from 'vitest';

import * as uuidModule from '@/utils/uuid';

import { ChatResp } from '../../wenxin/type';
import { WenxinResultToStream, WenxinStream } from './wenxin';

const dataStream = [
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089502,
    sentence_id: 0,
    is_end: false,
    is_truncated: false,
    result: '当然可以，',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
  },
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089504,
    sentence_id: 1,
    is_end: false,
    is_truncated: false,
    result:
      '以下是一些建议的自驾游路线，它们涵盖了各种不同的风景和文化体验：\n\n1. **西安-敦煌历史文化之旅**：\n\n\n\t* 路线：西安',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
  },
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089506,
    sentence_id: 2,
    is_end: false,
    is_truncated: false,
    result: ' - 天水 - 兰州 - 嘉峪关 - 敦煌\n\t* 特点：此路线让您领略到中国西北的丰富历史文化。',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
  },
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089508,
    sentence_id: 3,
    is_end: false,
    is_truncated: false,
    result: '您可以参观西安的兵马俑、大雁塔，体验兰州的黄河风情，以及在敦煌欣赏壮丽的莫高窟。',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
  },
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089511,
    sentence_id: 4,
    is_end: false,
    is_truncated: false,
    result: '\n2. **海南环岛热带风情游**：\n\n\n\t* 路线：海口 - 三亚 - 陵水 - 万宁 - 文昌 - 海',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
  },
  {
    id: 'as-vb0m37ti8y',
    object: 'chat.completion',
    created: 1709089512,
    sentence_id: 5,
    is_end: false,
    is_truncated: false,
    result:
      '口\n\t* 特点：海南岛是中国唯一的黎族聚居区，这里有独特的热带风情、美丽的海滩和丰富的水果。',
    need_clear_history: false,
    finish_reason: 'normal',
    usage: { prompt_tokens: 5, completion_tokens: 153, total_tokens: 158 },
  },
];

describe('WenxinStream', () => {
  it('should transform Wenxin stream to protocol stream', async () => {
    vi.spyOn(uuidModule, 'nanoid').mockReturnValueOnce('1');

    const mockWenxinStream: AsyncIterable<ChatResp> = {
      // @ts-ignore
      async *[Symbol.asyncIterator]() {
        for (const item of dataStream) {
          yield item;
        }
      },
    };

    const stream = WenxinResultToStream(mockWenxinStream);

    const onStartMock = vi.fn();
    const onTextMock = vi.fn();
    const onTokenMock = vi.fn();
    const onCompletionMock = vi.fn();

    const protocolStream = WenxinStream(stream, {
      onStart: onStartMock,
      onText: onTextMock,
      onToken: onTokenMock,
      onCompletion: onCompletionMock,
    });

    const decoder = new TextDecoder();
    const chunks = [];

    // @ts-ignore
    for await (const chunk of protocolStream) {
      chunks.push(decoder.decode(chunk, { stream: true }));
    }

    expect(chunks).toEqual(
      [
        'id: chat_1',
        'event: text',
        `data: "当然可以，"\n`,
        'id: chat_1',
        'event: text',
        `data: "以下是一些建议的自驾游路线，它们涵盖了各种不同的风景和文化体验：\\n\\n1. **西安-敦煌历史文化之旅**：\\n\\n\\n\\t* 路线：西安"\n`,
        'id: chat_1',
        'event: text',
        `data: " - 天水 - 兰州 - 嘉峪关 - 敦煌\\n\\t* 特点：此路线让您领略到中国西北的丰富历史文化。"\n`,
        'id: chat_1',
        'event: text',
        `data: "您可以参观西安的兵马俑、大雁塔，体验兰州的黄河风情，以及在敦煌欣赏壮丽的莫高窟。"\n`,
        'id: chat_1',
        'event: text',
        `data: "\\n2. **海南环岛热带风情游**：\\n\\n\\n\\t* 路线：海口 - 三亚 - 陵水 - 万宁 - 文昌 - 海"\n`,
        'id: chat_1',
        'event: text',
        `data: "口\\n\\t* 特点：海南岛是中国唯一的黎族聚居区，这里有独特的热带风情、美丽的海滩和丰富的水果。"\n`,
      ].map((item) => `${item}\n`),
    );

    expect(onStartMock).toHaveBeenCalledTimes(1);
    expect(onTextMock).toHaveBeenNthCalledWith(1, '"当然可以，"');
    expect(onTextMock).toHaveBeenNthCalledWith(2, '"以下是一些建议的自驾游路线，它们涵盖了各种不同的风景和文化体验：\\n\\n1. **西安-敦煌历史文化之旅**：\\n\\n\\n\\t* 路线：西安"');
    expect(onTokenMock).toHaveBeenCalledTimes(6);
    expect(onCompletionMock).toHaveBeenCalledTimes(1);
  });
});

进展

• ✨ feat: add wenxin model provider #4018