roog/ars-backend
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

main: 扩展 Agent Run 调度与队列功能

Browse Source 
- 增加 Agent Run MVP-0，包括 RunDispatcher 和 AgentRunJob
- 优化队列配置，支持 Redis 队列驱动，添加 Horizon 容器
- 更新 Docker 配置，细化角色分工，新增 Horizon 配置
- 增加测试任务 `TestJob`，扩展队列使用示例
- 更新 OpenAPI 规范，添加 Agent Run 相关接口及示例
- 编写文档，详细描述 Agent Run 流程与 MVP-0 功能
- 优化相关服务与文档，支持队列与异步运行
This commit is contained in:
Roog
2025-12-17 02:39:31 +08:00
parent dafa8f6b06
commit c55534ad20
42 changed files with 2596 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/Http/Controllers/ChatSessionController.php
View File

@@ -9,12 +9,16 @@ use App\Http\Requests\UpdateSessionRequest;
use App\Http\Resources\ChatSessionResource;
use App\Http\Resources\MessageResource;
use App\Services\ChatService;
use App\Services\RunDispatcher;
use Illuminate\Http\JsonResponse;
use Illuminate\Http\Request;
class ChatSessionController extends Controller
{
public function __construct(private readonly ChatService $service)
public function __construct(
private readonly ChatService $service,
private readonly RunDispatcher $runDispatcher,
)
{
}
@@ -45,6 +49,10 @@ class ChatSessionController extends Controller
'session_id' => $sessionId,
...$request->validated(),
]);
if ($message->role === 'USER' && $message->type === 'user.prompt') {
$this->runDispatcher->dispatchForPrompt($sessionId, $message->message_id);
}
} catch (ChatSessionStatusException $e) {
return response()->json(['message' => $e->getMessage()], 403);
}

99
app/Http/Controllers/ChatSessionSseController.php
View File

