laravel 和 sdk 实现 ai 对话流式传输

php_yt 的个人博客 / 363 / 2 / 创建于 2个月前 / 更新于 1个月前

openai php8.1 openai-php/client laravel10
打字机输出的效果

php sdk

openai api 官方的包有 Python、 Node.js，其他语言的见社区维护的库，其中 openai-php/client 是较不错的 PHP 库，要求 guzzlehttp/guzzle 包以及 PHP 8.1+。

首要条件

已有 apikey，有国外服务器且搭建好了反向代理国外服务器反代 nginx 配置。

怎么应用到老项目中

如果 sdk 的代码没办法在老项目的 php 环境中运行，我是这么解决的，将域名的二级目录如 /ai/ 代理到新的 laravel 项目。

location ^~ /ai/ {
    proxy_pass http://127.0.0.1:8070/; # laravel 项目地址
    proxy_redirect off;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_buffering off; # 关闭缓冲区，这样才能流式返回
    proxy_cache off; # 关闭代理缓存
}

php 版本和 composer

openai-php/client 需 php8.1 以上。

centos 安装 php74 和 php82 及与 nginx 集成介绍了怎么独立安装 php82 搭建一个网站，且不影响目前已有项目的业务。

安装 php82 对应的 composer，参考 linux 安装 composer ⌈ 要用 php82 命令来安装 composer，把 composer 换个其他目录存放，比如 mv composer.phar /usr/local/composer/composer82 ⌋。

在之后的 laravel 项目中的命令，php 命令换成 php82， composer 命令换成 php82 /usr/local/composer/composer82，以安装 laravel10 示例：

php82 /usr/local/composer/composer82 create-project --prefer-dist laravel/laravel:^10.0 ai

流式客户端

要实现市面很多 ai 聊天产品打字机输出的效果，一般两种技术：EventSource（简称 SSE）和 websocket。

websocket 是较理想的，但实现较复杂。

sse 是单工，即服务器主动向客户端发送消息，与 websocket 一样，绝大多数浏览器都支持 sse。但原生 SSE 的缺点是只允许 GET 请求，不满足聊天场景中大文字的需求。几番搜索找到了 sse.js，支持 POST。

流式服务端

服务端要按照 SSE 的文档要求返回数据，并且要让 php 实时输出。

引入 openai-php/client

php82 /usr/local/composer/composer82 require openai-php/client

laravel 项目中创建一个 ai 的工具类。

<?php
namespace App\Lib;

use OpenAI;
use Symfony\Component\HttpFoundation\StreamedResponse;

class Ai
{
    /**
     * https://github.com/openai-php/client
     * ip:8050 反代服务器
     */
    public static function getClient()
    {
        $apiKey = config('app.openai_key');
        return OpenAI::factory()
          ->withApiKey($apiKey)
          ->withBaseUri('http://ip:8050/v1')
          ->withHttpHeader('OpenAI-Beta', 'assistants=v1')
          ->make();
    }

    /**
     * 流式响应。应用于即时chat
     * @param stream $client->chat()->createStreamed()
     * @param answeredCallback 回调函数，在输出后执行业务逻辑
     * @param completedCallback 聊天完成事件
     */
    public static function getStreamResponse($stream, $answeredCallback = null, $completedCallback = null)
    {
        $response = new StreamedResponse();
        $response->headers->set('Content-Type', 'text/event-stream');
        $response->headers->set('Cache-Control', 'no-cache');
        $response->headers->set('Connection', 'keep-alive');
        $response->headers->set('X-Accel-Buffering', 'no');

        $response->setCallback(function () use ($stream, $answeredCallback, $completedCallback) {

            $answer = ''; //回答

            try {
                foreach($stream as $index => $data) {
                    $choice = $data->choices[0]->toArray();

                    if ($index == 0) {
                        self::eventsourceBeginAnswer();
                    }

                    if ($choice['finish_reason'] == 'stop') {

                        if ( $completedCallback != null ) {
                            // 比如扣减使用次数
                            $completedCallback();
                        }

                        echo 'retry: 86400000' . PHP_EOL;
                        echo 'event: complete' . PHP_EOL;
                        echo 'data: chat complete, Connection closed' . PHP_EOL . PHP_EOL;
                    }
                    else {
                        $content = $choice['delta']['content'];
                        $item = json_encode(['content' => $content], JSON_UNESCAPED_UNICODE);
                        echo 'data: ' . $item . "\n\n";

                        $answer .= $content; //拼接回答
                    }

                    ob_flush();
                    flush();

                    if (connection_aborted()) {break;}// 如果客户端中止连接（关闭页面），则中断循环
                }
            } catch (\Exception $e) {}

            // 在这里将回答入库
            if ( $answeredCallback != null ) {
                $answeredCallback($answer);
            }
        });

        return $response;
    }

