workerman 接入文心一言的流式输出
<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
require_once __DIR__ . '/vendor/autoload.php';
// 注意：这里与上个例子不同，使用的是websocket协议
$ws_worker = new Worker("websocket://0.0.0.0:2000");
// 启动4个进程对外提供服务
$ws_worker->count = 4;
// 当收到客户端发来的数据后返回hello $data给客户端
$ws_worker->onMessage = function(TcpConnection $connection, $data)
{
// 向客户端发送hello $data
$connection->send('hello ' . $data);
$token = getAccessToken();
$url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions_pro?access_token=".$token;
/*$map = '{
"messages": [
{
"role": "user",
"content": "给我推荐一些自驾游路线"
}
],
"stream":"True"
}';*/
//$maparr = ['messages'=>['role'=>'user','content'=>'hello'],'stream'=>true];
$maparr = ["messages" => [["role" => "user", "content" => 'hello']],"stream" => true];
curlStreamRequest(
$url, // 替换为实际的 API URL
['Content-Type: application/json'], // 替换为实际的请求头
json_encode($maparr), // 替换为实际的 POST 数据（如果需要）
'handleResponseData' // 传递回调函数名作为字符串（如果回调函数在全局作用域中）
// 或者直接传递闭包函数
// function($data) {
// handleResponseData($data);
// }
);
};
function getAccessToken(){
$curl = curl_init();
$postData1 = array(
'grant_type' => 'client_credentials',
'client_id' => '****', //接入文心一言key
'client_secret' =>'****' 接入文心一言的secret
);
curl_setopt_array($curl, array(
CURLOPT_URL => 'https://aip.baidubce.com/oauth/2.0/token',
CURLOPT_CUSTOMREQUEST => 'POST',
CURLOPT_SSL_VERIFYPEER => false,
CURLOPT_SSL_VERIFYHOST => false,
CURLOPT_RETURNTRANSFER => true,
CURLOPT_POSTFIELDS => http_build_query($postData1)
));
$response = curl_exec($curl);
curl_close($curl);
$rtn = json_decode($response);
return $rtn->access_token;
}
function curlStreamRequest(string $url, array $headers = [], $postData = null, callable $callback) {
echo $postData;
$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, $url);
curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, false); // 不将响应保存为字符串，直接处理
curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false); // 注意：在生产环境中应启用 SSL 验证
curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, false); // 注意：同上
curl_setopt($ch, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
curl_setopt($ch, CURLOPT_POST, is_array($postData) || !empty($postData));
curl_setopt($ch, CURLOPT_POSTFIELDS, $postData);
curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($ch, $data) use ($callback) {
// 调用回调函数处理数据
$callback($data);
return strlen($data); // 返回接收到的数据长度
});
// 执行请求并获取响应
curl_exec($ch);
// 检查是否有错误发生
if (curl_errno($ch)) {
throw new \Exception(curl_error($ch));
}
// 关闭 cURL 句柄
curl_close($ch);
}
/**
* 示例回调函数，用于处理接收到的数据并返回给客户端
*
* @param string $data 接收到的数据片段
*/
function handleResponseData($data) {
// 在这里，你可以将数据写入输出缓冲区或直接发送给客户端
// 例如，使用 echo 或 SSE 发送数据
echo $data; // 假设这里直接将数据发送给客户端
flush(); // 刷新输出缓冲区
}
// 使用示例
/*curlStreamRequest(
'https://example.com/api/stream', // 替换为实际的 API URL
['Content-Type: application/json'], // 替换为实际的请求头
json_encode(['key' => 'value']), // 替换为实际的 POST 数据（如果需要）
'handleResponseData' // 传递回调函数名作为字符串（如果回调函数在全局作用域中）
// 或者直接传递闭包函数
// function($data) {
// handleResponseData($data);
// }
);
*/
// 运行worker
Worker::runAll();