@@ -24,7 +24,12 @@ class ChatSessionSseController extends Controller
$limit = (int) $request->query('limit', 200);
$limit = $limit > 0 && $limit <= 500 ? $limit : 200;
if (app()->runningUnitTests() || app()->environment('testing') || ! class_exists(\Redis::class)) {
$useBacklogOnly = app()->runningUnitTests()
|| app()->environment('testing')
|| defined('PHPUNIT_COMPOSER_INSTALL')
|| ! class_exists(\Redis::class);
if ($useBacklogOnly) {
$lastSentSeq = $afterSeq;
ob_start();
$this->sendBacklog($sessionId, $lastSentSeq, $limit, false);
@@ -42,41 +47,75 @@ class ChatSessionSseController extends Controller
$this->sendBacklog($sessionId, $lastSentSeq, $limit);
$redis = Redis::connection()->client();
if (method_exists($redis, 'setOption')) {
$redis->setOption(\Redis::OPT_READ_TIMEOUT, 5);
}
$channel = "session:{$sessionId}:messages";
$pubSub = $redis->pubSubLoop();
$pubSub->subscribe($channel);
$lastPing = time();
foreach ($pubSub as $message) {
if ($message->kind === 'subscribe') {
continue;
try {
$redis = Redis::connection()->client();
if (method_exists($redis, 'setOption')) {
$redis->setOption(\Redis::OPT_READ_TIMEOUT, 5);
}
if (connection_aborted()) {
$pubSub->unsubscribe();
break;
}
$channel = "session:{$sessionId}:messages";
$lastPing = time();
logger()->info('sse open');
if (method_exists($redis, 'pubSubLoop')) {
$pubSub = $redis->pubSubLoop();
$pubSub->subscribe($channel);
$payloadId = $message->payload ?? null;
if ($payloadId) {
$msg = $this->service->getMessage($sessionId, $payloadId);
if ($msg && $msg->seq > $lastSentSeq) {
$this->emitMessage($msg);
$lastSentSeq = $msg->seq;
foreach ($pubSub as $message) {
if ($message->kind === 'subscribe') {
continue;
}
if (connection_aborted()) {
$pubSub->unsubscribe();
break;
}
$payloadId = $message->payload ?? null;
if ($payloadId) {
$msg = $this->service->getMessage($sessionId, $payloadId);
if ($msg && $msg->seq > $lastSentSeq) {
$this->emitMessage($msg);
$lastSentSeq = $msg->seq;
}
}
if (time() - $lastPing >= 180) {
logger()->info('ping: sent'.$sessionId);
echo ": ping\n\n";
@ob_flush();
@flush();
$lastPing = time();
}
}
}
logger()->info('close: sent'.$sessionId);
unset($pubSub);
} else {
// Fallback for Redis drivers without pubSubLoop (older phpredis)
$redis->subscribe([$channel], function ($redisInstance, $chan, $payload) use (&$lastSentSeq, $sessionId) {
if (connection_aborted()) {
$redisInstance->unsubscribe([$chan]);
return;
}
if (time() - $lastPing >= 20) {
echo ": ping\n\n";
@ob_flush();
@flush();
$lastPing = time();
if (! $payload) {
return;
}
$msg = $this->service->getMessage($sessionId, $payload);
if ($msg && $msg->seq > $lastSentSeq) {
$this->emitMessage($msg);
$lastSentSeq = $msg->seq;
}
});
}
} catch (\RedisException $exception) {
logger()->warning('SSE redis subscription failed', [
'session_id' => $sessionId,
'message' => $exception->getMessage(),
]);
echo ": redis-error\n\n";
@ob_flush();
@flush();
}
});

28
app/Http/Controllers/RunController.php Normal file
View File

@@ -0,0 +1,28 @@
<?php
namespace App\Http\Controllers;
use App\Http\Requests\DispatchRunRequest;
use App\Jobs\TestJob;
use App\Services\RunDispatcher;
use Illuminate\Http\JsonResponse;
class RunController extends Controller
{
public function __construct(private readonly RunDispatcher $dispatcher)
{
}
public function store(string $sessionId, DispatchRunRequest $request): JsonResponse
{
$runId = $this->dispatcher->dispatchForPrompt($sessionId, $request->validated()['trigger_message_id']);
return response()->json(['run_id' => $runId], 201);
}
public function test()
{
$job = TestJob::dispatch();
unset($job);
}
}

23
app/Http/Requests/DispatchRunRequest.php Normal file
View File

@@ -0,0 +1,23 @@
<?php
namespace App\Http\Requests;
use Illuminate\Foundation\Http\FormRequest;
class DispatchRunRequest extends FormRequest
{
public function authorize(): bool
{
return true;
}
/**
* @return array<string, mixed>
*/
public function rules(): array
{
return [
'trigger_message_id' => ['required', 'uuid'],
];
}
}

35
app/Jobs/AgentRunJob.php Normal file
View File

@@ -0,0 +1,35 @@
<?php
namespace App\Jobs;
use App\Services\OutputSink;
use App\Services\RunLoop;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class AgentRunJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public function __construct(public string $sessionId, public string $runId)
{
}
public function handle(RunLoop $loop, OutputSink $sink): void
{
try {
$loop->run($this->sessionId, $this->runId);
} catch (\Throwable $e) {
$sink->appendError($this->sessionId, $this->runId, 'run.failed', $e->getMessage());
$sink->appendRunStatus($this->sessionId, $this->runId, 'FAILED', [
'error' => $e->getMessage(),
'dedupe_key' => "run:{$this->runId}:status:FAILED",
]);
throw $e;
}
}
}

23
app/Jobs/TestJob.php Normal file
View File

@@ -0,0 +1,23 @@
<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class TestJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public function __construct()
{
}
public function handle(): void
{
logger('TestJob');
}
}

4
app/Providers/AppServiceProvider.php
View File

@@ -11,7 +11,9 @@ class AppServiceProvider extends ServiceProvider
*/
public function register(): void
{
//
$this->app->bind(\App\Services\Agent\AgentProviderInterface::class, function () {
return new \App\Services\Agent\DummyAgentProvider();
});
}
/**

31
app/Providers/HorizonServiceProvider.php Normal file
View File

@@ -0,0 +1,31 @@
<?php
namespace App\Providers;
use Illuminate\Support\ServiceProvider;
use Laravel\Horizon\Horizon;
class HorizonServiceProvider extends ServiceProvider
{
/**
* Register any application services.
*/
public function register(): void
{
//
}
/**
* Bootstrap any application services.
*/
public function boot(): void
{
if (! class_exists(Horizon::class)) {
return;
}
Horizon::auth(function ($request) {
return app()->environment('local');
});
}
}

55
app/Providers/TelescopeServiceProvider.php Normal file
View File

@@ -0,0 +1,55 @@
<?php
namespace App\Providers;
use Illuminate\Support\Facades\Gate;
use Laravel\Telescope\IncomingEntry;
use Laravel\Telescope\Telescope;
use Laravel\Telescope\TelescopeApplicationServiceProvider;
class TelescopeServiceProvider extends TelescopeApplicationServiceProvider
{
/**
* Register any application services.
*/
public function register(): void
{
Telescope::night();
$this->hideSensitiveRequestDetails();
$isLocal = $this->app->environment('local');
}
/**
* Prevent sensitive request details from being logged by Telescope.
*/
protected function hideSensitiveRequestDetails(): void
{
if ($this->app->environment('local')) {
return;
}
Telescope::hideRequestParameters(['_token']);
Telescope::hideRequestHeaders([
'cookie',
'x-csrf-token',
'x-xsrf-token',
]);
}
/**
* Register the Telescope gate.
*
* This gate determines who can access Telescope in non-local environments.
*/
protected function gate(): void
{
Gate::define('viewTelescope', function ($user) {
return in_array($user->email, [
//
]);
});
}
}

12
app/Services/Agent/AgentProviderInterface.php Normal file
View File

@@ -0,0 +1,12 @@
<?php
namespace App\Services\Agent;
interface AgentProviderInterface
{
/**
* @param array<string, mixed> $context
* @param array<string, mixed> $options
*/
public function generate(array $context, array $options = []): string;
}

26
app/Services/Agent/DummyAgentProvider.php Normal file
View File

@@ -0,0 +1,26 @@
<?php
namespace App\Services\Agent;
class DummyAgentProvider implements AgentProviderInterface
{
/**
* @param array<string, mixed> $context
* @param array<string, mixed> $options
*/
public function generate(array $context, array $options = []): string
{
$messages = $context['messages'] ?? [];
$lastUser = null;
foreach (array_reverse($messages) as $msg) {
if (($msg['role'] ?? '') === 'USER' && ($msg['type'] ?? '') === 'user.prompt') {
$lastUser = $msg['content'] ?? null;
break;
}
}
$summary = $lastUser ? mb_substr($lastUser, 0, 80) : 'no user prompt';
return sprintf('MVP reply: based on last user input -> %s', $summary);
}
}

42
app/Services/Agent/HttpAgentProvider.php Normal file
View File

@@ -0,0 +1,42 @@
<?php
namespace App\Services\Agent;
use Illuminate\Support\Facades\Http;
class HttpAgentProvider implements AgentProviderInterface
{
protected string $endpoint;
public function __construct(?string $endpoint = null)
{
$this->endpoint = $endpoint ?? config('services.agent_provider.endpoint', '');
}
/**
* @param array<string, mixed> $context
* @param array<string, mixed> $options
*/
public function generate(array $context, array $options = []): string
{
if (empty($this->endpoint)) {
// placeholder to avoid accidental outbound calls when未配置
return (new DummyAgentProvider())->generate($context, $options);
}
$payload = [
'context' => $context,
'options' => $options,
];
$response = Http::post($this->endpoint, $payload);
if (! $response->successful()) {
throw new \RuntimeException('Agent provider failed: '.$response->body());
}
$data = $response->json();
return is_string($data) ? $data : ($data['content'] ?? '');
}
}

18
app/Services/CancelChecker.php Normal file
View File

@@ -0,0 +1,18 @@
<?php
namespace App\Services;
use App\Models\Message;
class CancelChecker
{
public function isCanceled(string $sessionId, string $runId): bool
{
return Message::query()
->where('session_id', $sessionId)
->where('type', 'run.cancel.request')
->whereIn('role', [Message::ROLE_USER, Message::ROLE_SYSTEM])
->where('payload->run_id', $runId)
->exists();
}
}

49
app/Services/ContextBuilder.php Normal file
View File

@@ -0,0 +1,49 @@
<?php
namespace App\Services;
use App\Models\Message;
use Illuminate\Support\Collection;
class ContextBuilder
{
public function __construct(private readonly int $limit = 20)
{
}
/**
* @return array<string, mixed>
*/
public function build(string $sessionId, string $runId): array
{
$messages = $this->loadRecentMessages($sessionId);
return [
'run_id' => $runId,
'session_id' => $sessionId,
'system_prompt' => 'You are an agent inside ARS. Respond concisely in plain text.',
'messages' => $messages->map(function (Message $message) {
return [
'message_id' => $message->message_id,
'role' => $message->role,
'type' => $message->type,
'content' => $message->content,
'seq' => $message->seq,
];
})->values()->all(),
];
}
private function loadRecentMessages(string $sessionId): Collection
{
return Message::query()
->where('session_id', $sessionId)
->whereIn('role', [Message::ROLE_USER, Message::ROLE_AGENT])
->whereIn('type', ['user.prompt', 'agent.message'])
->orderByDesc('seq')
->limit($this->limit)
->get()
->sortBy('seq')
->values();
}
}

63
app/Services/OutputSink.php Normal file
View File

@@ -0,0 +1,63 @@
<?php
namespace App\Services;
use App\Models\Message;
class OutputSink
{
public function __construct(private readonly ChatService $chatService)
{
}
/**
* @param array<string, mixed> $meta
*/
public function appendAgentMessage(string $sessionId, string $runId, string $content, array $meta = []): Message
{
return $this->chatService->appendMessage([
'session_id' => $sessionId,
'role' => Message::ROLE_AGENT,
'type' => 'agent.message',
'content' => $content,
'payload' => array_merge($meta, ['run_id' => $runId]),
]);
}
/**
* @param array<string, mixed> $meta
*/
public function appendRunStatus(string $sessionId, string $runId, string $status, array $meta = []): Message
{
$dedupeKey = $meta['dedupe_key'] ?? null;
unset($meta['dedupe_key']);
return $this->chatService->appendMessage([
'session_id' => $sessionId,
'role' => Message::ROLE_SYSTEM,
'type' => 'run.status',
'payload' => array_merge($meta, [
'run_id' => $runId,
'status' => $status,
]),
'dedupe_key' => $dedupeKey,
]);
}
/**
* @param array<string, mixed> $meta
*/
public function appendError(string $sessionId, string $runId, string $code, string $message, array $meta = []): Message
{
return $this->chatService->appendMessage([
'session_id' => $sessionId,
'role' => Message::ROLE_SYSTEM,
'type' => 'error',
'content' => $code,
'payload' => array_merge($meta, [
'run_id' => $runId,
'message' => $message,
]),
]);
}
}

60
app/Services/RunDispatcher.php Normal file
View File

@@ -0,0 +1,60 @@
<?php
namespace App\Services;
use App\Jobs\AgentRunJob;
use App\Models\Message;
use Illuminate\Database\Eloquent\ModelNotFoundException;
use Illuminate\Support\Str;
class RunDispatcher
{
public function __construct(
private readonly ChatService $chatService,
private readonly OutputSink $outputSink,
) {
}
/**
* @throws ModelNotFoundException
*/
public function dispatchForPrompt(string $sessionId, string $triggerMessageId): string
{
$triggerMessage = $this->chatService->getMessage($sessionId, $triggerMessageId);
if (! $triggerMessage) {
throw (new ModelNotFoundException())->setModel(Message::class, [$triggerMessageId]);
}
$existingForTrigger = Message::query()
->where('session_id', $sessionId)
->where('type', 'run.status')
->where('payload->trigger_message_id', $triggerMessageId)
->orderByDesc('seq')
->first();
if ($existingForTrigger && ($existingForTrigger->payload['run_id'] ?? null)) {
return $existingForTrigger->payload['run_id'];
}
$latestStatus = Message::query()
->where('session_id', $sessionId)
->where('type', 'run.status')
->orderByDesc('seq')
->first();
if ($latestStatus && ($latestStatus->payload['status'] ?? null) === 'RUNNING' && ($latestStatus->payload['run_id'] ?? null)) {
return $latestStatus->payload['run_id'];
}
$runId = (string) Str::uuid();
$this->outputSink->appendRunStatus($sessionId, $runId, 'RUNNING', [
'trigger_message_id' => $triggerMessageId,
'dedupe_key' => 'run:trigger:'.$triggerMessageId,
]);
dispatch(new AgentRunJob($sessionId, $runId));
return $runId;
}
}

52
app/Services/RunLoop.php Normal file
View File

@@ -0,0 +1,52 @@
<?php
namespace App\Services;
use App\Services\Agent\AgentProviderInterface;
use App\Services\Agent\DummyAgentProvider;
class RunLoop
{
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->cancelChecker->isCanceled($sessionId, $runId)) {
$this->outputSink->appendRunStatus($sessionId, $runId, 'CANCELED', [
'dedupe_key' => "run:{$runId}:status:CANCELED",
]);
return;
}
$context = $this->contextBuilder->build($sessionId, $runId);
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
$this->outputSink->appendRunStatus($sessionId, $runId, 'CANCELED', [
'dedupe_key' => "run:{$runId}:status:CANCELED",
]);
return;
}
$reply = $this->provider->generate($context);
if ($this->cancelChecker->isCanceled($sessionId, $runId)) {
$this->outputSink->appendRunStatus($sessionId, $runId, 'CANCELED', [
'dedupe_key' => "run:{$runId}:status:CANCELED",
]);
return;
}
$this->outputSink->appendAgentMessage($sessionId, $runId, $reply, [
'provider' => $this->provider instanceof DummyAgentProvider ? 'dummy' : get_class($this->provider),
]);
$this->outputSink->appendRunStatus($sessionId, $runId, 'DONE', [
'dedupe_key' => "run:{$runId}:status:DONE",
]);
}
}