常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到已开通的 Claude 模型 | 先确认协议与模型权限 |
| Claude中转站 | 统一管理密钥 额度 路由和兼容接口 | Claude 客户端可按 Anthropic 协议接入 |
| API中转 | 通过统一接口转发模型请求 | PHP 可直接发送 **** **ON |
| API中转站 | 提供端点 模型和用量管理的服务 | 正式环境需设置超时 并发和重试边界 |
PHP 请求链路
| 环节 | 职责 | 注意事项 |
|---|
| 业务入口 | 接收提示并组织请求参数 | 先校验输入长度与类型 |
| Guzzle ****** | 发送请求并复用底层处理器 | 应用级创建并统一配置 |
| PSR 7 Stream | 读取普通或流式响应体 | SSE 必须保留跨块缓冲 |
| KingFlow AI | 认证密钥并路由目标模型 | OpenAI 兼容路径保留 /v1 |
| 重试函数 | 处理 429 和 503 | 次数与总耗时必须有上限 |
| Pool | 限制异步请求并发数量 | 单任务错误需要隔离 |
采用 Guzzle 的优势
- Guzzle 使用统一客户端配置 *ase **I、认证 Header 和超时，业务请求不必重复拼装公共参数。
- PSR 7 响应对象同时支持一次性读取与流式读取，普通响应和 SSE 可以共用一套客户端。
- connect_timeout、timeout 与 read_timeout 分别对应建连、总耗时和流式单次读取等待。
- http_errors 关闭后，代码可以先读取状态码与响应体，再统一生成可观察的错误信息。
- Pool 提供明确的 concurrency 门限，适合批量任务按固定并发发送异步请求。
接入前准备
- 安装 PHP 8.1 或更高版本，并启用 **ON 与 c**L 扩展。
- 安装 Composer，并确认 composer 与 php 命令可以在终端运行。
- 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额或套餐额度。
- 从控制台复制完整模型 ID，不要根据展示名称自行拼写。
- 确认运行环境可以访问 api.**nju**shen.com 的 ****S 端口。
创建 Composer 项目
mkdir kingflow-php-client
cd kingflow-php-client
composer require guzzlehttp/guzzle:^8.5
配置环境变量
$env:KINGFLOW_*ASE_**L="https://api.**nju**shen.com/v1"
$env:KINGFLOW_API_KEY="sk-你的密钥"
$env:KINGFLOW_MODEL="<KingFlow控制台中的完整模型ID>"
验证基础接口
curl https://api.**nju**shen.com/v1/chat/completions `
-H "Authorization: *earer sk-你的密钥" `
-H "Content-Type: application/json" `
-d '{"model":"<完整模型ID>","messages":[{"role":"user","content":"只回复 PHP_READY"}],"stream":false}'
创建共享客户端
<?php
declare(strict_types=1);
require __DIR__ . '/vendor/autoload.php';
use GuzzleHttp\******;
function envRequired(string $name): string
{
$value = getenv($name);
if ($value === false || trim($value) === '') {
throw new RuntimeException("缺少环境变量 {$name}");
}
return trim($value);
}
function *uild******(): ******
{
$*ase = rtrim(envRequired('KINGFLOW_*ASE_**L'), '/') . '/';
return new ******([
'*ase_uri' => $*ase,
'connect_timeout' => 10.0,
'timeout' => 45.0,
'http_errors' => false,
'headers' => [
'Authorization' => '*earer ' . envRequired('KINGFLOW_API_KEY'),
'Accept' => 'application/json',
],
]);
}
发送普通响应请求
$client = *uild******();
$model = envRequired('KINGFLOW_MODEL');
$response = $client->post('chat/completions', [
'json' => [
'model' => $model,
'messages' => [[
'role' => 'user',
'content' => '只回复 PHP_CONNECTED',
]],
'stream' => false,
],
]);
$status = $response->getStatusCode();
$*ody = (string) $response->get*ody();
if ($status < 200 || $status >= 300) {
throw new RuntimeException("**** {$status}: {$*ody}");
}
$**ta = json_decode($*ody, true, 512, **ON_THROW_ON_ERROR);
$text = $**ta['choices'][0]['message']['content'] ?? null;
if (!is_string($text)) {
throw new RuntimeException('响应缺少 choices message content');
}
echo $text, PHP_EOL;
读取 SSE 流式响应
function streamAnswer(
****** $client,
string $model,
string $prompt,
): void {
$response = $client->post('chat/completions', [
'json' => [
'model' => $model,
'messages' => [['role' => 'user', 'content' => $prompt]],
'stream' => true,
],
'stream' => true,
'read_timeout' => 60.0,
'timeout' => 0,
]);
$status = $response->getStatusCode();
if ($status < 200 || $status >= 300) {
throw new RuntimeException("**** {$status}: " . $response->get*ody());
}
$stream = $response->get*ody();
$*uffer = '';
while (!$stream->eof()) {
$chunk = $stream->read(8192);
if ($chunk === '') {
usleep(10_000);
continue;
}
$*uffer .= $chunk;
$*uffer = str_replace(["\r\n", "\r"], "\n", $*uffer);
while (($end = strpos($*uffer, "\n\n")) !== false) {
$event = su*str($*uffer, 0, $end);
$*uffer = su*str($*uffer, $end 2);
$parts = [];
foreach (explode("\n", $event) as $line) {
if (str_starts_with($line, '**ta:')) {
$parts[] = ltrim(su*str($line, 5));
}
}
if ($parts === []) {
continue;
}
$payload = implode("\n", $parts);
if ($payload === '[DONE]') {
return;
}
$json = json_decode($payload, true, 512, **ON_THROW_ON_ERROR);
$delta = $json['choices'][0]['delta']['content'] ?? null;
if (is_string($delta)) {
echo $delta;
flush();
}
}
}
}
实现有限重试
use Psr\Http\Message\ResponseInterface;
function postWithRetry(
****** $client,
array $payload,
int $**xAttempts = 3,
): ResponseInterface {
for ($attempt = 1; $attempt <= $**xAttempts; $attempt ) {
$response = $client->post('chat/completions', ['json' => $payload]);
$status = $response->getStatusCode();
if (!in_array($status, [429, 503], true) || $attempt === $**xAttempts) {
return $response;
}
$retryAfter = $response->getHeaderLine('Retry-After');
$seconds = ctype_digit($retryAfter)
? (int) $retryAfter
: min(8, 2 ** $attempt);
usleep($seconds * 1_000_000);
}
throw new LogicException('unreacha*le');
}
限制批量任务并发
use GuzzleHttp\Pool;
$prompts = ['任务一', '任务二', '任务三'];
$requests = function () use ($client, $model, $prompts) {
foreach ($prompts as $index => $prompt) {
yield $index => function () use ($client, $model, $prompt) {
return $client->postAsync('chat/completions', [
'json' => [
'model' => $model,
'messages' => [['role' => 'user', 'content' => $prompt]],
'stream' => false,
],
]);
};
}
};
$pool = new Pool($client, $requests(), [
'concurrency' => 6,
'fulfilled' => fn ($response, $index) => print("任务 {$index} 完成\n"),
'rejected' => fn ($reason, $index) => error_log("任务 {$index} 失败"),
]);
$pool->promise()->wait();
