常用接入术语

PHP 请求链路

采用 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();

运行和验收

1. 设置三个环境变量，并确认 *ase **L 末尾没有重复斜杠。
2. 先执行 curl，确认返回 PHP_READY。
3. 创建 Composer 项目并安装 Guzzle。
4. 保存共享客户端与普通响应示例，执行 php -l 检查语法。
5. 运行程序，确认返回 PHP_CONNECTED。
6. 加入 SSE 函数，确认文本按事件持续输出且 DONE 正常结束。
7. 模拟 429 和 503，确认总尝试次数不超过三次。
8. 用少量任务验证 Pool 的并发门限与单任务错误隔离。
9. 在 KingFlow AI 控制台核对模型、调用次数和 Token 用量。

正式环境配置

• 在应用生命周期内复用一个 Guzzle ******，并集中配置 *ase **I、认证 Header 与超时。
• 使用 c**L Handler 以获得 connect_timeout 支持，并保留有效的 TLS 证书校验。
• 普通请求设置总超时；SSE 使用 read_timeout 控制单次读取等待，并由业务层控制流式总时长。
• SSE 解析必须保留跨块缓冲，不能假设一次 read 正好得到一个完整事件。
• 流式响应已经输出内容后不要自动从头重试，否则用户可能看到重复片段。
• 只对 429、503 和确认可重试的传输错误执行有限退避，并设置总时长上限。
• 记录请求 ID、模型、状态码、耗时和 Token 用量，不记录完整密钥与敏感原文。

密钥和数据安全

• API Key 只保存在环境变量、密钥管理器或部署平台 Secret 中，并定期轮换。
• 开发、测试和生产使用不同密钥与额度，避免跨环境共享凭据。
• 不要关闭 TLS verify，也不要把 Authorization Header 写入公开日志。
• 把模型输出视为不可信数据，进入页面、命令或数据库前执行校验与转义。

上线检查清单

• 基础 curl 请求已经返回 PHP_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• 源码和日志中没有明文 API Key。
• 应用级复用 Guzzle ******，公共配置保持一致。
• 建连、总耗时和流式读超时已经分别设置。
• SSE 解析保留跨块缓冲并识别 **ta 与 DONE。
• Pool concurrency 限制了并发，入口队列也有边界。
• 重试次数有限，只覆盖明确可重试的状态码。
• 控制台中的模型、调用次数和用量符合预期。

常见故障排查

最小上线策略

• 先用一个模型和少量并发上线，确认状态码、延迟与 Token 用量符合预期后再逐步扩大流量。
• 把 ******、模型 ID、并发门限和超时参数放在集中配置层，不要散落在业务代码中。
• 为 401、404、429、503 和超时设置独立指标，告警中保留请求 ID，但不记录完整提示与密钥。
• 发布前保存上一版配置和可运行构建，出现连续错误或延迟异常时可以快速回滚。
• 每次更换模型、额度方案或重试策略后，重新执行普通响应、SSE 和并发验收。