← All changes
|
app/Services/Integrations/FluentBot/FluentBotAPI.php
+156
-62
2.2.1
→
2.4.5
View file →
| @@ -7,11 +7,14 @@ | ||
| 7 | 7 | class FluentBotAPI |
| 8 | 8 | { |
| 9 | 9 | protected $apiUrl; |
| 10 | 10 | |
| 11 | - public function __construct(string $apiUrl) | |
| 11 | + protected $apiKey; | |
| 12 | + | |
| 13 | + public function __construct(string $apiUrl, string $apiKey = '') | |
| 12 | 14 | { |
| 13 | 15 | $this->apiUrl = $apiUrl; |
| 16 | + $this->apiKey = $apiKey; | |
| 14 | 17 | } |
| 15 | 18 | |
| 16 | 19 | public function makeRequest(int $ticketId, $prompt, array $args = []) |
| 17 | 20 | { |
| @@ -51,9 +54,8 @@ | ||
| 51 | 54 | $tokenUsage = $responseBody['token_usage'] ?? []; |
| 52 | 55 | $totalTokens = ($tokenUsage['input_tokens'] ?? 0) + ($tokenUsage['output_tokens'] ?? 0); |
| 53 | 56 | do_action('fluent_support/ai_response_success', $ticketId, $prompt, $totalTokens, "FluentBot"); |
| 54 | 57 | |
| 55 | - // Return both content and chat_id if available | |
| 56 | 58 | return [ |
| 57 | 59 | 'content' => $content, |
| 58 | 60 | 'chat_id' => $responseBody['chat_id'] ?? null |
| 59 | 61 | ]; |
| @@ -62,10 +64,9 @@ | ||
| 62 | 64 | public function makeStreamRequest(int $ticketId, $prompt, array $args = []) |
| 63 | 65 | { |
| 64 | 66 | $timeout = apply_filters('fs_ai_request_timeout', 120); |
| 65 | 67 | |
| 66 | - // Use cURL for streaming | |
| 67 | - // Note: Using cURL here because WordPress HTTP API doesn't support streaming SSE responses | |
| 68 | + // Raw cURL because the WP HTTP API can't stream SSE. | |
| 68 | 69 | // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init |
| 69 | 70 | // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt |
| 70 | 71 | // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec |
| 71 | 72 | // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close |
| @@ -75,11 +76,13 @@ | ||
| 75 | 76 | $ch = curl_init(); |
| 76 | 77 | curl_setopt($ch, CURLOPT_URL, $this->apiUrl); |
| 77 | 78 | curl_setopt($ch, CURLOPT_POST, true); |
| 78 | 79 | curl_setopt($ch, CURLOPT_POSTFIELDS, wp_json_encode($args)); |
| 79 | - curl_setopt($ch, CURLOPT_HTTPHEADER, [ | |
| 80 | - 'Content-Type: application/json', | |
| 81 | - ]); | |
| 80 | + $streamHeaders = ['Content-Type: application/json']; | |
| 81 | + if ($this->apiKey !== '') { | |
| 82 | + $streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey; | |
| 83 | + } | |
| 84 | + curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders); | |
| 82 | 85 | |
| 83 | 86 | $buffer = ''; |
| 84 | 87 | $conversationId = null; |
| 85 | 88 | $streamTokens = 0; |
| @@ -84,64 +87,16 @@ | ||
| 84 | 87 | $conversationId = null; |
| 85 | 88 | $streamTokens = 0; |
| 86 | 89 | |
| 87 | 90 | curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer, &$conversationId, &$streamTokens) { |
| 88 | - $buffer .= $data; | |
| 89 | - | |
| 90 | - // Process complete SSE events from the AI API | |
| 91 | - $events = explode("\n\n", $buffer); | |
| 92 | - $buffer = array_pop($events); // Keep incomplete event in buffer | |
| 93 | - | |
| 94 | - foreach ($events as $event) { | |
| 95 | - if (trim($event)) { | |
| 96 | - $lines = explode("\n", $event); | |
| 97 | - $eventType = ''; | |
| 98 | - $eventId = ''; | |
| 99 | - $eventDataLines = []; | |
| 100 | - | |
| 101 | - foreach ($lines as $line) { | |
| 102 | - if (strpos($line, 'event: ') === 0) { | |
| 103 | - $eventType = trim((string)substr($line, 7)); | |
| 104 | - } elseif (strpos($line, 'id: ') === 0) { | |
| 105 | - $eventId = trim((string)substr($line, 4)); | |
| 106 | - } elseif (strpos($line, 'data: ') === 0) { | |
| 107 | - $eventDataLines[] = (string)substr($line, 6); | |
| 108 | - } | |
| 91 | + foreach ($this->drainSseChunk($data, $buffer) as $parsed) { | |
| 92 | + if ($parsed['event'] === 'chat_id' && !empty($parsed['data'])) { | |
| 93 | + $conversationId = $parsed['data'][0]; | |
| 94 | + } elseif ($parsed['event'] === 'token_usage' && !empty($parsed['data'])) { | |
| 95 | + $usage = json_decode($parsed['data'][0], true); | |
| 96 | + if (is_array($usage)) { | |
| 97 | + $streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0); | |
| 109 | 98 | } |
| 110 | - | |
| 111 | - // Forward the event to the browser with proper formatting | |
| 112 | - if ($eventType) { | |
| 113 | - echo "event: ".esc_html($eventType)."\n"; | |
| 114 | - | |
| 115 | - // Include ID if present | |
| 116 | - if ($eventId !== '') { | |
| 117 | - echo "id: ".esc_html($eventId)."\n"; | |
| 118 | - } | |
| 119 | - | |
| 120 | - // Handle multiple data lines properly | |
| 121 | - if (!empty($eventDataLines)) { | |
| 122 | - foreach ($eventDataLines as $dataLine) { | |
| 123 | - // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side | |
| 124 | - echo "data: ".$dataLine."\n"; | |
| 125 | - } | |
| 126 | - } else { | |
| 127 | - echo "data: \n"; | |
| 128 | - } | |
| 129 | - | |
| 130 | - echo "\n"; | |
| 131 | - | |
| 132 | - // Store chat_id for later use | |
| 133 | - if ($eventType === 'chat_id' && !empty($eventDataLines)) { | |
| 134 | - $conversationId = $eventDataLines[0]; | |
| 135 | - } elseif ($eventType === 'token_usage' && !empty($eventDataLines)) { | |
| 136 | - $usage = json_decode($eventDataLines[0], true); | |
| 137 | - if (is_array($usage)) { | |
| 138 | - $streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0); | |
| 139 | - } | |
| 140 | - } | |
| 141 | - | |
| 142 | - flush(); | |
| 143 | - } | |
| 144 | 99 | } |
| 145 | 100 | } |
| 146 | 101 | |
| 147 | 102 | return strlen($data); |
| @@ -173,13 +128,152 @@ | ||
| 173 | 128 | do_action('fluent_support/ai_response_success', $ticketId, $prompt, $streamTokens, "FluentBot"); |
| 174 | 129 | } |
| 175 | 130 | } |
| 176 | 131 | |
| 132 | + /** | |
| 133 | + * Forward every complete SSE frame in $data to the browser, return the parsed frames. | |
| 134 | + * $buffer holds the partial trailing frame across cURL writes, so pass it by reference. | |
| 135 | + * | |
| 136 | + * @param string $data raw bytes from cURL | |
| 137 | + * @param string $buffer incomplete frame carried over from the previous write | |
| 138 | + * @return array<int, array{event: string, data: array<int, string>}> | |
| 139 | + */ | |
| 140 | + private function drainSseChunk(string $data, string &$buffer): array | |
| 141 | + { | |
| 142 | + $buffer .= $data; | |
| 143 | + | |
| 144 | + $events = explode("\n\n", $buffer); | |
| 145 | + $buffer = array_pop($events); // Keep incomplete event in buffer | |
| 146 | + | |
| 147 | + $parsed = []; | |
| 148 | + | |
| 149 | + foreach ($events as $event) { | |
| 150 | + if (!trim($event)) { | |
| 151 | + continue; | |
| 152 | + } | |
| 153 | + | |
| 154 | + // Forward keepalive comments as-is so proxies don't hit an idle timeout. | |
| 155 | + if (strpos(ltrim($event), ':') === 0) { | |
| 156 | + echo $event . "\n\n"; | |
| 157 | + flush(); | |
| 158 | + continue; | |
| 159 | + } | |
| 160 | + | |
| 161 | + $lines = explode("\n", $event); | |
| 162 | + $eventType = ''; | |
| 163 | + $eventId = ''; | |
| 164 | + $eventDataLines = []; | |
| 165 | + | |
| 166 | + foreach ($lines as $line) { | |
| 167 | + if (strpos($line, 'event: ') === 0) { | |
| 168 | + $eventType = trim((string)substr($line, 7)); | |
| 169 | + } elseif (strpos($line, 'id: ') === 0) { | |
| 170 | + $eventId = trim((string)substr($line, 4)); | |
| 171 | + } elseif (strpos($line, 'data: ') === 0) { | |
| 172 | + $eventDataLines[] = (string)substr($line, 6); | |
| 173 | + } | |
| 174 | + } | |
| 175 | + | |
| 176 | + if (!$eventType) { | |
| 177 | + continue; | |
| 178 | + } | |
| 179 | + | |
| 180 | + echo "event: ".esc_html($eventType)."\n"; | |
| 181 | + | |
| 182 | + if ($eventId !== '') { | |
| 183 | + echo "id: ".esc_html($eventId)."\n"; | |
| 184 | + } | |
| 185 | + | |
| 186 | + if (!empty($eventDataLines)) { | |
| 187 | + foreach ($eventDataLines as $dataLine) { | |
| 188 | + // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side | |
| 189 | + echo "data: ".$dataLine."\n"; | |
| 190 | + } | |
| 191 | + } else { | |
| 192 | + echo "data: \n"; | |
| 193 | + } | |
| 194 | + | |
| 195 | + echo "\n"; | |
| 196 | + flush(); | |
| 197 | + | |
| 198 | + $parsed[] = ['event' => $eventType, 'data' => $eventDataLines]; | |
| 199 | + } | |
| 200 | + | |
| 201 | + return $parsed; | |
| 202 | + } | |
| 203 | + | |
| 204 | + /** | |
| 205 | + * Reconnect to an in-flight turn: replay its buffered SSE and tail it live. | |
| 206 | + * Same wire format as makeStreamRequest. Upstream ends on `__done__` or an `idle` frame. | |
| 207 | + */ | |
| 208 | + public function makeResumeStreamRequest() | |
| 209 | + { | |
| 210 | + // Raw cURL because the WP HTTP API can't stream SSE. | |
| 211 | + // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init | |
| 212 | + // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt | |
| 213 | + // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec | |
| 214 | + // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close | |
| 215 | + // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error | |
| 216 | + // PluginCheck:ignoreFile | |
| 217 | + $ch = curl_init(); | |
| 218 | + curl_setopt($ch, CURLOPT_URL, $this->apiUrl); | |
| 219 | + curl_setopt($ch, CURLOPT_HTTPGET, true); | |
| 220 | + | |
| 221 | + $streamHeaders = ['Accept: text/event-stream']; | |
| 222 | + if ($this->apiKey !== '') { | |
| 223 | + $streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey; | |
| 224 | + } | |
| 225 | + curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders); | |
| 226 | + | |
| 227 | + $buffer = ''; | |
| 228 | + | |
| 229 | + curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer) { | |
| 230 | + // Stop when the agent leaves. The turn itself is persisted upstream. | |
| 231 | + if (connection_aborted()) { | |
| 232 | + return 0; | |
| 233 | + } | |
| 234 | + | |
| 235 | + $this->drainSseChunk($data, $buffer); | |
| 236 | + | |
| 237 | + return strlen($data); | |
| 238 | + }); | |
| 239 | + | |
| 240 | + // connection_aborted() can lag behind proxies and buffering, so cap the tail | |
| 241 | + // or a refresh mid-turn can pin a worker. The client refetches a cut tail. | |
| 242 | + curl_setopt($ch, CURLOPT_TIMEOUT, apply_filters('fs_ai_resume_timeout', 120)); | |
| 243 | + curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 15); | |
| 244 | + // Upstream sends keepalives during quiet gaps, so a stall here means it's dead. | |
| 245 | + curl_setopt($ch, CURLOPT_LOW_SPEED_LIMIT, 1); | |
| 246 | + curl_setopt($ch, CURLOPT_LOW_SPEED_TIME, apply_filters('fs_ai_resume_stall_timeout', 60)); | |
| 247 | + curl_setopt($ch, CURLOPT_TCP_KEEPALIVE, 1); | |
| 248 | + curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true); | |
| 249 | + curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming | |
| 250 | + | |
| 251 | + curl_exec($ch); | |
| 252 | + | |
| 253 | + if (curl_error($ch)) { | |
| 254 | + echo "event: error\n"; | |
| 255 | + echo "data: " . wp_json_encode(['error' => curl_error($ch)]) . "\n\n"; | |
| 256 | + flush(); | |
| 257 | + } | |
| 258 | + | |
| 259 | + curl_close($ch); | |
| 260 | + // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init | |
| 261 | + // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt | |
| 262 | + // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec | |
| 263 | + // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close | |
| 264 | + // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error | |
| 265 | + } | |
| 266 | + | |
| 177 | 267 | protected function sendRequest(array $payload) |
| 178 | 268 | { |
| 179 | 269 | $headers = [ |
| 180 | 270 | 'Content-Type' => 'application/json', |
| 181 | 271 | ]; |
| 272 | + | |
| 273 | + if ($this->apiKey !== '') { | |
| 274 | + $headers['Authorization'] = 'Bearer ' . $this->apiKey; | |
| 275 | + } | |
| 182 | 276 | |
| 183 | 277 | $timeout = apply_filters('fs_ai_request_timeout', 60); |
| 184 | 278 | |
| 185 | 279 | return wp_remote_post($this->apiUrl, [ |