diff --git a/README_tools.md b/README_tools.md index baf5b2d..de61209 100644 --- a/README_tools.md +++ b/README_tools.md @@ -152,7 +152,7 @@ result = getTime_tool.execute("call-1", Dict("city" => "Tokyo"), sig, op) **Via agent loop (production):** ``` 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 → LLM returns tool_calls → 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` -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 @@ -404,7 +404,7 @@ The loop tracks 6 states (documented at `agentCore.jl:39-75`): ### Loop Logic (simplified) ```julia -function _agent_loop(agent::yiemAgent) +function _agentLoop(agent::yiemAgent) while true # 1. Wait for message from inputChannel (blocking poll) msg = fetch!(agent.inputChannel) # agentCore.jl:84 @@ -1332,7 +1332,7 @@ USER SENDS MESSAGE 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) ── _process_message ────────────────────────────────────────────── diff --git a/src/agentCore.jl b/src/agentCore.jl index 1526ce2..e438904 100644 --- a/src/agentCore.jl +++ b/src/agentCore.jl @@ -1,6 +1,6 @@ module agentCore -export yiemAgent, _agent_loop, OpenAiToUserMessage +export yiemAgent, _agentLoop, OpenAiToUserMessage using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization, DataFrames, Base.Threads @@ -27,10 +27,10 @@ mutable struct yiemAgent <: agent # High-level agent wrapper # and all followUp messages. 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, - # 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} prepareContext::Union{Function, Nothing} @@ -131,7 +131,7 @@ function yiemAgent( ) # Spawn the background loop and attach it - agent._agent_loop = @spawn _agent_loop(agent) + agent._agentLoop = @spawn _agentLoop(agent) return agent end @@ -141,7 +141,7 @@ end Private agent loop. Runs in a background `@spawn` task. 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. # Arguments @@ -160,7 +160,8 @@ to `outputChannel`. Exits on `:shutdown` signal. julia> # Called automatically by yiemAgent constructor ``` """ -function _agent_loop(agent::yiemAgent) +function _agentLoop(agent::yiemAgent) + processMessageInputCh = Channel(32) try processingTask = nothing @@ -208,8 +209,8 @@ function _agent_loop(agent::yiemAgent) while msg === nothing if isready(agent.inputChannel) - # message will be taken in _process_message() - msg = fetch(agent.inputChannel) + # message will be taken in _processMessage() + msg = take!(agent.inputChannel) agent.agentEventSink("new user msg") else yield() @@ -232,29 +233,50 @@ function _agent_loop(agent::yiemAgent) #TODO make sure every running tools ended properly break + else + agent.agentEventSink("_agentLoop push 1") + put!(processMessageInputCh, msg) #WORKING + agent.agentEventSink("_agentLoop push 2") end - # start _process_message loop + # start _processMessage loop if agent._state.activeRun == false - agent.agentEventSink("_agent_loop 2") + agent.agentEventSink("_agentLoop 2") # 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.agentEventSink("_agent_loop 3") + agent.agentEventSink("_agentLoop 3") 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 + 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 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 @@ -268,6 +290,7 @@ function _agent_loop(agent::yiemAgent) agent._state.activeRun = false # reset processingTask = nothing # reset end + agent.agentEventSink("_agentLoop 6") end catch e # 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 ``` """ -function _process_message(agent::yiemAgent)::assistantMessage - agent.agentEventSink("_process_message 1") +function _processMessage( + 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 final_response = nothing while true - """ example message in agent.inputChannel + """ example message in inputChannel Dict( "role" => "user", "content" => [ @@ -323,42 +358,44 @@ function _process_message(agent::yiemAgent)::assistantMessage """ # Drain inputChannel and convert OpenAI-format messages to userMessage type - while isready(agent.inputChannel) - agent.agentEventSink("_process_message 2") - raw_msg = take!(agent.inputChannel) - agent.agentEventSink("_process_message 3") + while isready(inputChannel) + agentEventSink("_processMessage 2") + raw_msg = take!(inputChannel) + agentEventSink("_processMessage 3") if raw_msg === :shutdown - agent.agentEventSink("_process_message 4") + agentEventSink("_processMessage 4") # Re-emit shutdown signal for the loop to handle - put!(agent.inputChannel, :shutdown) + put!(inputChannel, :shutdown) break end - agent.agentEventSink("_process_message 5") + agentEventSink("_processMessage 5") user_msg = OpenAiToUserMessage(raw_msg) - push!(agent._state.messages, user_msg) - agent.agentEventSink("_process_message 6") + push!(messages, user_msg) + agentEventSink("_processMessage 6") 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() - preparedContext = agent.prepareContext(agent._state) - - # Call agent.formatMsgForLLM(agent._state) to format for LLM - formatted_messages = agent.formatMsgForLLM(preparedContext) - - agent.agentEventSink("_process_message 7") + agentEventSink("_processMessage 10") # Call llmCall() (blocking — the task waits here) + response = llmCall(formattedMessages) + agentEventSink(response) + agentEventSink("_processMessage 11") error("debug marker") - response = agent.llmCall(formatted_messages) - agent.agentEventSink("_process_message 8") - - #WORKING Check if LLM used tool calls (inspect content for tool_call blocks) - has_tool_calls = false - tool_call_list = agentToolCall[] + # Check if LLM used tool calls (inspect content for tool_call blocks) + hasToolCalls = false + toolCallList = agentToolCall[] for content_block in response.content if content_block isa Dict if get(content_block, :type, "") == "tool_calls" - has_tool_calls = true + hasToolCalls = true for tc_data in get(content_block, :tool_calls, []) tc = agentToolCall( type="function", @@ -366,10 +403,10 @@ function _process_message(agent::yiemAgent)::assistantMessage name=get(tc_data, :function, Dict{String,Any}())[:name], arguments=get(tc_data, :function, Dict{String,Any}())[:arguments], ) - push!(tool_call_list, tc) + push!(toolCallList, tc) end elseif get(content_block, :type, "") == "tool_call" - has_tool_calls = true + hasToolCalls = true tc_data = content_block tc = agentToolCall( type="function", @@ -377,35 +414,35 @@ function _process_message(agent::yiemAgent)::assistantMessage name=get(tc_data, :name, ""), arguments=get(tc_data, :arguments, Dict{String,Any}()), ) - push!(tool_call_list, tc) + push!(toolCallList, tc) end end end - if has_tool_calls && length(tool_call_list) > 0 + if hasToolCalls && length(toolCallList) > 0 # Build context and config for executeToolCalls context = agentContext( - agent._state.systemPrompt, - agent._state.messages, - agent._state.tools, + systemPrompt, + messages, + tools, ) config = agentLoopConfig( - agent._state.tools, - agent.beforeToolCall, - agent.afterToolCall, - agent.parallelToolExecute ? "parallel" : "sequential", + tools, + beforeToolCall, + afterToolCall, + parallelToolExecute ? "parallel" : "sequential", ) signal = nothing - emit = agent.agentEventSink + emit = agentEventSink # 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 - push!(agent._state.messages, tool_result) + push!(messages, tool_result) end if batch.terminate diff --git a/src/api.jl b/src/api.jl index 524d09f..50d6dbd 100644 --- a/src/api.jl +++ b/src/api.jl @@ -116,7 +116,7 @@ julia> stopAgent(agent) function stopAgent(agent::yiemAgent) put!(agent.inputChannel, :shutdown) try - fetch(agent._agent_loop) + fetch(agent._agentLoop) catch e if e isa TaskFailedException rethrow(e) diff --git a/src/utils.jl b/src/utils.jl index 37848c8..67c4a1b 100644 --- a/src/utils.jl +++ b/src/utils.jl @@ -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 ) """ @@ -217,6 +233,12 @@ function formatMsgForLLM(ctx::agentContext)::Dict{String, Any} 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) end