This commit is contained in:
2026-08-12 20:10:51 +07:00
parent 4e592173a6
commit 77adeb3a6b
4 changed files with 119 additions and 60 deletions
+4 -4
View File
@@ -152,7 +152,7 @@ result = getTime_tool.execute("call-1", Dict("city" => "Tokyo"), sig, op)
**Via agent loop (production):** **Via agent loop (production):**
``` ```
user message → runAgent(agent, Dict("role"=>"user", "content"=>...)) user message → runAgent(agent, Dict("role"=>"user", "content"=>...))
→ _agent_loop detects message → @spawn _process_message(agent) → _agentLoop detects message → @spawn _process_message(agent)
→ prepareContext → formatMsgForLLM → llmCall → prepareContext → formatMsgForLLM → llmCall
→ LLM returns tool_calls → LLM returns tool_calls
→ executeToolCalls(context, response, tool_call_list, config, signal, emit) → executeToolCalls(context, response, tool_call_list, config, signal, emit)
@@ -376,7 +376,7 @@ This ensures that `yiemAgent` instances with different `tool_store` references o
**Source:** `agentCore.jl:35-145` **Source:** `agentCore.jl:35-145`
The `_agent_loop()` function runs as a background `@spawn` task, created when `yiemAgent` is constructed. The `_agentLoop()` function runs as a background `@spawn` task, created when `yiemAgent` is constructed.
### Channel Architecture ### Channel Architecture
@@ -404,7 +404,7 @@ The loop tracks 6 states (documented at `agentCore.jl:39-75`):
### Loop Logic (simplified) ### Loop Logic (simplified)
```julia ```julia
function _agent_loop(agent::yiemAgent) function _agentLoop(agent::yiemAgent)
while true while true
# 1. Wait for message from inputChannel (blocking poll) # 1. Wait for message from inputChannel (blocking poll)
msg = fetch!(agent.inputChannel) # agentCore.jl:84 msg = fetch!(agent.inputChannel) # agentCore.jl:84
@@ -1332,7 +1332,7 @@ USER SENDS MESSAGE
LOOP ITERATION 1 — LLM DECIDES TO USE A TOOL LOOP ITERATION 1 — LLM DECIDES TO USE A TOOL
└─> _agent_loop: detects msg in inputChannel └─> _agentLoop: detects msg in inputChannel
└─> Threads.@spawn _process_message(agent) └─> Threads.@spawn _process_message(agent)
── _process_message ────────────────────────────────────────────── ── _process_message ──────────────────────────────────────────────
+91 -54
View File
@@ -1,6 +1,6 @@
module agentCore module agentCore
export yiemAgent, _agent_loop, OpenAiToUserMessage export yiemAgent, _agentLoop, OpenAiToUserMessage
using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization, using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization,
DataFrames, Base.Threads DataFrames, Base.Threads
@@ -27,10 +27,10 @@ mutable struct yiemAgent <: agent # High-level agent wrapper
# and all followUp messages. # and all followUp messages.
outputChannel::Channel outputChannel::Channel
_agent_loop::Union{Task, Nothing} # agent loop running in the background _agentLoop::Union{Task, Nothing} # agent loop running in the background
# Preprocess/transform messages and context (modify, filter, prune, inject context from memory, # Preprocess/transform messages and context (modify, filter, prune, inject context from memory,
# reorder, ...) for a single LLM call in _process_message()'s loop. # reorder, ...) for a single LLM call in _processMessage()'s loop.
# returns new Vector{agentMessage} # returns new Vector{agentMessage}
prepareContext::Union{Function, Nothing} prepareContext::Union{Function, Nothing}
@@ -131,7 +131,7 @@ function yiemAgent(
) )
# Spawn the background loop and attach it # Spawn the background loop and attach it
agent._agent_loop = @spawn _agent_loop(agent) agent._agentLoop = @spawn _agentLoop(agent)
return agent return agent
end end
@@ -141,7 +141,7 @@ end
Private agent loop. Runs in a background `@spawn` task. Private agent loop. Runs in a background `@spawn` task.
Waits on `inputChannel` and `followUpChannel`, processing whichever has a message first. Waits on `inputChannel` and `followUpChannel`, processing whichever has a message first.
On each iteration, dispatches the message through `_process_message` and sends the result On each iteration, dispatches the message through `_processMessage` and sends the result
to `outputChannel`. Exits on `:shutdown` signal. to `outputChannel`. Exits on `:shutdown` signal.
# Arguments # Arguments
@@ -160,7 +160,8 @@ to `outputChannel`. Exits on `:shutdown` signal.
julia> # Called automatically by yiemAgent constructor julia> # Called automatically by yiemAgent constructor
``` ```
""" """
function _agent_loop(agent::yiemAgent) function _agentLoop(agent::yiemAgent)
processMessageInputCh = Channel(32)
try try
processingTask = nothing processingTask = nothing
@@ -208,8 +209,8 @@ function _agent_loop(agent::yiemAgent)
while msg === nothing while msg === nothing
if isready(agent.inputChannel) if isready(agent.inputChannel)
# message will be taken in _process_message() # message will be taken in _processMessage()
msg = fetch(agent.inputChannel) msg = take!(agent.inputChannel)
agent.agentEventSink("new user msg") agent.agentEventSink("new user msg")
else else
yield() yield()
@@ -232,29 +233,50 @@ function _agent_loop(agent::yiemAgent)
#TODO make sure every running tools ended properly #TODO make sure every running tools ended properly
break break
else
agent.agentEventSink("_agentLoop push 1")
put!(processMessageInputCh, msg) #WORKING
agent.agentEventSink("_agentLoop push 2")
end end
# start _process_message loop # start _processMessage loop
if agent._state.activeRun == false if agent._state.activeRun == false
agent.agentEventSink("_agent_loop 2") agent.agentEventSink("_agentLoop 2")
# Dispatch message through the processing pipeline # Dispatch message through the processing pipeline
processingTask = Threads.@spawn _process_message(agent) 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._state.activeRun = true
agent.agentEventSink("_agent_loop 3") agent.agentEventSink("_agentLoop 3")
end end
# during agent runs, check followUp message after _process_message() is done # during agent runs, check followUp message after _processMessage() is done
if typeof(processingTask) == Task && istaskdone(processingTask) == false if typeof(processingTask) == Task && istaskdone(processingTask) == false
agent.agentEventSink("_agentLoop 4")
# if followUp message available, add them all to agent.inputChannel # if followUp message available, add them all to agent.inputChannel
if isready(agent.followUpChannel) if isready(agent.followUpChannel)
agent.agentEventSink("_agentLoop 4-1")
while isready(agent.followUpChannel) while isready(agent.followUpChannel)
agent.agentEventSink("_agentLoop 4-2")
followMsg = take!(agent.followUpChannel) followMsg = take!(agent.followUpChannel)
put!(agent.inputChannel, followMsg) put!(agent.inputChannel, followMsg)
end end
end end
agent.agentEventSink("_agentLoop 4-3")
continue # continue to process user message in the next loop continue # continue to process user message in the next loop
elseif typeof(processingTask) == Task && istaskdone(processingTask) == true 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. # 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. # when agent work is done it should not accept follow up msg.
# user should put new message in inputChannel instead # user should put new message in inputChannel instead
@@ -268,6 +290,7 @@ function _agent_loop(agent::yiemAgent)
agent._state.activeRun = false # reset agent._state.activeRun = false # reset
processingTask = nothing # reset processingTask = nothing # reset
end end
agent.agentEventSink("_agentLoop 6")
end end
catch e catch e
# On any error, send error response and exit the loop # On any error, send error response and exit the loop
@@ -303,13 +326,25 @@ should be implemented. Currently a placeholder that echoes back the received mes
julia> # Currently returns a placeholder echo response julia> # Currently returns a placeholder echo response
``` ```
""" """
function _process_message(agent::yiemAgent)::assistantMessage function _processMessage(
agent.agentEventSink("_process_message 1") inputChannel::Channel,
agentEventSink,
messages::Vector{agentMessage},
systemPrompt::String,
tools::OrderedDict{String, agentTool},
prepareContext::Function,
formatMessagesForLLM::Function,
llmCall,
beforeToolCall::Union{Function, Nothing},
afterToolCall::Union{Function, Nothing},
parallelToolExecute::Bool,
)::assistantMessage
agentEventSink("_processMessage 1")
# loop until llmCall() response didn't use tool calls # loop until llmCall() response didn't use tool calls
final_response = nothing final_response = nothing
while true while true
""" example message in agent.inputChannel """ example message in inputChannel
Dict( Dict(
"role" => "user", "role" => "user",
"content" => [ "content" => [
@@ -323,42 +358,44 @@ function _process_message(agent::yiemAgent)::assistantMessage
""" """
# Drain inputChannel and convert OpenAI-format messages to userMessage type # Drain inputChannel and convert OpenAI-format messages to userMessage type
while isready(agent.inputChannel) while isready(inputChannel)
agent.agentEventSink("_process_message 2") agentEventSink("_processMessage 2")
raw_msg = take!(agent.inputChannel) raw_msg = take!(inputChannel)
agent.agentEventSink("_process_message 3") agentEventSink("_processMessage 3")
if raw_msg === :shutdown if raw_msg === :shutdown
agent.agentEventSink("_process_message 4") agentEventSink("_processMessage 4")
# Re-emit shutdown signal for the loop to handle # Re-emit shutdown signal for the loop to handle
put!(agent.inputChannel, :shutdown) put!(inputChannel, :shutdown)
break break
end end
agent.agentEventSink("_process_message 5") agentEventSink("_processMessage 5")
user_msg = OpenAiToUserMessage(raw_msg) user_msg = OpenAiToUserMessage(raw_msg)
push!(agent._state.messages, user_msg) push!(messages, user_msg)
agent.agentEventSink("_process_message 6") agentEventSink("_processMessage 6")
end end
agentEventSink("_processMessage 7")
# call prepareContext()
state = agentState(systemPrompt, nothing, tools, messages)
agentEventSink("_processMessage 8")
preparedContext = prepareContext(state)
agentEventSink("_processMessage 8")
# Call formatMessagesForLLM() to format for LLM
formattedMessages = formatMessagesForLLM(preparedContext)
# call agent.prepareContext() agentEventSink("_processMessage 10")
preparedContext = agent.prepareContext(agent._state)
# Call agent.formatMsgForLLM(agent._state) to format for LLM
formatted_messages = agent.formatMsgForLLM(preparedContext)
agent.agentEventSink("_process_message 7")
# Call llmCall() (blocking — the task waits here) # Call llmCall() (blocking — the task waits here)
response = llmCall(formattedMessages)
agentEventSink(response)
agentEventSink("_processMessage 11")
error("debug marker") error("debug marker")
response = agent.llmCall(formatted_messages) # Check if LLM used tool calls (inspect content for tool_call blocks)
agent.agentEventSink("_process_message 8") hasToolCalls = false
toolCallList = agentToolCall[]
#WORKING Check if LLM used tool calls (inspect content for tool_call blocks)
has_tool_calls = false
tool_call_list = agentToolCall[]
for content_block in response.content for content_block in response.content
if content_block isa Dict if content_block isa Dict
if get(content_block, :type, "") == "tool_calls" if get(content_block, :type, "") == "tool_calls"
has_tool_calls = true hasToolCalls = true
for tc_data in get(content_block, :tool_calls, []) for tc_data in get(content_block, :tool_calls, [])
tc = agentToolCall( tc = agentToolCall(
type="function", type="function",
@@ -366,10 +403,10 @@ function _process_message(agent::yiemAgent)::assistantMessage
name=get(tc_data, :function, Dict{String,Any}())[:name], name=get(tc_data, :function, Dict{String,Any}())[:name],
arguments=get(tc_data, :function, Dict{String,Any}())[:arguments], arguments=get(tc_data, :function, Dict{String,Any}())[:arguments],
) )
push!(tool_call_list, tc) push!(toolCallList, tc)
end end
elseif get(content_block, :type, "") == "tool_call" elseif get(content_block, :type, "") == "tool_call"
has_tool_calls = true hasToolCalls = true
tc_data = content_block tc_data = content_block
tc = agentToolCall( tc = agentToolCall(
type="function", type="function",
@@ -377,35 +414,35 @@ function _process_message(agent::yiemAgent)::assistantMessage
name=get(tc_data, :name, ""), name=get(tc_data, :name, ""),
arguments=get(tc_data, :arguments, Dict{String,Any}()), arguments=get(tc_data, :arguments, Dict{String,Any}()),
) )
push!(tool_call_list, tc) push!(toolCallList, tc)
end end
end end
end end
if has_tool_calls && length(tool_call_list) > 0 if hasToolCalls && length(toolCallList) > 0
# Build context and config for executeToolCalls # Build context and config for executeToolCalls
context = agentContext( context = agentContext(
agent._state.systemPrompt, systemPrompt,
agent._state.messages, messages,
agent._state.tools, tools,
) )
config = agentLoopConfig( config = agentLoopConfig(
agent._state.tools, tools,
agent.beforeToolCall, beforeToolCall,
agent.afterToolCall, afterToolCall,
agent.parallelToolExecute ? "parallel" : "sequential", parallelToolExecute ? "parallel" : "sequential",
) )
signal = nothing signal = nothing
emit = agent.agentEventSink emit = agentEventSink
# call executeToolCalls() # call executeToolCalls()
batch = executeToolCalls(context, response, tool_call_list, config, signal, emit) batch = executeToolCalls(context, response, toolCallList, config, signal, emit)
# save toolResults to agent._state.messages # save toolResults to messages
for tool_result in batch.messages for tool_result in batch.messages
push!(agent._state.messages, tool_result) push!(messages, tool_result)
end end
if batch.terminate if batch.terminate
+1 -1
View File
@@ -116,7 +116,7 @@ julia> stopAgent(agent)
function stopAgent(agent::yiemAgent) function stopAgent(agent::yiemAgent)
put!(agent.inputChannel, :shutdown) put!(agent.inputChannel, :shutdown)
try try
fetch(agent._agent_loop) fetch(agent._agentLoop)
catch e catch e
if e isa TaskFailedException if e isa TaskFailedException
rethrow(e) rethrow(e)
+22
View File
@@ -192,6 +192,22 @@ function formatMsgForLLM(ctx::agentContext)::Dict{String, Any}
] ]
), ),
], ],
"tools"=> [
Dict(
"type" => "function",
"function" => Dict(
"name" => "get_weather",
"description" => "Get current weather",
"parameters" => Dict(
"type" => "object",
"properties" => Dict(
"city" => Dict("type" => "string")
),
"required" => ["city"]
)
)
)
],
"temperature" => 0.7 "temperature" => 0.7
) )
""" """
@@ -217,6 +233,12 @@ function formatMsgForLLM(ctx::agentContext)::Dict{String, Any}
end end
end end
#WORKING convert ctx.tools into openai's tools format.
# ctx.tools has the following format
# tools = OrderedDict{String, YiemAgent.type.agentTool}("getTime" => YiemAgent.type.agentTool("getTime", "Time Lookup", "Get current local time for a timezone or city.", Dict{String, Any}("properties" => Dict("city" => Dict("type" => "string", "description" => "City name as fallback"), "timezone" => Dict("type" => "string", "description" => "IANA timezone, e.g. 'America/New_York'")), "required" => Any[], "type" => "object"), YiemAgent.toolRegistry._tool_getTime.executeTool, nothing, YiemAgent.toolRegistry._tool_getTime.validateRequiredArgs, false), "getWeather" => YiemAgent.type.agentTool("getWeather", "Weather Lookup", "Fetch current weather and forecast for a given city.", Dict{String, Any}("properties" => Dict{String, Dict{String}}("units" => Dict{String, Any}("default" => "celsius", "type" => "string", "description" => "Temperature scale", "enum" => ["celsius", "fahrenheit"]), "city" => Dict("type" => "string", "description" => "City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'")), "required" => ["city"], "type" => "object"), YiemAgent.toolRegistry._tool_getWeather.executeTool, nothing, nothing, false), "listTools" => YiemAgent.type.agentTool("listTools", "List Tools", "List all available tools with their names, labels, and descriptions. Use this before creating a new tool to check for name collisions.", Dict{String, Any}("properties" => Dict{String, Any}(), "required" => Any[], "type" => "object"), YiemAgent.toolRegistry.var"#listTool##0#listTool##1"{YiemAgent.toolRegistry.toolStore}(YiemAgent.toolRegistry.toolStore(OrderedDict{String, YiemAgent.type.agentTool}(#= circular reference @-4 =#), "myagent")), nothing, nothing, false))
return Dict("messages" => messages) return Dict("messages" => messages)
end end