module agentCore export yiemAgent, _agent_loop, OpenAiToUserMessage using JSON, DataStructures, Dates, UUIDs, HTTP, Random, PrettyPrinting, Serialization, DataFrames, Base.Threads using GeneralUtils using ..type, ..utils, ..toolRegistry # ---------------------------------------------- 100 --------------------------------------------- # """ docstring """ mutable struct yiemAgent <: agent # High-level agent wrapper _state::agentState # Current state (prompt, model, messages, tools, etc.) # user sends prompt message to agent. if agent is idle, it process user message right away. # if agent is running, it process user message after the current tool call finished. inputChannel::Channel # Buffers messages the user sends while the agent is busy. Processed after all inputChannel # messages are handled and the agent is idle (not using a tool call). followUpChannel::Channel # agent sends response message to user after processing all user messages in inputChannel # and all followUp messages. outputChannel::Channel _agent_loop::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. # returns new Vector{agentMessage} prepareContext::Union{Function, Nothing} # Convert prepareContext()'s new Vector{agentMessage} to LLM message format formatMsgForLLM::Function # A callable struct. Actually invoke the LLM to get a completion response. # The LLM response comes back as an assistantMessage whose content is an array of content blocks. # Each block has a type — "text", "thinking", or "toolCall". # The code filters for type === "toolCall" blocks, then passes them to executeToolCalls(). llmCall # Callback invoked before executing a tool call (ask for user permission/confirmation/abort, etc..) beforeToolCall::Union{Function, Nothing} # Callback invoked after executing a tool call to sanitize tools output so the output is ready # to be converted into toolResults message afterToolCall::Union{Function, Nothing} # prepareNextTurn::Union{Function, Nothing} # Callback to prepare the next conversation turn # prepareNextTurnWithContext::Union{Function, Nothing} # Same but receives context sessionId::Union{String, Nothing} # Optional session identifier maxRetryDelayMs::Union{Int64, Nothing} # Maximum delay between retries (ms) parallelToolExecute::Bool # Default: false agentEventSink::Function # agent emits its status via this function end """ Create a new yiemAgent instance with a background loop task. Spawns a background `@spawn` task that runs the agent loop, listening on `inputChannel` and `followUpChannel` channels concurrently. # Keyword Arguments - `systemPrompt::String`: System prompt for the agent - `model`: LLM model to use - `tools::OrderedDict{String, agentTool}`: Available tools keyed by name (default: empty) - `messages::Vector{agentMessage}`: Initial conversation messages (default: empty) - `formatMsgForLLM::Function`: Convert agent messages to LLM message format (default: `defaultformatMsgForLLM`) - `llmCall::Function`: Function to invoke the LLM (required) - `prepareContext::Union{Function, Nothing}`: Preprocess/transform messages before sending to LLM (default: `nothing`) - `beforeToolCall::Union{Function, Nothing}`: Callback invoked before executing a tool call (default: `nothing`) - `afterToolCall::Union{Function, Nothing}`: Callback invoked after executing a tool call (default: `nothing`) - `prepareNextTurn::Union{Function, Nothing}`: Callback to prepare the next conversation turn (default: `nothing`) - `prepareNextTurnWithContext::Union{Function, Nothing}`: Same but receives context (default: `nothing`) - `sessionId::Union{String, Nothing}`: Optional session identifier (default: `nothing`) - `maxRetryDelayMs::Union{Int64, Nothing}`: Maximum delay between retries in milliseconds (default: `nothing`) - `parallelToolExecute::Bool`: Run tool calls in parallel (default: `false`) - `agentEventSink::Function`: Callback to receive agent events # Returns - A new `yiemAgent` instance with an active background task """ function yiemAgent( toolsFolderPath::String, llmCall, ; systemPrompt::String="You are helpful assistant.", model=nothing, messages::Vector{agentMessage}=agentMessage[], prepareContext::Function=prepareContext, formatMsgForLLM::Function=formatMsgForLLM, beforeToolCall::Function=beforeToolCall, afterToolCall::Function=afterToolCall, # prepareNextTurn::Union{Function, Nothing}=nothing, # prepareNextTurnWithContext::Union{Function, Nothing}=nothing, sessionId::Union{String, Nothing}=nothing, maxRetryDelayMs::Union{Int64, Nothing}=nothing, parallelToolExecute::Bool=false, agentEventSink::Function=agentEventSink, ) # Create channels: input (user -> agent), followUp (async queue), output (agent -> user) inputChannel = Channel(16) followUp = Channel(32) outputChannel = Channel(16) # load tools from toolsFolderPath toolStore1 = toolStore(name="myagent") loadTools(toolStore1, toolsFolderPath) # Create struct with a placeholder task, then spawn and replace it agent = yiemAgent( agentState(systemPrompt, model, getTools(toolStore1), messages), inputChannel, followUp, outputChannel, nothing, # placeholder — replaced below prepareContext, formatMsgForLLM, llmCall, beforeToolCall, afterToolCall, # prepareNextTurn, # prepareNextTurnWithContext, sessionId, maxRetryDelayMs, parallelToolExecute, agentEventSink, ) # Spawn the background loop and attach it agent._agent_loop = @spawn _agent_loop(agent) return agent 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 to `outputChannel`. Exits on `:shutdown` signal. # Arguments - `agent::yiemAgent`: The agent whose loop to run # Returns - `nothing` — the loop runs until `:shutdown` is received or an error occurs # Notes - This function is automatically spawned as a background task when a `yiemAgent` is created. - On any error, logs the error with `@error` and exits the loop. - Message priority: `inputChannel` messages are checked before `followUpChannel` messages. # Examples ```jldoctest julia> # Called automatically by yiemAgent constructor ``` """ function _agent_loop(agent::yiemAgent) try processingTask = 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 if isready(agent.inputChannel) # message will be taken in _process_message() msg = fetch!(agent.inputChannel) else 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 # start _process_message loop if agent._state.activeRun == false # Dispatch message through the processing pipeline processingTask = Threads.@spawn _process_message(agent) agent._state.activeRun = true end # during agent runs, check followUp message after _process_message() is done if typeof(processingTask) == Task && istaskdone(processingTask) == 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(processingTask) == Task && istaskdone(processingTask) == 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(processingTask) put!(agent.outputChannel, result) agent._state.activeRun = false # reset processingTask = nothing # reset end end catch e # On any error, send error response and exit the loop @error "Agent loop failed" error=e end end """ Process a single message through the agent pipeline. This is the core processing function where LLM calls, tool execution, and response generation should be implemented. Currently a placeholder that echoes back the received message. # Arguments - `agent::yiemAgent`: The agent processing the message - `msg`: The message to process (from `inputChannel` or `followUpChannel`) # Returns - An `assistantMessage` instance with the processed response # Notes - Implement the full processing pipeline: 1. Add `msg` to `agent._state.messages` 2. Call `agent.formatMsgForLLM(agent._state)` 1 to format for LLM 3. If `agent.prepareContext` is set, call it on the formatted messages 4. Call the LLM (blocking — the task waits here) 5. If agent has tools, handle tool calls in a loop 6. Build `assistantMessage` and return it # Examples ```jldoctest julia> # Currently returns a placeholder echo response ``` """ function _process_message(agent::yiemAgent)::assistantMessage # loop until llmCall() response didn't use tool calls final_response = nothing while true """ example message in agent.inputChannel Dict( "role" => "user", "content" => [ Dict("type" => "text", "text" => "Do you have something similar to the one in the image?"), Dict( "type" => "image_url", "image_url" => Dict("url" => "data:mime_type;base64,image2_base64_string") ) ] ), """ # Drain inputChannel and convert OpenAI-format messages to userMessage type while isready(agent.inputChannel) raw_msg = take!(agent.inputChannel) if raw_msg === :shutdown # Re-emit shutdown signal for the loop to handle put!(agent.inputChannel, :shutdown) break end user_msg = OpenAiToUserMessage(raw_msg) push!(agent._state.messages, user_msg) end # call agent.prepareContext() preparedContext = agent.prepareContext(agent._state) # Call agent.formatMsgForLLM(agent._state) to format for LLM formatted_messages = agent.formatMsgForLLM(preparedContext) # Call llmCall() (blocking — the task waits here) response = agent.llmCall(formatted_messages) error(5555555) #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 if content_block isa Dict if get(content_block, :type, "") == "tool_calls" has_tool_calls = true for tc_data in get(content_block, :tool_calls, []) tc = agentToolCall( type="function", id=get(tc_data, :id, string(uuid4())), name=get(tc_data, :function, Dict{String,Any}())[:name], arguments=get(tc_data, :function, Dict{String,Any}())[:arguments], ) push!(tool_call_list, tc) end elseif get(content_block, :type, "") == "tool_call" has_tool_calls = true tc_data = content_block tc = agentToolCall( type="function", id=get(tc_data, :id, string(uuid4())), name=get(tc_data, :name, ""), arguments=get(tc_data, :arguments, Dict{String,Any}()), ) push!(tool_call_list, tc) end end end if has_tool_calls && length(tool_call_list) > 0 # Build context and config for executeToolCalls context = agentContext( agent._state.systemPrompt, agent._state.messages, agent._state.tools, ) config = agentLoopConfig( agent._state.tools, agent.beforeToolCall, agent.afterToolCall, agent.parallelToolExecute ? "parallel" : "sequential", ) signal = nothing emit = agent.agentEventSink # call executeToolCalls() batch = executeToolCalls(context, response, tool_call_list, config, signal, emit) # save toolResults to agent._state.messages for tool_result in batch.messages push!(agent._state.messages, tool_result) end if batch.terminate # If batch requested termination, build a final response final_content = [textContent("Tool execution completed.")] for tool_result in batch.messages for content_block in tool_result.content if content_block isa textContent append!(final_content, [content_block]) elseif content_block isa Dict if haskey(content_block, :text) push!(final_content, textContent(content_block[:text])) end end end end final_response = assistantMessage( role="assistant", content=final_content, api=response.api, model=response.model, usage=response.usage, stopReason="tool_use_terminated", errorMessage=if any(x -> x.isError, batch.messages) "One or more tool calls failed" else nothing end, timestamp=now(), ) break end else # LLM did not use tool calls — this is the final response final_response = response break end end return final_response end """ createErrorToolResult(msg::String) -> agentToolResult Builds an `agentToolResult` containing a single text content item with the provided error message and an empty details dictionary. Used when a tool call cannot be executed due to errors. Returning a result instead of throwing ensures that errors at any point in the tool call pipeline are fed back to the LLM as a tool result message. This allows the model to see the error and decide whether to retry, re-issue the call with different arguments, or report failure to the user. # Arguments - `msg::String`: The error message to embed in the result # Returns - `agentToolResult`: A result with `content = [textContent("text", msg)]` # Examples ```julia julia> createErrorToolResult("Tool not found") agentToolResult([textContent("text", "Tool not found")], Dict{Any,Any}()) ``` """ function createErrorToolResult(msg::String)::agentToolResult return agentToolResult([textContent("text", msg)], dict{any,any}()) end """ createToolResultMessage(f::finalizedOutcome) -> toolResultMessage Constructs a `toolResultMessage` from a `finalizedOutcome`. Normalizes missing content to an empty array and includes the `addedToolNames` field only when the tool dynamically registered new tools during execution. This conversion is necessary because the tool result is an `agentToolResult` used by tool implementations, while the agent loop consumes `toolResultMessage` objects that become part of the conversation history. The message format includes metadata like timestamp and tool call ID that the raw result does not carry, and it is the object emitted via `messageStart`/`messageEnd` events so the LLM receives the result as a proper assistant/user message in the context window. # Arguments - `f::finalizedOutcome`: The finalized tool call outcome # Returns - `toolResultMessage`: A message ready for the agent loop context # Examples ```julia julia> outcome = finalizedOutcome(tc, agentToolResult(content, details, usage, false), false); julia> createToolResultMessage(outcome) toolResultMessage("toolResult", "call_1", "search_wine", content, details, usage, [], false, 1234567890) ``` """ function createToolResultMessage(f::finalizedOutcome)::toolResultMessage return toolResultMessage( "toolResult", f.toolCall.id, f.toolCall.name, f.result.content, f.result.details, f.result.usage, get(f.result, :addedToolNames, string[]), f.isError, nowMillis() ) end """ shouldTerminate(finalizedCalls::Vector{finalizedOutcome}) -> Bool Returns `true` only when every finalized call in the batch has `result.terminate == true`. All tools must agree — if any tool did not request termination, the agent continues. This prevents a single tool that happens to set `terminate: true` (e.g. for metadata purposes) from accidentally stopping the agent when other tools in the batch did not intend to terminate. # Arguments - `finalizedCalls`: Vector of finalized tool call outcomes # Returns - `Bool`: `true` if all calls requested termination # Examples ```julia julia> shouldTerminate(finalizedOutcome[]) false julia> shouldTerminate([finalizedOutcome(tc, agentToolResult([], dict{any,any}(), dict{any,any}(), false), false) for _ in 1:2]) false julia> shouldTerminate([finalizedOutcome(tc, agentToolResult([], dict{any,any}(), dict{any,any}(), true), false) for _ in 1:2]) true ``` """ function shouldTerminate(batches::Vector{finalizedOutcome})::Bool return !isempty(batches) && all(b -> b.result.terminate, batches) end """ prepareToolCallArguments(tool::agentTool, toolCall::agentToolCall) -> agentToolCall Calls the tool's optional `prepareArguments` hook to transform the raw argument values from the LLM before schema validation. If the tool has no hook or the hook returns the same object reference, the original call is returned unchanged. This hook allows tools to normalize arguments that the LLM may have produced in a non-standard format — for example, converting a date string to a timestamp, expanding a short file path to an absolute path, or normalizing casing. It runs before schema validation so the validator sees the normalized form rather than raw LLM output. # Arguments - `tool::agentTool`: The tool definition (may have a `prepareArguments` hook) - `toolCall::agentToolCall`: The raw tool call from the assistant # Returns - `agentToolCall`: The tool call with potentially transformed arguments # Examples ```julia # No prepareArguments hook — returns input unchanged prepareToolCallArguments(noHookTool, tc) # => tc # same reference # With hook that normalizes arguments prepareToolCallArguments(normalizeTool, tc) # => agentToolCall{..., arguments=Dict("date" => 1700000000)} # "2024-01-15" → timestamp ``` """ function prepareToolCallArguments(tool::agentTool, toolCall::agentToolCall)::agentToolCall if tool.prepareArguments === nothing return toolCall end prepared = tool.prepareArguments(toolCall.arguments) if prepared == toolCall.arguments return toolCall end return merge(toolCall, dict(:arguments => prepared)) end """ prepareToolCall(context, assistantMsg, toolCall, config, signal) -> Union{preparedToolCall,immediateOutcome} Resolves the tool by name, prepares and validates its arguments, and runs the `beforeToolCall` hook. Returns a `preparedToolCall` if successful or an `immediateOutcome` if the tool is not found, validation fails, the hook blocks execution, or the signal is aborted. Errors during preparation are caught and returned as immediate error outcomes so the agent loop can feed them back to the model. The key design decision here is that preparation never throws. Every failure path returns an `immediateOutcome` with an error result. This ensures the agent loop always receives a valid tool result message for every tool call the assistant requested, regardless of whether preparation succeeded. The LLM can then use the error message to decide whether to retry with different arguments or acknowledge the failure. # Arguments - `context::agentContext`: Current agent context with tools and messages - `assistantMsg::assistantMessage`: The assistant message containing the tool call - `toolCall::agentToolCall`: The tool call to prepare - `config::agentLoopConfig`: Loop configuration (may include `beforeToolCall`) - `signal::Union{Nothing,AbortSignal}`: Optional abort signal # Returns - `preparedToolCall`: If preparation succeeded (tool found, arguments valid, not blocked) - `immediateOutcome`: If preparation failed (tool missing, invalid args, blocked, aborted) # Notes - Tool lookup is by name via `context.tools` - Validation uses `validateToolArguments` which coerces types per the tool schema - The `beforeToolCall` hook can block execution by returning `{ block: true }` # Examples ```julia # Success path prepareToolCall(context, msg, tc, config, signal) # => preparedToolCall(tool, tc, validatedArgs) # Tool not found prepareToolCall(context, msg, tcNoMatch, config, signal) # => immediateOutcome(createErrorToolResult("Tool fake_tool not found"), true) # Validation failure prepareToolCall(context, msg, tcBadArgs, config, signal) # => immediateOutcome(createErrorToolResult("Validation failed..."), true) # Aborted during preparation prepareToolCall(context, msg, tc, config, abortedSignal) # => immediateOutcome(createErrorToolResult("Operation aborted"), true) ``` """ function prepareToolCall( context::agentContext, assistantMsg::assistantMessage, toolCall::agentToolCall, config::agentLoopConfig, signal::Union{Nothing, abortSignal}, )::Union{preparedToolCall,immediateOutcome} tool = get(context.tools, toolCall.name, nothing) if tool === nothing return immediateOutcome(createErrorToolResult("Tool $toolCall.name not found"), true) end try # 1. prepare arguments (tool-specific transform) prepared = prepareToolCallArguments(tool, toolCall) validatedArgs = validateToolArguments(tool, prepared) # 2. beforeToolCall hook — can block if config.beforeToolCall !== nothing before = config.beforeToolCall( beforeToolCallContext(assistantMsg, toolCall, validatedArgs, context), signal ) if signal !== nothing && signal.aborted return immediateOutcome(createErrorToolResult("Operation aborted"), true) end if before !== nothing && before.block return immediateOutcome( createErrorToolResult(get(before, :reason, "Tool execution was blocked")), true) end end return preparedToolCall(tool, toolCall, validatedArgs) catch err return immediateOutcome(createErrorToolResult(sprint(showerror, err)), true) end end # ── per-call execution ────────────────────────────────────────── """ executePreparedToolCall(prep, signal, emit) -> executedOutcome Executes the tool by calling `tool.execute()` with the validated arguments, the abort signal, and a callback for streaming partial results. Emits `toolExecutionUpdate` events for each partial result batch. Waits for all pending update events to settle before returning. Catches execution errors and returns them as an error outcome. The `accepting` guard prevents emitting updates after the call has finished. Long-running tools (e.g. file uploads, model training, web scraping) may take seconds or minutes. The streaming update mechanism allows UI listeners and other consumers to show progress in real time rather than waiting for the entire call to complete. The `accepting` guard ensures that if the tool's execute function yields after emitting updates but before returning, no duplicate or stale updates are emitted after the result has already been captured. # Arguments - `prep::preparedToolCall`: The prepared tool call (resolved tool + validated args) - `signal::Union{Nothing,AbortSignal}`: Optional abort signal - `emit::Function`: Event emitter for lifecycle events # Returns - `executedOutcome`: The execution result and whether it was an error # Examples ```julia # Successful execution executePreparedToolCall(prep, nothing, emit) # => executedOutcome(agentToolResult([textContent("text", "Found 3 wines")], dict{any,any}(), dict{any,any}()), false) # Execution error executePreparedToolCall(prep, nothing, emit) # => executedOutcome(createErrorToolResult("Connection timeout"), true) ``` """ function executePreparedToolCall( prep::preparedToolCall, signal::Union{Nothing,abortSignal}, emit::Function, )::executedOutcome updateEvents = promise[] accepting = true try result = prep.tool.execute( prep.toolCall.id, prep.args, signal, partialResult -> begin if accepting push!(updateEvents, emit(toolExecUpdateEvent(prep.toolCall.id, prep.toolCall.name, prep.toolCall.arguments, partialResult))) end end ) accepting = false wait.(updateEvents) return executedOutcome(result, false) catch err accepting = false wait.(updateEvents) return executedOutcome(createErrorToolResult(sprint(showerror, err)), true) end end # ── per-call finalization ─────────────────────────────────────── """ finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal) -> finalizedOutcome Runs the `afterToolCall` hook on the executed result, allowing the consumer to mutate the result content, details, usage, termination flag, or error status. Catches errors from the hook and converts them to error outcomes. Returns a `finalizedOutcome` that is used to construct the tool result message. The `afterToolCall` hook exists as a post-processing step that runs after every tool call regardless of success or failure. Common use cases include: - Masking sensitive data from result content before the LLM sees it (e.g. removing API keys from error messages). - Normalizing usage tracking data into a consistent format. - Inspecting the result and deciding to flip `terminate: true` based on business logic (e.g. "if deployment failed, stop the agent rather than retrying"). - Wrapping an error result in a friendlier message for the LLM to understand. If the hook itself throws, the error is caught and the result becomes an error outcome. This ensures the tool pipeline never breaks due to a buggy hook. # Arguments - `context::agentContext`: Current agent context - `assistantMsg::assistantMessage`: The assistant message that made the tool call - `prep::preparedToolCall`: The originally prepared tool call - `executed::executedOutcome`: The raw execution result - `config::agentLoopConfig`: Loop configuration (may include `afterToolCall`) - `signal::Union{Nothing,AbortSignal}`: Optional abort signal # Returns - `finalizedOutcome`: The finalized outcome ready for message construction # Examples ```julia # No afterToolCall hook — returns executed result unchanged finalizeExecutedToolCall(context, msg, prep, execOk, config, nothing) # => finalizedOutcome(tc, execOk.result, false) # afterToolCall masks sensitive data finalizeExecutedToolCall(context, msg, prep, execOk, configWithHook, nothing) # => finalizedOutcome(tc, maskedResult, false) # afterToolCall flips terminate based on business logic finalizeExecutedToolCall(context, msg, prep, execFail, config, nothing) # => finalizedOutcome(tc, {terminate: true}, true) ``` """ function finalizeExecutedToolCall( context::agentContext, assistantMsg::assistantMessage, prep::preparedToolCall, executed::executedOutcome, config::agentLoopConfig, signal::Union{Nothing,abortSignal}, )::finalizedOutcome result = executed.result isError = executed.isError if config.afterToolCall !== nothing try after = config.afterToolCall( afterToolCallContext(assistantMsg, prep.toolCall, prep.args, result, isError, context), signal ) if after !== nothing result = merge(result, dict(:content=>get(after,:content,result.content), :details=>get(after,:details,result.details), :usage=>get(after,:usage,result.usage), :terminate=>get(after,:terminate,result.terminate))) isError = get(after, :isError, isError) end catch err result = createErrorToolResult(sprint(showerror, err)) isError = true end end return finalizedOutcome(prep.toolCall, result, isError) end # ── sequential execution ──────────────────────────────────────── """ executeToolCallsSequential(context, assistantMsg, toolCalls, config, signal, emit) -> agentToolCallBatch Executes tool calls one at a time in the order they appear. For each call: emits `toolExecutionStart`, runs `prepareToolCall`, then either resolves the immediate outcome or executes/finalizes the prepared call. Emits `toolExecutionEnd` and creates the tool result message before proceeding to the next call. Respects the abort signal — if aborted, remaining calls are skipped. Returns a batch with `terminate` determined by whether all results set the termination flag. Sequential execution is required when tool calls have implicit dependencies — for example, a `create_database` tool must complete before `create_table` can reference it. It is also the safer default because it prevents race conditions when multiple tools share state (e.g. writing to the same file or API rate limits). Use parallel only when you are confident the tools are independent. # Arguments - `context::agentContext`: Current agent context - `assistantMsg::assistantMessage`: The assistant message containing tool calls - `toolCalls::Vector{agentToolCall}`: Tool calls to execute (ordered) - `config::agentLoopConfig`: Loop configuration - `signal::Union{Nothing,AbortSignal}`: Optional abort signal - `emit::Function`: Event emitter # Returns - `agentToolCallBatch`: Result messages and termination flag # Notes - Calls execute strictly in order; each completes fully before the next begins - Aborting during one call skips all remaining calls - If any call returns `terminate: true`, it is included in the batch but does not force termination unless all calls do # Examples ```julia # Two independent reads — both succeed executeToolCallsSequential(ctx, msg, [readTc, readTc2], config, nothing, emit) # => agentToolCallBatch([result1, result2], false) # One tool fails, next is skipped due to abort executeToolCallsSequential(ctx, msg, [tc1, tc2, tc3], config, abortedSignal, emit) # => agentToolCallBatch([result1], false) # tc2 failed, tc3 skipped # All tools request termination executeToolCallsSequential(ctx, msg, [deployTc], config, nothing, emit) # => agentToolCallBatch([deployResult], true) ``` """ function executeToolCallsSequential( context::agentContext, assistantMsg::assistantMessage, toolCalls::Vector{agentToolCall}, config::agentLoopConfig, signal::Union{Nothing,abortSignal}, emit::Function, )::agentToolCallBatch finalizedCalls = finalizedOutcome[] messages = toolResultMessage[] for tc in toolCalls emit(toolExecStartEvent(tc.id, tc.name, tc.arguments)) prep = prepareToolCall(context, assistantMsg, tc, config, signal) if prep isa immediateOutcome finalized = finalizedOutcome(tc, prep.result, prep.isError) else executed = executePreparedToolCall(prep, signal, emit) finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal) end emit(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name, finalized.result, finalized.isError)) push!(messages, createToolResultMessage(finalized)) push!(finalizedCalls, finalized) if signal !== nothing && signal.aborted break end end return agentToolCallBatch(messages, shouldTerminate(finalizedCalls)) end # ── parallel execution ────────────────────────────────────────── """ executeToolCallsParallel(context, assistantMsg, toolCalls, config, signal, emit) -> agentToolCallBatch Prepares all tool calls concurrently and spawns a task for each prepared call. Immediate outcomes are resolved instantly. Task entries are collected in order, then `fetch`ed to await all concurrent executions. Tool result messages are created from finalized outcomes in order and returned as a batch. Respects the abort signal — if aborted during preparation, remaining calls are skipped. Finalization order preserves the original call order. Parallel execution is appropriate when the assistant requests independent tools — for example, reading multiple files, querying separate databases, or making independent API calls. It reduces wall-clock time compared to sequential execution. The tradeoff is that parallel calls can overwhelm external resources (rate limits, connection pools, disk I/O). Finalization preserves the original call order so tool result messages appear in the same order the assistant requested them, regardless of which call finishes first. # Arguments - `context::agentContext`: Current agent context - `assistantMsg::assistantMessage`: The assistant message containing tool calls - `toolCalls::Vector{agentToolCall}`: Tool calls to execute (order preserved in output) - `config::agentLoopConfig`: Loop configuration - `signal::Union{Nothing,AbortSignal}`: Optional abort signal - `emit::Function`: Event emitter # Returns - `agentToolCallBatch`: Result messages (in original call order) and termination flag # Notes - All tool calls are prepared before any execution begins - Execution tasks run concurrently; `fetch` waits for completion in order - Immediate outcomes (errors/blocks) resolve instantly without spawning tasks - Aborting during preparation skips remaining preparations but does not cancel tasks already running # Examples ```julia # Three independent reads — all succeed, results ordered by original call order executeToolCallsParallel(ctx, msg, [readA, readB, readC], config, nothing, emit) # => agentToolCallBatch([resultA, resultB, resultC], false) # Mix of immediate error and concurrent success executeToolCallsParallel(ctx, msg, [badTc, goodTc], config, nothing, emit) # => agentToolCallBatch([errorResult, goodResult], false) # Abort during preparation executeToolCallsParallel(ctx, msg, [tc1, tc2, tc3], config, abortedSignal, emit) # => agentToolCallBatch([...], false) # only prepared calls complete ``` """ function executeToolCallsParallel( context::agentContext, assistantMsg::assistantMessage, toolCalls::Vector{agentToolCall}, config::agentLoopConfig, signal::Union{Nothing,abortSignal}, emit::Function, )::agentToolCallBatch entries = union{finalizedOutcome,task{finalizedOutcome}}[] for tc in toolCalls emit(toolExecStartEvent(tc.id, tc.name, tc.arguments)) prep = prepareToolCall(context, assistantMsg, tc, config, signal) if prep isa immediateOutcome finalized = finalizedOutcome(tc, prep.result, prep.isError) emit(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name, finalized.result, finalized.isError)) push!(entries, finalized) else task = task() do executed = executePreparedToolCall(prep, signal, emit) finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal) emit(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name, finalized.result, finalized.isError)) return finalized end schedule(task) push!(entries, task) end if signal !== nothing && signal.aborted break end end finalizedCalls = finalizedOutcome[] for entry in entries outcome = entry isa task ? fetch(entry) : entry push!(finalizedCalls, outcome) end messages = toolResultMessage[] for f in finalizedCalls push!(messages, createToolResultMessage(f)) end return agentToolCallBatch(messages, shouldTerminate(finalizedCalls)) end """ executeToolCalls(context, assistantMsg, toolCalls, config, signal, emit) -> agentToolCallBatch Dispatches to sequential or parallel execution. Uses sequential mode when `config.toolExecution == "sequential"` or when any of the tool calls reference a tool with `executionMode: "sequential"`. Otherwise uses parallel execution. This is the entry point called from `streamAssistantResponse` in the agent loop. The sequential mode takes priority over parallel because it is the safe default. If even one tool in a batch is marked sequential, all tools execute sequentially — this prevents a single dependent tool from racing with an otherwise independent one. The per-tool `executionMode` allows fine-grained control (e.g. most tools are parallel but a specific write tool is sequential), while the config-level `toolExecution` provides a global override. # Arguments - `context::agentContext`: Current agent context (used for per-tool `executionMode` lookup) - `assistantMsg::assistantMessage`: The assistant message containing tool calls - `toolCalls::Vector{agentToolCall}`: Tool calls to execute - `config::agentLoopConfig`: Loop configuration (`toolExecution` mode) - `signal::Union{Nothing,AbortSignal}`: Optional abort signal - `emit::Function`: Event emitter # Returns - `agentToolCallBatch`: The result batch from the selected execution strategy # Notes - Per-tool `executionMode` is checked against `context.tools` for each tool call - If any tool is sequential, the entire batch runs sequentially - `config.toolExecution` can override all per-tool settings globally # Examples ```julia # Parallel dispatch — no sequential tools in batch executeToolCalls(ctx, msg, [searchTc, fetchTc], configParallel, nothing, emit) # => agentToolCallBatch(results, false) # parallel execution # Sequential fallback — one tool is marked sequential executeToolCalls(ctx, msg, [searchTc, writeTc], configParallel, nothing, emit) # => agentToolCallBatch(results, false) # sequential because writeTc is sequential # Global override — config forces sequential regardless of per-tool settings executeToolCalls(ctx, msg, [searchTc, fetchTc], configSequential, nothing, emit) # => agentToolCallBatch(results, false) # sequential because config says so ``` """ function executeToolCalls( context::agentContext, assistantMsg::assistantMessage, toolCalls::Vector{agentToolCall}, config::agentLoopConfig, signal::Union{Nothing,abortSignal}, emit::Function, )::agentToolCallBatch hasSequential = false for tc in toolCalls t = get(context.tools, tc.name, nothing) if t !== nothing && !t.parallelToolExecute hasSequential = true break end end if config.toolExecution == "sequential" || hasSequential return executeToolCallsSequential(context, assistantMsg, toolCalls, config, signal, emit) else return executeToolCallsParallel(context, assistantMsg, toolCalls, config, signal, emit) end end """ OpenAiToUserMessage(msg::Dict) -> userMessage Converts an OpenAI-format message dictionary into a `userMessage` type. Parses `content` blocks: `text` blocks become `textContent`, `image_url` blocks have their data URI (`data:;base64,`) parsed via regex to extract the base64 data and MIME type as separate `imageContent` fields. The OpenAI-format dictionary: ``` Dict( "role" => "user", "content" => [ Dict("type" => "text", "text" => "..."), Dict("type" => "image_url", "image_url" => Dict("url" => "data:image/png;base64,...")) ] ) ``` # Arguments - `msg`: A dictionary with `"role"` and `"content"` keys in OpenAI format # Returns - `userMessage`: Instance with `content` as `Vector{messageContent}` # Examples ```julia msg = Dict("role" => "user", "content" => [Dict("type" => "text", "text" => "Hello")]) OpenAiToUserMessage(msg) # => userMessage("user", [textContent("Hello")], DateTime(...)) ``` """ function OpenAiToUserMessage(msg::Dict)::userMessage content_blocks = Vector{messageContent}() raw_content = get(msg, "content", Any[]) if raw_content isa Vector for block in raw_content if block isa Dict block_type = get(block, "type", "") if block_type == "text" text = get(block, "text", "") push!(content_blocks, textContent(text)) elseif block_type == "image_url" image_url = get(block, "image_url", Dict()) url = get(image_url, "url", "") m = match(r"^data:([a-z0-9/_-]+);base64,(.+)$", url) if m !== nothing push!(content_blocks, imageContent(m.captures[2], m.captures[1])) else push!(content_blocks, imageContent(url, "image/png")) end end end end end return userMessage(content=content_blocks) end end # end of module