Skip to content

Commit a540049

Browse files
tymondesignstymondesigns
committed
fixes
1 parent a591664 commit a540049

File tree

15 files changed

+435
-155
lines changed

15 files changed

+435
-155
lines changed

‎src/Agents/Agent.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -431,7 +431,7 @@ protected function invokePipeline(
431431
if ($streaming) {
432432
$event->config->stream->push(new ChatGenerationChunk(ChunkType::Error));
433433
} else {
434-
$this->dispatchEvent(new AgentStepError($this, $event->config->exception, $event->config));
434+
$this->dispatchEvent(new AgentStepError($this, $event->exception, $event->config));
435435
}
436436
})
437437
->invoke($payload, $config);

‎src/Agents/Prebuilt/WeatherAgent.php‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,9 @@ public function prompt(): ChatPromptTemplate|ChatPromptBuilder|string
3434

3535
public function llm(): LLM|string|null
3636
{
37-
return Cortex::llm('ollama', 'gpt-oss:20b')->ignoreFeatures();
37+
// return Cortex::llm('openai_responses', 'gpt-5-mini')->ignoreFeatures();
38+
// return Cortex::llm('ollama', 'gpt-oss:20b')->ignoreFeatures();
39+
return Cortex::llm('openai', 'gpt-4o-mini')->ignoreFeatures();
3840
}
3941

4042
#[Override]