    // 终止链接事件，并返回错误信息
    public static function eventsourceClose($error)
    {
      header('Content-Type: text/event-stream');
      header('Cache-Control: no-cache');
      echo 'retry: 86400000' . PHP_EOL;
      echo 'event: close' . PHP_EOL;
      echo 'data: '. $error . PHP_EOL . PHP_EOL;
      die;
    }

    // 创建会话ID后事件
    public static function eventsourceSetConversationId($conversation_id)
    {
      header('Content-Type: text/event-stream');
      header('Cache-Control: no-cache');
      echo 'event: create_conversation' . PHP_EOL;
      echo 'data: '. json_encode(['conversation_id' => $conversation_id]) . PHP_EOL . PHP_EOL;
    }

    // 创建消息后事件
    public static function eventsourceCreateMsg($user_message_id, $assistant_message_id)
    {
      header('Content-Type: text/event-stream');
      header('Cache-Control: no-cache');
      echo 'event: create_msg' . PHP_EOL;
      echo 'data: '. json_encode([
            'user_message_id' => $user_message_id, //用户消息ID
            'assistant_message_id' => $assistant_message_id, //assistant消息ID
        ]) . PHP_EOL . PHP_EOL;
    }

    // 开始回答事件
    public static function eventsourceBeginAnswer()
    {
      echo 'event: begin_answer' . PHP_EOL;
      echo 'data: '. '' . PHP_EOL . PHP_EOL;
    }
}

控制器如何调用

use App\Lib\Ai;

if ( empty($question) ) {
 Ai::eventsourceClose('内容不能为空: question cant be empty!');
}
$client = Ai::getClient();
$stream = $client->chat()->createStreamed([
 'model'  => $config['model'],
 'messages'  => $messages,
 'temperature'  => $config['temperature'],
 'max_tokens' => $config['max_tokens']
]);
return  Ai::getStreamResponse($stream, $doneEvent, $completeEvent);

SSE 客户端

<script src="/ai/js/sse.js"></script>
<script>

function addSSEListener(sse) {
    sse.addEventListener('begin_answer', function (e) {
        console.log('begin_answer');
    });
    sse.addEventListener('message', function (e) {
        let eventData = JSON.parse(e.data);
        let content = eventData.content;
    });
    sse.addEventListener('complete', function (e) {
        console.log('complete: ', e.data);
        sse.close();
    });
    sse.addEventListener('close', function (e) {
        console.log('close: ', e.data);
        alert(e.data);
        sse.close();
    });
    sse.addEventListener('error', function (e) {
        console.log('error: ', e.data);
    });
}

let payload = { content: 'hello, can I ask a question?' };

let sse = new SSE('https://site/ai/v1/chatstream', {
    start: false,
    withCredentials: true,
    debug: false,
    headers: {
        'Content-Type': 'application/json',
        'Sse-Request': 'true'
    },
    payload: JSON.stringify(payload),
});

addSSEListener(sse);
sse.stream();

</script>