main: 增强 Agent Run 逻辑与消息处理

- 添加流式文本推送,支持 `message.delta` 消息类型
- 优化 Run 主流程,增加工具调用与流式数据发布逻辑
- 更新 `phpunit.xml` 环境变量,支持 Agent 配置项
- 扩展文档,完善工具调用与消息类型说明
This commit is contained in:
2025-12-22 17:51:56 +08:00
parent 59d4831f00
commit 663e15395b
5 changed files with 766 additions and 143 deletions

View File

@@ -37,10 +37,17 @@ class RunLoop
}
/**
* 运行单次 Agent Run run_id 幂等)负责取消检查、Provider 调用和结果落库
* 运行单次 Agent Run run_id 幂等)。
*
* 主流程:
* 1. 检查 run 是否已终止
* 2. 进入主循环,持续调用 Provider 直到完成或失败
* 3. 每轮迭代可能触发工具调用,工具完成后继续下一轮
* 4. 没有工具调用时,写入最终回复并标记 DONE
*/
public function run(string $sessionId, string $runId): void
{
// 1. 幂等性检查:避免重复执行已完成的 run
if ($this->isRunTerminal($sessionId, $runId)) {
return;
}
@@ -48,137 +55,346 @@ class RunLoop
$providerName = $this->resolveProviderName();
$toolCallCount = 0;
// 2. 主循环:持续调用 Provider 直到完成或失败
while (true) {
if ($this->isCanceled($sessionId, $runId)) {
$this->appendCanceled($sessionId, $runId);
// 2.1 检查用户是否取消
if ($this->checkAndHandleCancel($sessionId, $runId)) {
return;
}
$context = $this->contextBuilder->build($sessionId, $runId);
$providerOptions = [
'should_stop' => fn () => $this->isCanceled($sessionId, $runId),
];
// 2.2 执行一轮 Provider 调用
$iterationResult = $this->executeProviderIteration(
$sessionId,
$runId,
$providerName,
$toolCallCount
);
// 达到工具调用上限后强制关闭后续工具调用,避免再次触发 TOOL_CALL_LIMIT。
if ($toolCallCount >= $this->maxToolCalls) {
$providerOptions['tool_choice'] = 'none';
}
$logOptions = $providerOptions;
unset($logOptions['should_stop']);
logger('agent provider context', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'context' => $context,
'provider_options' => $logOptions,
]);
$startedAt = microtime(true);
logger('agent provider request', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'iteration' => $toolCallCount,
]);
// 单轮 Agent 调用(可能触发工具调用,后续再进下一轮)
$streamState = $this->consumeProviderStream($sessionId, $runId, $context, $providerName, $startedAt, $providerOptions);
if ($streamState['canceled'] || $streamState['failed']) {
// 2.3 处理失败或取消
if ($iterationResult['should_exit']) {
return;
}
if (! empty($streamState['tool_calls'])) {
$toolCallCount += count($streamState['tool_calls']);
if ($toolCallCount > $this->maxToolCalls) {
$this->appendProviderFailure(
$sessionId,
$runId,
'TOOL_CALL_LIMIT',
'Tool call limit reached for this run',
$providerName,
$this->latencyMs($startedAt),
[],
'TOOL_CALL_LIMIT'
);
// 2.4 如果有工具调用,处理工具执行流程
if ($iterationResult['has_tool_calls']) {
$shouldExit = $this->handleToolCalls(
$sessionId,
$runId,
$providerName,
$iterationResult,
$toolCallCount
);
if ($shouldExit) {
return;
}
// 工具调用:先调度子 Run再等待 tool.result随后继续下一轮 Provider 调用
$toolCalls = $this->dispatchToolRuns($sessionId, $runId, $streamState['tool_calls']);
$waitState = $this->awaitToolResults($sessionId, $runId, $toolCalls, $providerName);
if ($waitState['failed'] || $waitState['canceled']) {
return;
}
// 工具结果已写回上下文,继续下一轮 Agent 调用。
// 更新工具调用计数,继续下一轮 Provider 调用
$toolCallCount = $iterationResult['updated_tool_count'];
continue;
}
$latencyMs = $this->latencyMs($startedAt);
logger('agent provider response', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'latency_ms' => $latencyMs,
]);
if ($this->isCanceled($sessionId, $runId)) {
$this->appendCanceled($sessionId, $runId);
return;
}
if (! $streamState['received_event']) {
$this->appendProviderFailure(
$sessionId,
$runId,
'EMPTY_STREAM',
'Agent provider returned no events',
$providerName,
$latencyMs,
[],
'EMPTY_STREAM'
);
return;
}
if ($streamState['done_reason'] === null) {
$this->appendProviderFailure(
$sessionId,
$runId,
'STREAM_INCOMPLETE',
'Agent provider stream ended unexpectedly',
$providerName,
$latencyMs,
[],
'STREAM_INCOMPLETE'
);
return;
}
$this->outputSink->appendAgentMessage($sessionId, $runId, $streamState['reply'], [
'provider' => $providerName,
'done_reason' => $streamState['done_reason'],
], "run:{$runId}:agent:message");
if ($this->isCanceled($sessionId, $runId)) {
$this->appendCanceled($sessionId, $runId);
return;
}
$this->outputSink->appendRunStatus($sessionId, $runId, 'DONE', [
'dedupe_key' => "run:{$runId}:status:DONE",
]);
// 2.5 没有工具调用,完成 run
$this->completeRun(
$sessionId,
$runId,
$providerName,
$iterationResult['stream_state'],
$iterationResult['latency_ms']
);
return;
}
}
/**
* 检查并处理取消状态。
*
* @return bool 是否已处理取消true 表示已取消并写入状态)
*/
private function checkAndHandleCancel(string $sessionId, string $runId): bool
{
if ($this->isCanceled($sessionId, $runId)) {
$this->appendCanceled($sessionId, $runId);
return true;
}
return false;
}
/**
* 执行一轮 Provider 调用迭代。
*
* 包括:
* - 构建上下文
* - 准备 Provider 选项(工具调用限制、取消回调等)
* - 调用 Provider 流式接口
* - 记录日志
*
* @return array{
* stream_state: array,
* has_tool_calls: bool,
* updated_tool_count: int,
* should_exit: bool,
* latency_ms: int
* }
*/
private function executeProviderIteration(
string $sessionId,
string $runId,
string $providerName,
int $currentToolCallCount
): array {
// 1. 构建上下文和 Provider 选项
$context = $this->contextBuilder->build($sessionId, $runId);
$providerOptions = $this->buildProviderOptions($sessionId, $runId, $currentToolCallCount);
// 2. 记录调用日志
$this->logProviderRequest($sessionId, $runId, $providerName, $context, $providerOptions, $currentToolCallCount);
// 3. 调用 Provider 并消费事件流
$startedAt = microtime(true);
$streamState = $this->consumeProviderStream(
$sessionId,
$runId,
$context,
$providerName,
$startedAt,
$providerOptions
);
$latencyMs = $this->latencyMs($startedAt);
// 4. 检查流式调用是否失败或取消
if ($streamState['canceled'] || $streamState['failed']) {
return [
'stream_state' => $streamState,
'has_tool_calls' => false,
'updated_tool_count' => $currentToolCallCount,
'should_exit' => true,
'latency_ms' => $latencyMs,
];
}
// 5. 检查是否有工具调用
$hasToolCalls = !empty($streamState['tool_calls']);
$updatedToolCount = $currentToolCallCount + count($streamState['tool_calls']);
return [
'stream_state' => $streamState,
'has_tool_calls' => $hasToolCalls,
'updated_tool_count' => $updatedToolCount,
'should_exit' => false,
'latency_ms' => $latencyMs,
];
}
/**
* 构建 Provider 调用选项。
*
* 包括:
* - 取消检查回调
* - 工具调用限制控制
*/
private function buildProviderOptions(string $sessionId, string $runId, int $toolCallCount): array
{
$options = [
'should_stop' => fn () => $this->isCanceled($sessionId, $runId),
];
// 达到工具调用上限后,强制禁用工具调用,避免再次触发 TOOL_CALL_LIMIT 错误
if ($toolCallCount >= $this->maxToolCalls) {
$options['tool_choice'] = 'none';
}
return $options;
}
/**
* 记录 Provider 请求日志。
*/
private function logProviderRequest(
string $sessionId,
string $runId,
string $providerName,
AgentContext $context,
array $providerOptions,
int $iteration
): void {
// 日志选项(移除不可序列化的回调)
$logOptions = $providerOptions;
unset($logOptions['should_stop']);
logger('agent provider context', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'context' => $context,
'provider_options' => $logOptions,
]);
logger('agent provider request', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'iteration' => $iteration,
]);
}
/**
* 处理工具调用流程。
*
* 流程:
* 1. 检查工具调用数量是否超限
* 2. 分发工具子 Run
* 3. 等待工具执行结果
*
* @return bool 是否应该退出主循环(超限、失败或取消时返回 true
*/
private function handleToolCalls(
string $sessionId,
string $runId,
string $providerName,
array $iterationResult,
int $originalToolCallCount
): bool {
$streamState = $iterationResult['stream_state'];
$latencyMs = $iterationResult['latency_ms'];
$updatedToolCount = $iterationResult['updated_tool_count'];
// 1. 检查工具调用数量是否超限
if ($updatedToolCount > $this->maxToolCalls) {
$this->appendProviderFailure(
$sessionId,
$runId,
'TOOL_CALL_LIMIT',
'Tool call limit reached for this run',
$providerName,
$latencyMs,
[],
'TOOL_CALL_LIMIT'
);
return true; // 退出主循环
}
// 2. 分发工具子 Run
$toolCalls = $this->dispatchToolRuns($sessionId, $runId, $streamState['tool_calls']);
// 3. 等待所有工具执行完成
$waitState = $this->awaitToolResults($sessionId, $runId, $toolCalls, $providerName);
// 4. 检查等待过程中是否失败或取消
if ($waitState['failed'] || $waitState['canceled']) {
return true; // 退出主循环
}
// 工具结果已写回上下文,继续下一轮 Agent 调用
return false;
}
/**
* 完成 Run 并写入最终状态。
*
* 流程:
* 1. 验证流式响应的有效性
* 2. 写入最终 agent.message
* 3. 再次检查取消状态
* 4. 写入 run.status = DONE
*/
private function completeRun(
string $sessionId,
string $runId,
string $providerName,
array $streamState,
int $latencyMs
): void {
// 1. 记录响应日志
logger('agent provider response', [
'sessionId' => $sessionId,
'runId' => $runId,
'provider' => $providerName,
'latency_ms' => $latencyMs,
]);
// 2. 再次检查取消状态(在写入最终消息前)
if ($this->checkAndHandleCancel($sessionId, $runId)) {
return;
}
// 3. 验证流式响应的有效性
if (!$this->validateStreamResponse($sessionId, $runId, $providerName, $streamState, $latencyMs)) {
return;
}
// 4. 写入最终 agent.message
$this->outputSink->appendAgentMessage($sessionId, $runId, $streamState['reply'], [
'provider' => $providerName,
'done_reason' => $streamState['done_reason'],
], "run:{$runId}:agent:message");
// 5. 最后一次检查取消状态(在写入 DONE 前)
if ($this->checkAndHandleCancel($sessionId, $runId)) {
return;
}
// 6. 写入 run.status = DONE
$this->outputSink->appendRunStatus($sessionId, $runId, 'DONE', [
'dedupe_key' => "run:{$runId}:status:DONE",
]);
}
/**
* 验证流式响应的有效性。
*
* 检查:
* - 是否收到任何事件(避免空流)
* - 流是否正常结束(有 done_reason
*
* @return bool 是否有效true 表示有效false 表示无效并已写入错误)
*/
private function validateStreamResponse(
string $sessionId,
string $runId,
string $providerName,
array $streamState,
int $latencyMs
): bool {
// 1. 检查是否收到任何事件
if (!$streamState['received_event']) {
$this->appendProviderFailure(
$sessionId,
$runId,
'EMPTY_STREAM',
'Agent provider returned no events',
$providerName,
$latencyMs,
[],
'EMPTY_STREAM'
);
return false;
}
// 2. 检查流是否正常结束
if ($streamState['done_reason'] === null) {
$this->appendProviderFailure(
$sessionId,
$runId,
'STREAM_INCOMPLETE',
'Agent provider stream ended unexpectedly',
$providerName,
$latencyMs,
[],
'STREAM_INCOMPLETE'
);
return false;
}
return true;
}
/**
* 判断指定 run 是否已到终态,避免重复执行。
*/