diff --git a/src/agentCore.jl b/src/agentCore.jl index 876051a..57524ea 100644 --- a/src/agentCore.jl +++ b/src/agentCore.jl @@ -264,18 +264,18 @@ function _agentLoop(agent::yiemAgent) while true while newUserMsg === nothing if isready(agent.inputChannel) - agent.eventSink("_agentLoop 1 agent._state.messages length $(length(agent._state.messages))"; publish=false, log=true) + agent.eventSink("_agentLoop 1 agent._state.messages length $(length(agent._state.messages))"; publish=true) # agent process new user msg immediately after the current tool call finished. newUserMsg = take!(agent.inputChannel) - # agent.eventSink("new user msg\n$newUserMsg"; publish=false, log=true) + # agent.eventSink("new user msg\n$newUserMsg"; publish=true) else # check followUp message after _processMessage() is done if typeof(processingTask) == Task && istaskdone(processingTask) == true - agent.eventSink("_agentLoop 2 agent._state.messages length $(length(agent._state.messages))"; publish=false, log=true) + agent.eventSink("_agentLoop 2 agent._state.messages length $(length(agent._state.messages))"; publish=true) # if agent runs is done but followUpChannel has messages, # put new message in inputChannel instead if isready(agent.followUpChannel) - agent.eventSink("_agentLoop 3 agent._state.messages length $(length(agent._state.messages))"; publish=false, log=true) + agent.eventSink("_agentLoop 3 agent._state.messages length $(length(agent._state.messages))"; publish=true) while isready(agent.followUpChannel) followUpMsg = take!(agent.followUpChannel) put!(agent.inputChannel, followUpMsg) @@ -283,18 +283,18 @@ function _agentLoop(agent::yiemAgent) processingTask = nothing # reset result = nothing # reset else # _processMessage() done and no followUp message. - agent.eventSink("_agentLoop 4 agent._state.messages length $(length(agent._state.messages))"; publish=false, log=true) + agent.eventSink("_agentLoop 4 agent._state.messages length $(length(agent._state.messages))"; publish=true) result = deepcopy(agent._state.messages[end]) - agent.eventSink("_agentLoop 4-1 "; publish=false, log=true) + agent.eventSink("_agentLoop 4-1 "; publish=true) # filter out reasoningContent in-place filter!(c -> !(c isa reasoningContent), result.content) - agent.eventSink("_agentLoop 5 "; publish=false, log=true) + agent.eventSink("_agentLoop 5 "; publish=true) # format output respondToUI = _assistantMessageToOpenAI(result) - agent.eventSink("_agentLoop 6 "; publish=false, log=true) + agent.eventSink("_agentLoop 6 "; publish=true) put!(agent.outputChannel, respondToUI) if !isempty(result.content) && result.content[1] isa textContent - agent.eventSink(result.content[1].text; publish=false, log=true) + agent.eventSink(result.content[1].text; publish=true) end processingTask = nothing # reset result = nothing # reset @@ -325,7 +325,7 @@ function _agentLoop(agent::yiemAgent) else # spawn new _processMessage() if it is not already running. if processingTask === nothing - agent.eventSink("_agentLoop 5 agent._state.messages length $(length(agent._state.messages))"; publish=false, log=true) + agent.eventSink("_agentLoop 5 agent._state.messages length $(length(agent._state.messages))"; publish=true) # discard all messages in followUpChannel while isready(agent.followUpChannel) _ = take!(agent.followUpChannel) @@ -356,7 +356,7 @@ function _agentLoop(agent::yiemAgent) showerror(io, e, bt) println(io) end - eventSink(errMsg; publish=false, log=true) + eventSink(errMsg; publish=true) # On any error, send error response and exit the loop @error "Agent loop failed" error=e @@ -404,7 +404,7 @@ function _processMessage( afterToolCall::Union{Function, Nothing}, parallelToolExecute::Bool, )::Nothing - eventSink("_processMessage 1 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 1 _state.messages length $(length(agentMsgHistory))"; publish=true) # loop until llmCall() response didn't use tool calls final_response = nothing @@ -425,43 +425,43 @@ function _processMessage( try # Drain inputChannel and convert OpenAI-format messages to userMessage type while isready(inputChannel) - eventSink("_processMessage 2 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 2 _state.messages length $(length(agentMsgHistory))"; publish=true) newUserMsg_openai = take!(inputChannel) - eventSink("_processMessage 3 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 3 _state.messages length $(length(agentMsgHistory))"; publish=true) if newUserMsg_openai === :shutdown - eventSink("_processMessage 4 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 4 _state.messages length $(length(agentMsgHistory))"; publish=true) # Re-emit shutdown signal for the loop to handle put!(inputChannel, :shutdown) break end - eventSink("_processMessage 5 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 5 _state.messages length $(length(agentMsgHistory))"; publish=true) newUserMsg = OpenAiToUserMessage(newUserMsg_openai) push!(agentMsgHistory, newUserMsg) - eventSink("_processMessage 6 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 6 _state.messages length $(length(agentMsgHistory))"; publish=true) end - eventSink("_processMessage 7 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 7 _state.messages length $(length(agentMsgHistory))"; publish=true) # call prepareContext() state = agentState(systemPrompt, nothing, tools, agentMsgHistory) - eventSink("_processMessage 8 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 8 _state.messages length $(length(agentMsgHistory))"; publish=true) preparedContext = prepareContext(state, eventSink) - eventSink("_processMessage 9 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 9 _state.messages length $(length(agentMsgHistory))"; publish=true) # Call formatMessagesForLLM() to format for LLM formattedMessages = formatMessagesForLLM(preparedContext, eventSink) - # eventSink("_processMessage 10 formattedMessages $formattedMessages"; publish=false, log=true) + # eventSink("_processMessage 10 formattedMessages $formattedMessages"; publish=true) """ response example response = JSON.Object{String, Any}("finish_reason" => "tool_calls", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "", "reasoning_content" => "Here's a thinking process:\n\n1. **Identify User Request**: The user is asking for the weather in Bangkok.\n2. **Locate Relevant Tool**: I have a `getWeather` function available.\n3. **Check Function Parameters**:\n - `city` (required): City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\n - `units` (optional, default \"celsius\"): Temperature scale (\"celsius\" or \"fahrenheit\")\n4. **Prepare Parameters**:\n - `city`: \"Bangkok, Thailand\" (adding country for clarity, though just \"Bangkok\" might work, following the example format is safer)\n - `units`: Not specified, so I'll use the default (\"celsius\")\n5. **Execute Tool Call**: Call `getWeather` with `city: \"Bangkok, Thailand\"`\n6. **Anticipate Response**: The function will return current weather and forecast data for Bangkok. I'll then format it nicely for the user.\n - *Self-Correction/Verification during thought*: The prompt says \"city: City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\". I'll use \"Bangkok, Thailand\". The `units` parameter is optional, so I'll omit it to use the default.\n - Proceed with tool call.✅\n", "tool_calls" => Any[JSON.Object{String, Any}("type" => "function", "function" => JSON.Object{String, Any}("name" => "getWeather", "arguments" => "{\"city\":\"Bangkok, Thailand\"}"), "id" => "6fOilR5QPcdppbAAHluhkRyUDu3oWMAL")])) """ response = llmCall(formattedMessages, eventSink=eventSink) - # eventSink("LLM response " * string(response); publish=false, log=true) + # eventSink("LLM response " * string(response); publish=true) - eventSink("_processMessage 11 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 11 _state.messages length $(length(agentMsgHistory))"; publish=true) # Extract tool calls from LLM response content blocks hasToolCalls, toolCallList, assistant_msg = _extractToolCalls(response) - eventSink("hasToolCalls: $hasToolCalls\ntoolCallList: $toolCallList _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) - eventSink("assistant_msg " * string(assistant_msg); publish=false, log=true) - eventSink("_processMessage 11-1"; publish=false, log=true) + eventSink("hasToolCalls: $hasToolCalls\ntoolCallList: $toolCallList _state.messages length $(length(agentMsgHistory))"; publish=true) + eventSink("assistant_msg " * string(assistant_msg); publish=true) + eventSink("_processMessage 11-1"; publish=true) # Add assistant message (tool calls or text) to history for next LLM turn push!(agentMsgHistory, assistant_msg) @@ -476,21 +476,21 @@ function _processMessage( ) signal = abortSignal(false) - eventSink("_processMessage 12 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 12 _state.messages length $(length(agentMsgHistory))"; publish=true) # call executeToolCalls() toolResultBatch = executeToolCalls(preparedContext, assistant_msg, toolCallList, config, signal, eventSink) - eventSink("_processMessage 13 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 13 _state.messages length $(length(agentMsgHistory))"; publish=true) # save toolResults to messages for toolResult in toolResultBatch.messages - # eventSink("toolResult " * string(toolResult); publish=false, log=true) + # eventSink("toolResult " * string(toolResult); publish=true) push!(agentMsgHistory, toolResult) end - eventSink("_processMessage 14 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 14 _state.messages length $(length(agentMsgHistory))"; publish=true) if toolResultBatch.terminate - eventSink("_processMessage 15 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 15 _state.messages length $(length(agentMsgHistory))"; publish=true) # If toolResultBatch requested termination, build a final response final_content = [textContent("Tool execution completed.")] for toolResult in toolResultBatch.messages @@ -504,7 +504,7 @@ function _processMessage( end end end - eventSink("_processMessage 16 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 16 _state.messages length $(length(agentMsgHistory))"; publish=true) final_response = assistantMessage( role="assistant", content=final_content, @@ -523,7 +523,7 @@ function _processMessage( break end else - eventSink("_processMessage 17 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 17 _state.messages length $(length(agentMsgHistory))"; publish=true) # LLM did not use tool calls — break end @@ -534,12 +534,12 @@ function _processMessage( println(io) end - eventSink(err_msg; publish=false, log=true) + eventSink(err_msg; publish=true) end end - eventSink("_processMessage 18 _state.messages length $(length(agentMsgHistory))"; publish=false, log=true) + eventSink("_processMessage 18 _state.messages length $(length(agentMsgHistory))"; publish=true) return nothing end @@ -979,47 +979,47 @@ function prepareToolCall( signal::abortSignal, eventSink )::Union{preparedToolCall,immediateOutcome} - eventSink("prepareToolCall 1"; publish=false, log=true) + eventSink("prepareToolCall 1"; publish=true) # pick a called tool from tool store tool = get(context.tools, toolCall.name, nothing) if tool === nothing - eventSink("prepareToolCall 2"; publish=false, log=true) + eventSink("prepareToolCall 2"; publish=true) return immediateOutcome(createErrorToolResult("Tool $toolCall.name not found"), true) end try - eventSink("prepareToolCall 3"; publish=false, log=true) + eventSink("prepareToolCall 3"; publish=true) # 1. prepare arguments (tool-specific transform) prepared = prepareToolCallArguments(tool, toolCall) - eventSink("prepared " * string(prepared.arguments); publish=false, log=true) - eventSink("prepareToolCall 4"; publish=false, log=true) + eventSink("prepared " * string(prepared.arguments); publish=true) + eventSink("prepareToolCall 4"; publish=true) validatedArgs = validateToolArguments(tool, prepared) - eventSink("validatedArgs " * string(validatedArgs); publish=false, log=true) - eventSink("prepareToolCall 5"; publish=false, log=true) + eventSink("validatedArgs " * string(validatedArgs); publish=true) + eventSink("prepareToolCall 5"; publish=true) # 2. beforeToolCall hook — can block if config.beforeToolCall !== nothing - eventSink("prepareToolCall 6"; publish=false, log=true) + eventSink("prepareToolCall 6"; publish=true) before = config.beforeToolCall( beforeToolCallContext(assistantMsg, toolCall, validatedArgs, context), signal, eventSink ) - eventSink("prepareToolCall 7"; publish=false, log=true) + eventSink("prepareToolCall 7"; publish=true) if signal.aborted - eventSink("prepareToolCall 8"; publish=false, log=true) + eventSink("prepareToolCall 8"; publish=true) return immediateOutcome(createErrorToolResult("Operation aborted"), true) end if before !== nothing && before.block - eventSink("prepareToolCall 9"; publish=false, log=true) + eventSink("prepareToolCall 9"; publish=true) return immediateOutcome( createErrorToolResult(get(before, :reason, "Tool execution was blocked")), true) end end - eventSink("prepareToolCall 10"; publish=false, log=true) + eventSink("prepareToolCall 10"; publish=true) return preparedToolCall(tool, toolCall, validatedArgs) catch e bt = catch_backtrace() @@ -1028,7 +1028,7 @@ function prepareToolCall( println(io) end - eventSink(errMsg; publish=false, log=true) + eventSink(errMsg; publish=true) return immediateOutcome(createErrorToolResult(sprint(showerror, e)), true) end @@ -1080,16 +1080,11 @@ function executePreparedToolCall( signal::Union{Nothing,abortSignal}, eventSink, )::executedOutcome - eventSink("executePreparedToolCall 1"; publish=false, log=true) + eventSink("executePreparedToolCall 1"; publish=true) try - result = prep.tool.execute(prep.toolCall.id, prep.args, signal, eventSink) - - - eventSink("executePreparedToolCall 2"; publish=false, log=true) - eventSink("tool execute raw result: " * sprint(show, result); publish=false, log=true) - eventSink("executePreparedToolCall 3"; publish=false, log=true) + # eventSink("tool execute raw result: " * sprint(show, result); publish=true) return executedOutcome(result, false) catch e bt = catch_backtrace() @@ -1097,7 +1092,7 @@ function executePreparedToolCall( showerror(io, e, bt) println(io) end - eventSink(errMsg; publish=false, log=true) + eventSink(errMsg; publish=true) return executedOutcome(createErrorToolResult(sprint(showerror, e)), true) end @@ -1170,10 +1165,10 @@ function finalizeExecutedToolCall( signal::Union{Nothing,abortSignal}, eventSink )::finalizedOutcome - eventSink("finalizeExecutedToolCall 1"; publish=false, log=true) + eventSink("finalizeExecutedToolCall 1"; publish=true) result = executed.result isError = executed.isError - eventSink("finalizeExecutedToolCall result before afterToolCall\n$result"; publish=false, log=true) + # eventSink("finalizeExecutedToolCall result before afterToolCall\n$result"; publish=true) if config.afterToolCall !== nothing try after = config.afterToolCall( @@ -1181,7 +1176,7 @@ function finalizeExecutedToolCall( signal, eventSink ) - eventSink("finalizeExecutedToolCall 3"; publish=false, log=true) + eventSink("finalizeExecutedToolCall 3"; publish=true) if after !== nothing result = after isError = false @@ -1192,13 +1187,13 @@ function finalizeExecutedToolCall( showerror(io, e, bt) println(io) end - eventSink(errMsg; publish=false, log=true) + eventSink(errMsg; publish=true) result = createErrorToolResult(sprint(showerror, e)) isError = true end end - eventSink("finalizeExecutedToolCall 4"; publish=false, log=true) + eventSink("finalizeExecutedToolCall 4"; publish=true) return finalizedOutcome(prep.toolCall, result, isError) end @@ -1264,39 +1259,39 @@ function executeToolCallsSequential( signal::abortSignal, eventSink, )::agentToolCallBatch - eventSink("executeToolCallsSequential 1"; publish=false, log=true) + eventSink("executeToolCallsSequential 1"; publish=true) finalizedCalls = finalizedOutcome[] messages = toolResultMessage[] for tc in toolCalls - eventSink("start tool execute $(tc.id), $(tc.name), $(tc.arguments)"; publish=false, log=true) + eventSink("start tool execute $(tc.id), $(tc.name), $(tc.arguments)"; publish=true) prep = prepareToolCall(context, assistantMsg, tc, config, signal, eventSink) - eventSink("executeToolCallsSequential " * string(prep.args); publish=false, log=true) + eventSink("executeToolCallsSequential " * string(prep.args); publish=true) if prep isa immediateOutcome - eventSink("executeToolCallsSequential 2-1"; publish=false, log=true) + eventSink("executeToolCallsSequential 2-1"; publish=true) finalized = finalizedOutcome(tc, prep.result, prep.isError) - eventSink("executeToolCallsSequential 2-2"; publish=false, log=true) + eventSink("executeToolCallsSequential 2-2"; publish=true) else - eventSink("executeToolCallsSequential 3"; publish=false, log=true) + eventSink("executeToolCallsSequential 3"; publish=true) executed = executePreparedToolCall(prep, signal, eventSink) - eventSink("executeToolCallsSequential 3-1"; publish=false, log=true) + eventSink("executeToolCallsSequential 3-1"; publish=true) finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal, eventSink) - eventSink("executeToolCallsSequential 3-2"; publish=false, log=true) + eventSink("executeToolCallsSequential 3-2"; publish=true) end - eventSink("executeToolCallsSequential 4"; publish=false, log=true) + eventSink("executeToolCallsSequential 4"; publish=true) # eventSink("finalized $(finalized.toolCall.id), $(finalized.toolCall.name), - # $(finalized.result), $(finalized.isError)"; publish=false, log=true) + # $(finalized.result), $(finalized.isError)"; publish=true) push!(messages, createToolResultMessage(finalized)) push!(finalizedCalls, finalized) - eventSink("executeToolCallsSequential 5"; publish=false, log=true) + eventSink("executeToolCallsSequential 5"; publish=true) if signal !== nothing && signal.aborted break end end - eventSink("executeToolCallsSequential 6"; publish=false, log=true) + eventSink("executeToolCallsSequential 6"; publish=true) return agentToolCallBatch(messages, shouldTerminate(finalizedCalls)) end @@ -1369,7 +1364,7 @@ function executeToolCallsParallel( entries = Union{finalizedOutcome,Task}[] for tc in toolCalls - eventSink(toolExecStartEvent(tc.id, tc.name, tc.arguments); publish=false, log=true) + eventSink(toolExecStartEvent(tc.id, tc.name, tc.arguments); publish=true) prep = prepareToolCall(context, assistantMsg, tc, config, signal, eventSink) @@ -1468,7 +1463,7 @@ function executeToolCalls( eventSink, )::agentToolCallBatch - eventSink("_executeToolCalls 1"; publish=false, log=true) + eventSink("_executeToolCalls 1"; publish=true) hasSequential = false for tc in toolCalls t = get(context.tools, tc.name, nothing) @@ -1477,13 +1472,13 @@ function executeToolCalls( break end end - eventSink("_executeToolCalls 2"; publish=false, log=true) + eventSink("_executeToolCalls 2"; publish=true) if config.toolExecution == "sequential" || hasSequential - eventSink("_executeToolCalls 3"; publish=false, log=true) + eventSink("_executeToolCalls 3"; publish=true) return executeToolCallsSequential(context, assistantMsg, toolCalls, config, signal, eventSink) else - eventSink("_executeToolCalls 4"; publish=false, log=true) + eventSink("_executeToolCalls 4"; publish=true) return executeToolCallsParallel(context, assistantMsg, toolCalls, config, signal, eventSink) end