From fd616409dddd4a61eba90557eb860d8455d99a7f Mon Sep 17 00:00:00 2001 From: narawat Date: Sat, 15 Aug 2026 20:03:01 +0700 Subject: [PATCH] update --- etc.jl | 84 ++++--------------------------- src/agentCore.jl | 127 ++++++++++++++++++++--------------------------- src/type.jl | 2 - 3 files changed, 62 insertions(+), 151 deletions(-) diff --git a/etc.jl b/etc.jl index 7d92ff2..c6c563a 100644 --- a/etc.jl +++ b/etc.jl @@ -1,77 +1,11 @@ -i am not sure that's the case. see my NATS message log: - -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "new user msg" -┌ Info: debug -└ payload = "_process_message 3" -┌ Info: debug -└ payload = "_process_message 5" -┌ Info: debug -└ payload = "_process_message 6" -┌ Info: debug -└ payload = "_process_message 7" - - -my NATS receiver report the following for a long time -┌ Info: debug -└ payload = "new user msg" - -untill I Ctrl + d so shutdown the process then i got the following report -┌ Info: debug -└ payload = "_process_message 3" -┌ Info: debug -└ payload = "_process_message 5" -┌ Info: debug -└ payload = "_process_message 6" -┌ Info: debug -└ payload = "_process_message 7" - - - - - - - - - -my point is if _process_message() actually run then this code in _process_message() -"raw_msg = take!(agent.inputChannel)" -should take the new msg message out of agent.inputChannel and there should be only one debug message showing -┌ Info: debug -└ payload = "new user msg" - -before reaching error("debug marker") - - - - - - - - - - - - - - - - - - - - +_processMessage 10 +JSON.Object{String, Any}("finish_reason" => "stop", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "The weather in Bangkok is currently Sunny with a temperature of 22°C.", "reasoning_content" => "The user is asking for the weather in Bangkok.\nI have already retrieved the weather information in the previous turn and provided it to the user.\nThe user's current input is \"What's the weather in Bangkok?\", which is the same question as before.\nI should provide the same answer again.\nNo new tool calls are needed.\nI will simply state the weather information retrieved previously.\nWeather in Bangkok: Sunny, 22°C.\nI will output the answer directly.\n")) +_processMessage 11 +hasToolCalls: false +toolCallList: YiemAgent.type.agentToolCall[] +_processMessage 17 +_processMessage 18 +--- +here is my log. from the log, it seems like _processMessage() is finished but somehow the log didn't show _agentLoop 5 message. _processMessage() may not exit properly but why? diff --git a/src/agentCore.jl b/src/agentCore.jl index c45945e..c1a4168 100644 --- a/src/agentCore.jl +++ b/src/agentCore.jl @@ -169,62 +169,77 @@ julia> # Called automatically by yiemAgent constructor function _agentLoop(agent::yiemAgent) processMessageInputCh = Channel(32) try + newUserMsg = nothing processingTask = nothing + result = nothing """ cases: 1) agent -> idle, user msg -> nothing typeof(processingTask) == Nothing - agent._state.activeRun -> false agent.inputChannel -> nothing agent.followUpChannel -> nothing 2) agent -> idle, user msg -> new msg typeof(processingTask) == Nothing - agent._state.activeRun -> false agent.inputChannel -> new msg agent.followUpChannel -> nothing 3) agent -> running, user msg -> nothing typeof(processingTask) == Task, istaskdone(processingTask) -> false - agent._state.activeRun -> true agent.inputChannel -> nothing agent.followUpChannel -> nothing 4) agent -> running, user msg -> new msg typeof(processingTask) == Task, istaskdone(processingTask) -> false - agent._state.activeRun -> true agent.inputChannel -> new msg agent.followUpChannel -> nothing 5) agent -> running, user msg -> nothing, user msg follow up -> new msg typeof(processingTask) == Task, istaskdone(processingTask) -> false - agent._state.activeRun -> true agent.inputChannel -> nothing agent.followUpChannel -> new msg 6) agent -> idle, user msg -> nothing typeof(processingTask) == Task, istaskdone(processingTask) -> true - agent._state.activeRun -> false agent.inputChannel -> nothing agent.followUpChannel -> nothing """ while true - result = nothing - msg = nothing - while msg === nothing + while newUserMsg === nothing if isready(agent.inputChannel) - - # message will be taken in _processMessage() - msg = take!(agent.inputChannel) + agent.agentEventSink("_agentLoop 1") + # agent process new user msg immediately after the current tool call finished. + newUserMsg = take!(agent.inputChannel) agent.agentEventSink("new user msg") else + # check followUp message after _processMessage() is done + if typeof(processingTask) == Task && istaskdone(processingTask) == true + agent.agentEventSink("_agentLoop 2") + # if agent runs is done but followUpChannel has messages, + # put new message in inputChannel instead + if isready(agent.followUpChannel) + agent.agentEventSink("_agentLoop 3") + while isready(agent.followUpChannel) + followUpMsg = take!(agent.followUpChannel) + put!(agent.inputChannel, followUpMsg) + end + else # _processMessage() done and no followUp message. + agent.agentEventSink("_agentLoop 4") + result = fetch(processingTask) + put!(agent.outputChannel, result) + agent.agentEventSink(result.content[1].text) + processingTask = nothing # reset + newUserMsg = nothing # reset + result = nothing # reset + end + end yield() end end # Check for shutdown signal - if msg === :shutdown + if newUserMsg === :shutdown # Drain all remaining messages in the input channel if isready(agent.inputChannel) while isready(agent.inputChannel) @@ -238,67 +253,31 @@ function _agentLoop(agent::yiemAgent) end #TODO make sure every running tools ended properly + + newUserMsg = nothing # reset break else - agent.agentEventSink("_agentLoop push 1") - put!(processMessageInputCh, msg) - agent.agentEventSink("_agentLoop push 2") - end - - # start _processMessage loop - if agent._state.activeRun == false - agent.agentEventSink("_agentLoop 2") - # Dispatch message through the processing pipeline - processingTask = @spawn _processMessage( - processMessageInputCh, - agent.agentEventSink, - agent._state.messages, - agent._state.systemPrompt, - agent._state.tools, - agent.prepareContext, - agent.formatMsgForLLM, - agent.llmCall, - agent.beforeToolCall, - agent.afterToolCall, - agent.parallelToolExecute, - ) - agent._state.activeRun = true - agent.agentEventSink("_agentLoop 3") - end - - # during agent runs, check followUp message after _processMessage() is done - if typeof(processingTask) == Task && istaskdone(processingTask) == false - agent.agentEventSink("_agentLoop 4") - # if followUp message available, add them all to agent.inputChannel - if isready(agent.followUpChannel) - agent.agentEventSink("_agentLoop 4-1") - while isready(agent.followUpChannel) - agent.agentEventSink("_agentLoop 4-2") - followMsg = take!(agent.followUpChannel) - put!(agent.inputChannel, followMsg) - end + # spawn new _processMessage() if it is not already running. + if processingTask === nothing + agent.agentEventSink("_agentLoop 2") + # Dispatch message through the processing pipeline + processingTask = @spawn _processMessage( + processMessageInputCh, + agent.agentEventSink, + agent._state.messages, + agent._state.systemPrompt, + agent._state.tools, + agent.prepareContext, + agent.formatMsgForLLM, + agent.llmCall, + agent.beforeToolCall, + agent.afterToolCall, + agent.parallelToolExecute, + ) end - agent.agentEventSink("_agentLoop 4-3") - continue # continue to process user message in the next loop - - elseif typeof(processingTask) == Task && istaskdone(processingTask) == true - agent.agentEventSink("_agentLoop 5") - # if agent runs is done but followUpChannel has messages, discard all message in it. - # when agent work is done it should not accept follow up msg. - # user should put new message in inputChannel instead - if isready(agent.followUpChannel) - while isready(agent.followUpChannel) - _ = take!(agent.followUpChannel) - end - end - result = fetch(processingTask) - put!(agent.outputChannel, result) - agent._state.activeRun = false # reset - processingTask = nothing # reset + put!(processMessageInputCh, newUserMsg) + newUserMsg = nothing # reset end - agent.agentEventSink("_agentLoop 6") - agent.agentEventSink(string(typeof(processingTask))) - agent.agentEventSink("_agentLoop 7") end catch e # On any error, send error response and exit the loop @@ -368,17 +347,17 @@ function _processMessage( # Drain inputChannel and convert OpenAI-format messages to userMessage type while isready(inputChannel) agentEventSink("_processMessage 2") - raw_msg = take!(inputChannel) + newUserMsg_openai = take!(inputChannel) agentEventSink("_processMessage 3") - if raw_msg === :shutdown + if newUserMsg_openai === :shutdown agentEventSink("_processMessage 4") # Re-emit shutdown signal for the loop to handle put!(inputChannel, :shutdown) break end agentEventSink("_processMessage 5") - user_msg = OpenAiToUserMessage(raw_msg) - push!(agentMsgHistory, user_msg) + newUserMsg = OpenAiToUserMessage(newUserMsg_openai) + push!(agentMsgHistory, newUserMsg) agentEventSink("_processMessage 6") end agentEventSink("_processMessage 7") diff --git a/src/type.jl b/src/type.jl index 6eab694..54ff272 100644 --- a/src/type.jl +++ b/src/type.jl @@ -317,7 +317,6 @@ mutable struct agentState # Mutable runtime state of an agen messages::Vector{agentMessage} pendingToolCalls::Vector{String} # Tool call IDs waiting for results - activeRun::Bool # is agent processing user message? errorMessage::Union{String, Nothing} # Last error message end @@ -354,7 +353,6 @@ function agentState( deepcopy(tools), deepcopy(messages), Vector{String}(), - false, nothing, ) end