Roog
6d934f4e34
- 默认切换 AgentProvider 为 HttpAgentProvider，增强网络请求的容错和重试机制 - 优化 Run 逻辑，支持多场景去重与并发保护 - 添加 Redis 发布失败的日志记录以提升问题排查效率 - 扩展 OpenAPI 规范，新增 Error 和 Run 状态相关模型 - 增强测试覆盖，验证调度策略和重复请求的幂等性 - 增加数据库索引以优化查询性能 - 更新所有相关文档和配置文件
127 lines
4.0 KiB
PHP
127 lines
4.0 KiB
PHP
<?php
|
|
|
|
namespace App\Services;
|
|
|
|
use App\Services\Agent\AgentProviderInterface;
|
|
use App\Services\Agent\DummyAgentProvider;
|
|
use App\Services\Agent\ProviderException;
|
|
use App\Models\Message;
|
|
|
|
class RunLoop
|
|
{
|
|
private const TERMINAL_STATUSES = ['DONE', 'FAILED', 'CANCELED'];
|
|
|
|
public function __construct(
|
|
private readonly ContextBuilder $contextBuilder,
|
|
private readonly AgentProviderInterface $provider,
|
|
private readonly OutputSink $outputSink,
|
|
private readonly CancelChecker $cancelChecker,
|
|
) {
|
|
}
|
|
|
|
public function run(string $sessionId, string $runId): void
|
|
{
|
|
if ($this->isRunTerminal($sessionId, $runId)) {
|
|
return;
|
|
}
|
|
|
|
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
|
|
$this->appendCanceled($sessionId, $runId);
|
|
return;
|
|
}
|
|
|
|
$context = $this->contextBuilder->build($sessionId, $runId);
|
|
|
|
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
|
|
$this->appendCanceled($sessionId, $runId);
|
|
return;
|
|
}
|
|
|
|
$providerName = $this->resolveProviderName();
|
|
$startedAt = microtime(true);
|
|
|
|
logger('agent provider request', [
|
|
'sessionId' => $sessionId,
|
|
'runId' => $runId,
|
|
'provider' => $providerName,
|
|
]);
|
|
|
|
try {
|
|
$reply = $this->provider->generate($context);
|
|
} catch (ProviderException $exception) {
|
|
$latencyMs = (int) ((microtime(true) - $startedAt) * 1000);
|
|
|
|
$this->outputSink->appendError($sessionId, $runId, $exception->errorCode, $exception->getMessage(), [
|
|
'retryable' => $exception->retryable,
|
|
'http_status' => $exception->httpStatus,
|
|
'provider' => $providerName,
|
|
'latency_ms' => $latencyMs,
|
|
'raw_message' => $exception->rawMessage,
|
|
], "run:{$runId}:error:provider");
|
|
|
|
$this->outputSink->appendRunStatus($sessionId, $runId, 'FAILED', [
|
|
'error' => $exception->getMessage(),
|
|
'dedupe_key' => "run:{$runId}:status:FAILED",
|
|
]);
|
|
|
|
throw $exception;
|
|
}
|
|
$latencyMs = (int) ((microtime(true) - $startedAt) * 1000);
|
|
|
|
logger('agent provider response', [
|
|
'sessionId' => $sessionId,
|
|
'runId' => $runId,
|
|
'provider' => $providerName,
|
|
'latency_ms' => $latencyMs,
|
|
]);
|
|
|
|
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
|
|
$this->appendCanceled($sessionId, $runId);
|
|
return;
|
|
}
|
|
|
|
$this->outputSink->appendAgentMessage($sessionId, $runId, $reply, [
|
|
'provider' => $providerName,
|
|
], "run:{$runId}:agent:message");
|
|
|
|
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
|
|
$this->appendCanceled($sessionId, $runId);
|
|
return;
|
|
}
|
|
|
|
$this->outputSink->appendRunStatus($sessionId, $runId, 'DONE', [
|
|
'dedupe_key' => "run:{$runId}:status:DONE",
|
|
]);
|
|
}
|
|
|
|
private function isRunTerminal(string $sessionId, string $runId): bool
|
|
{
|
|
$latestStatus = Message::query()
|
|
->where('session_id', $sessionId)
|
|
->where('type', 'run.status')
|
|
->whereRaw("payload->>'run_id' = ?", [$runId])
|
|
->orderByDesc('seq')
|
|
->first();
|
|
|
|
$status = $latestStatus ? ($latestStatus->payload['status'] ?? null) : null;
|
|
|
|
return in_array($status, self::TERMINAL_STATUSES, true);
|
|
}
|
|
|
|
private function appendCanceled(string $sessionId, string $runId): void
|
|
{
|
|
$this->outputSink->appendRunStatus($sessionId, $runId, 'CANCELED', [
|
|
'dedupe_key' => "run:{$runId}:status:CANCELED",
|
|
]);
|
|
}
|
|
|
|
private function resolveProviderName(): string
|
|
{
|
|
if ($this->provider instanceof DummyAgentProvider) {
|
|
return 'dummy';
|
|
}
|
|
|
|
return str_replace("\0", '', get_class($this->provider));
|
|
}
|
|
}
|