update
This commit is contained in:
@@ -1,77 +1,11 @@
|
||||
i am not sure that's the case. see my NATS message log:
|
||||
<NATS debug message>
|
||||
┌ 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"
|
||||
</NATS debug message>
|
||||
|
||||
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?
|
||||
|
||||
|
||||
+38
-59
@@ -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,15 +253,12 @@ 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
|
||||
# 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(
|
||||
@@ -262,44 +274,11 @@ function _agentLoop(agent::yiemAgent)
|
||||
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)
|
||||
put!(processMessageInputCh, newUserMsg)
|
||||
newUserMsg = nothing # reset
|
||||
end
|
||||
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
|
||||
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
|
||||
@error "Agent loop failed" error=e
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user