diff --git a/src/agentCore.jl b/src/agentCore.jl index 2355c73..4cf2f7b 100644 --- a/src/agentCore.jl +++ b/src/agentCore.jl @@ -34,45 +34,109 @@ julia> # Called automatically by yiemAgent constructor """ function _agent_loop(agent::yiemAgent) try + processing_task = nothing + + """ cases: + 1) agent -> idle, user msg -> nothing + typeof(processing_task) == Nothing + agent._state.activeRun -> false + agent.inputChannel -> nothing + agent.followUpChannel -> nothing + + 2) agent -> idle, user msg -> new msg + typeof(processing_task) == Nothing + agent._state.activeRun -> false + agent.inputChannel -> new msg + agent.followUpChannel -> nothing + + 3) agent -> running, user msg -> nothing + typeof(processing_task) == Task, istaskdone(processing_task) -> false + agent._state.activeRun -> true + agent.inputChannel -> nothing + agent.followUpChannel -> nothing + + 4) agent -> running, user msg -> new msg + typeof(processing_task) == Task, istaskdone(processing_task) -> 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(processing_task) == Task, istaskdone(processing_task) -> false + agent._state.activeRun -> true + agent.inputChannel -> nothing + agent.followUpChannel -> new msg + + 6) agent -> idle, user msg -> nothing + typeof(processing_task) == Task, istaskdone(processing_task) -> true + agent._state.activeRun -> false + agent.inputChannel -> nothing + agent.followUpChannel -> nothing + """ + + while true + result = nothing msg = nothing while msg === nothing - if isready(agent.inputChannel) && agent._state.activeRun == false - # allow _process_message() to run - yield() - elseif isready(agent.inputChannel) && agent._state.activeRun == true - msg = fetch!(agent.inputChannel) - # Check for shutdown signal - if msg === :shutdown - msg = take!(agent.inputChannel) - end + if isready(agent.inputChannel) + + # message will be taken then process in _process_message() + msg = fetch!(agent.inputChannel) else - error("undefined condition: isready(inputChannel)=$(isready(agent.inputChannel)), activeRun=$(agent._state.activeRun), msg=$msg") + yield() end end # Check for shutdown signal if msg === :shutdown + # Drain all remaining messages in the input channel + if isready(agent.inputChannel) + while isready(agent.inputChannel) + _ = take!(agent.inputChannel) + end + end + if isready(agent.followUpChannel) + while isready(agent.followUpChannel) + _ = take!(agent.followUpChannel) + end + end + #TODO make sure every running tools ended properly break end # make active - agent._state.activeRun = true - # Dispatch message through the processing pipeline - result = _process_message(agent, msg) - - # check followUp message. if there are, add them all to agent.inputChannel - hasMore = isready(agent.followUpChannel) - while isready(agent.followUpChannel) - followMsg = take!(agent.followUpChannel) - put!(agent.inputChannel, followMsg) - hasMore = true + if agent._state.activeRun == false + # Dispatch message through the processing pipeline + processing_task = @spawn _process_message(agent, msg) + agent._state.activeRun = true end - if !isready(agent.inputChannel) && !hasMore - # no more messages queued — safe to send response + # during agent runs, check followUp message after _process_message() is done + if typeof(processing_task) == Task && istaskdone(processing_task) == false + # if followUp message available, add them all to agent.inputChannel + if isready(agent.followUpChannel) + while isready(agent.followUpChannel) + followMsg = take!(agent.followUpChannel) + put!(agent.inputChannel, followMsg) + end + end + + continue # continue to process user message in the next loop + elseif typeof(processing_task) == Task && istaskdone(processing_task) == true + # 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(processing_task) put!(agent.outputChannel, result) + agent._state.activeRun = false + processing_task = nothing end end catch e diff --git a/src/type.jl b/src/type.jl index a2d8bc0..dfd5b30 100644 --- a/src/type.jl +++ b/src/type.jl @@ -289,6 +289,7 @@ function agentState( deepcopy(tools), deepcopy(messages), Vector{String}(), + false, nothing, ) end