‎src/Agents/Stages/AppendUsage.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ class AppendUsage implements Pipeable
1818
public function handlePipeable(mixed $payload, RuntimeConfig $config, Closure $next): mixed
1919
{
2020
$usage = match (true) {
21-
$payload instanceof ChatResult, $payload instanceof ChatGenerationChunk && $payload->isFinal => $payload->usage,
21+
$payload instanceof ChatResult, $payload instanceof ChatGenerationChunk && $payload->usage !== null => $payload->usage,
2222
default => null,
2323
};
2424

‎src/Agents/Stages/HandleToolCalls.php‎

Lines changed: 97 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,11 @@
55
namespace Cortex\Agents\Stages;
66

77
use Closure;
8+
use Generator;
89
use Cortex\Pipeline;
910
use Cortex\Contracts\Pipeable;
1011
use Cortex\LLM\Data\ChatResult;
12+
use Cortex\LLM\Enums\ChunkType;
1113
use Cortex\Contracts\ChatMemory;
1214
use Cortex\Pipeline\RuntimeConfig;
1315
use Cortex\Support\Traits\CanPipe;
@@ -35,65 +37,118 @@ public function __construct(
3537
) {}
3638

3739
public function handlePipeable(mixed $payload, RuntimeConfig $config, Closure $next): mixed
40+
{
41+
return match (true) {
42+
$payload instanceof ChatGenerationChunk && $payload->type === ChunkType::ToolInputEnd => $this->handleStreamingChunk($payload, $config, $next),
43+
$payload instanceof ChatStreamResult => $this->handleStreamingResult($payload),
44+
default => $this->handleNonStreaming($payload, $config, $next),
45+
};
46+
}
47+
48+
/**
49+
* Handle streaming chunks (individual ChatGenerationChunk objects).
50+
*/
51+
protected function handleStreamingChunk(ChatGenerationChunk $chunk, RuntimeConfig $config, Closure $next): mixed
52+
{
53+
$processedChunk = $next($chunk, $config);
54+
55+
// Process tool calls if needed
56+
if ($chunk->message->hasToolCalls() && $this->currentStep++ < $this->maxSteps) {
57+
$nestedPayload = $this->processToolCalls($chunk, $config);
58+
59+
if ($nestedPayload !== null) {
60+
// Return stream with ToolInputEnd chunk + nested stream
61+
// AbstractLLM will yield from this stream
62+
return new ChatStreamResult(function () use ($processedChunk, $nestedPayload): Generator {
63+
if ($processedChunk instanceof ChatGenerationChunk) {
64+
yield $processedChunk;
65+
}
66+
67+
if ($nestedPayload instanceof ChatStreamResult) {
68+
foreach ($nestedPayload as $nestedChunk) {
69+
yield $nestedChunk;
70+
}
71+
}
72+
});
73+
}
74+
}
75+
76+
return $processedChunk;
77+
}
78+
79+
/**
80+
* Handle streaming results (ChatStreamResult from nested pipeline).
81+
*/
82+
protected function handleStreamingResult(ChatStreamResult $result): ChatStreamResult
83+
{
84+
// This happens when we return a nested stream - AbstractLLM will handle it
85+
return $result;
86+
}
87+
88+
/**
89+
* Handle non-streaming payloads (ChatResult, ChatGeneration, etc.).
90+
*/
91+
protected function handleNonStreaming(mixed $payload, RuntimeConfig $config, Closure $next): mixed
3892
{
3993
$generation = $this->getGeneration($payload);
4094

4195
while ($generation?->message?->hasToolCalls() && $this->currentStep++ < $this->maxSteps) {
42-
// Get the results of the tool calls, represented as tool messages.
43-
$toolMessages = $generation->message->toolCalls->invokeAsToolMessages($this->tools);
44-
45-
// If there are any tool messages, add them to the memory.
46-
// And send them to the execution pipeline to get a new generation.
47-
if ($toolMessages->isNotEmpty()) {
48-
// @phpstan-ignore argument.type
49-
$toolMessages->each(fn(ToolMessage $message) => $this->memory->addMessage($message));
50-
51-
// Track the next step before making the LLM call
52-
$config->context->addNextStep();
53-
54-
// Send the tool messages to the execution stages to get a new generation.
55-
// Create a temporary pipeline from the execution stages.
56-
// Since this is a new Pipeline instance, its Pipeline events won't trigger
57-
// the main pipeline's callbacks due to eventBelongsToThisInstance filtering.
58-
$nestedPipeline = new Pipeline(...$this->executionStages);
59-
60-
// Enable streaming on the nested pipeline if the config has streaming enabled
61-
if ($config->streaming) {
62-
$nestedPipeline->enableStreaming();
63-
}
64-
65-
$payload = $nestedPipeline->invoke([
66-
'messages' => $this->memory->getMessages(),
67-
...$this->memory->getVariables(),
68-
], $config);
69-
70-
// If the payload is a stream result, append any stream buffer chunks
71-
// (like StepStart/StepEnd) that were pushed during the nested pipeline execution
72-
if ($payload instanceof ChatStreamResult) {
73-
$payload = $payload->appendStreamBuffer($config);
74-
}
75-
76-
// Update the generation so that the loop can check the new generation for tool calls.
77-
$generation = $this->getGeneration($payload);
96+
$nestedPayload = $this->processToolCalls($generation, $config);
97+
98+
if ($nestedPayload !== null) {
99+
// Update the generation so that the loop can check the new generation for tool calls
100+
$generation = $this->getGeneration($nestedPayload);
101+
$payload = $nestedPayload;
78102
}
79103
}
80104

81-
// The final step is already properly set - no need to update it
82-
// If it has tool calls, they were set in the while loop
83-
// If it doesn't have tool calls, it was initialized with an empty ToolCallCollection
84-
85105
return $next($payload, $config);
86106
}
87107

108+
/**
109+
* Process tool calls and return the nested pipeline result.
110+
*
111+
* @return ChatResult|ChatStreamResult|null Returns null if no tool calls to process
112+
*/
113+
protected function processToolCalls(ChatGeneration|ChatGenerationChunk $generation, RuntimeConfig $config): ChatResult|ChatStreamResult|null
114+
{
115+
$toolMessages = $generation->message->toolCalls->invokeAsToolMessages($this->tools);
116+
117+
if ($toolMessages->isEmpty()) {
118+
return null;
119+
}
120+
121+
// @phpstan-ignore argument.type
122+
$toolMessages->each(fn(ToolMessage $message) => $this->memory->addMessage($message));
123+
124+
$config->context->addNextStep();
125+
126+
$nestedPipeline = new Pipeline(...$this->executionStages)
127+
->enableStreaming($config->streaming);
128+
129+
$nestedPayload = $nestedPipeline->invoke([
130+
'messages' => $this->memory->getMessages(),
131+
...$this->memory->getVariables(),
132+
], $config);
133+
134+
// If the payload is a stream result, append any stream buffer chunks
135+
// (like StepStart/StepEnd) that were pushed during the nested pipeline execution
136+
if ($nestedPayload instanceof ChatStreamResult) {
137+
return $nestedPayload->appendStreamBuffer($config);
138+
}
139+
140+
return $nestedPayload;
141+
}
142+
88143
/**
89144
* Get the generation from the payload.
90145
*/
91146
protected function getGeneration(mixed $payload): ChatGeneration|ChatGenerationChunk|null
92147
{
93148
return match (true) {
94149
$payload instanceof ChatGeneration => $payload,
95-
// When streaming, only the final chunk will contain the completed tool calls and content.
96-
$payload instanceof ChatGenerationChunk && $payload->isFinal => $payload,
150+
// When streaming, We need to wait for the ToolInputEnd chunk to get the completed tool calls and content.
151+
$payload instanceof ChatGenerationChunk && $payload->type === ChunkType::ToolInputEnd => $payload,
97152
$payload instanceof ChatResult => $payload->generation,
98153
default => null,
99154
};

‎src/Http/Controllers/AgentsController.php‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -23,23 +23,23 @@ public function invoke(string $agent, Request $request): JsonResponse
2323
try {
2424
$agent = Cortex::agent($agent);
2525
$agent->onStart(function (AgentStart $event): void {
26-
dump('-- agent start');
26+
// dump('-- agent start');
2727
});
2828
$agent->onEnd(function (AgentEnd $event): void {
29-
dump('-- agent end');
29+
// dump('-- agent end');
3030
});
3131

3232
$agent->onStepStart(function (AgentStepStart $event): void {
33-
dump(
34-
sprintf('---- step %d start', $event->config?->context?->getCurrentStepNumber()),
35-
// $event->config?->context->toArray(),
36-
);
33+
// dump(
34+
// sprintf('---- step %d start', $event->config?->context?->getCurrentStepNumber()),
35+
// // $event->config?->context->toArray(),
36+
// );
3737
});
3838
$agent->onStepEnd(function (AgentStepEnd $event): void {
39-
dump(
40-
sprintf('---- step %d end', $event->config?->context?->getCurrentStepNumber()),
41-
// $event->config?->toArray(),
42-
);
39+
// dump(
40+
// sprintf('---- step %d end', $event->config?->context?->getCurrentStepNumber()),
41+
// // $event->config?->toArray(),
42+
// );
4343
});
4444
$agent->onStepError(function (AgentStepError $event): void {
4545
// dump(sprintf('step error: %d', $event->config?->context?->getCurrentStepNumber()));
@@ -64,10 +64,10 @@ public function invoke(string $agent, Request $request): JsonResponse
6464

6565
return response()->json([
6666
'result' => $result,
67-
// 'config' => $agent->getRuntimeConfig()?->toArray(),
68-
// 'memory' => $agent->getMemory()->getMessages()->toArray(),
67+
'config' => $agent->getRuntimeConfig()?->toArray(),
6968
'steps' => $agent->getSteps()->toArray(),
7069
'total_usage' => $agent->getTotalUsage()->toArray(),
70+
// 'memory' => $agent->getMemory()->getMessages()->toArray(),
7171
]);
7272
}
7373

@@ -106,7 +106,7 @@ public function stream(string $agent, Request $request): void// : StreamedRespon
106106
try {
107107
foreach ($result as $chunk) {
108108
// dump($chunk->type->value);
109-
dump(sprintf('%s: %s', $chunk->type->value, $chunk->message->content));
109+
dump(sprintf('%s: %s', $chunk->type->value, $chunk->message->content()));
110110
}
111111

112112
// return $result->streamResponse();

‎src/LLM/Drivers/OpenAI/Chat/Concerns/MapsStreamResponse.php‎

Lines changed: 27 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@
55
namespace Cortex\LLM\Drivers\OpenAI\Chat\Concerns;
66

77
use Generator;
8-
use JsonException;
98
use DateTimeImmutable;
9+
use Cortex\LLM\Data\Usage;
1010
use Cortex\LLM\Data\ToolCall;
1111
use Cortex\LLM\Enums\ChunkType;
1212
use Cortex\LLM\Data\FunctionCall;
@@ -65,6 +65,7 @@ protected function mapStreamResponse(StreamResponse $response): ChatStreamResult
6565
$finishReason,
6666
$toolCallsSoFar,
6767
$isActiveText,
68+
$usage,
6869
);
6970

7071
// Now update content and tool call tracking
@@ -137,11 +138,24 @@ protected function mapStreamResponse(StreamResponse $response): ChatStreamResult
137138
}
138139

139140
// This is the last content chunk if we have a finish reason
140-
$isLastContentChunk = $finishReason !== null;
141+
$isLastContentChunk = $usage !== null;
141142

142143
$chunkType = $isActiveText && $isLastContentChunk
143144
? ChunkType::TextEnd
144145
: $chunkType;
146+
} elseif ($usage !== null) {
147+
// This else case will always represent the end of the stream,
148+
// since choices is empty and usage is present.
149+
150+
// We will also correct the end of the tool call if delta is currently set.
151+
if ($chunkType === ChunkType::ToolInputDelta) {
152+
$chunkType = ChunkType::ToolInputEnd;
153+
}
154+
155+
// And the end of the text if delta is currently set.
156+
if ($chunkType === ChunkType::TextDelta) {
157+
$chunkType = ChunkType::TextEnd;
158+
}
145159
}
146160

147161
$chatGenerationChunk = new ChatGenerationChunk(
@@ -159,10 +173,10 @@ protected function mapStreamResponse(StreamResponse $response): ChatStreamResult
159173
),
160174
),
161175
createdAt: DateTimeImmutable::createFromFormat('U', (string) $chunk->created),
162-
finishReason: $finishReason,
176+
finishReason: $usage !== null ? $finishReason : null,
163177
usage: $usage,
164178
contentSoFar: $contentSoFar,
165-
isFinal: $isLastContentChunk && $usage !== null,
179+
isFinal: $usage !== null,
166180
rawChunk: $this->includeRaw ? $chunk->toArray() : null,
167181
);
168182

@@ -175,7 +189,9 @@ protected function mapStreamResponse(StreamResponse $response): ChatStreamResult
175189

176190
yield from $this->streamBuffer?->drain() ?? [];
177191

178-
$this->dispatchEvent(new ChatModelStreamEnd($this, $chatGenerationChunk));
192+
if ($chatGenerationChunk !== null) {
193+
$this->dispatchEvent(new ChatModelStreamEnd($this, $chatGenerationChunk));
194+
}
179195
});
180196
}
181197

@@ -189,6 +205,7 @@ protected function resolveOpenAIChunkType(
189205
?FinishReason $finishReason,
190206
array $toolCallsSoFar,
191207
bool $isActiveText,
208+
?Usage $usage,
192209
): ChunkType {
193210
// Process tool calls
194211
foreach ($choice->delta->toolCalls as $toolCall) {
@@ -210,14 +227,6 @@ protected function resolveOpenAIChunkType(
210227

211228
// If we have arguments in this delta
212229
if ($toolCall->function->arguments !== '') {
213-
// Check if the accumulated arguments (including this delta) are now parseable JSON
214-
$accumulatedArgs = $existingToolCall['function']['arguments'] . $toolCall->function->arguments;
215-
216-
if ($this->isParsableJson($accumulatedArgs) && $finishReason !== null) {
217-
return ChunkType::ToolInputEnd;
218-
}
219-
220-
// Otherwise it's a delta
221230
return ChunkType::ToolInputDelta;
222231
}
223232
}
@@ -234,6 +243,10 @@ protected function resolveOpenAIChunkType(
234243
return ChunkType::TextDelta;
235244
}
236245

246+
if ($finishReason === FinishReason::ToolCalls && $usage !== null) {
247+
return ChunkType::ToolInputEnd;
248+
}
249+
237250
// Default fallback - this handles empty deltas and other cases
238251
// If we have tool calls accumulated, empty delta is ToolInputDelta
239252
// Otherwise, it's TextDelta (for text responses)
@@ -249,12 +262,6 @@ protected function isParsableJson(string $value): bool
249262
return false;
250263
}
251264

252-
try {
253-
json_decode($value, true, flags: JSON_THROW_ON_ERROR);
254-
255-
return true;
256-
} catch (JsonException) {
257-
return false;
258-
}
265+
return json_validate($value);
259266
}
260267
}

0 commit comments

Comments
 (0)